如何允许Web工作人员在仍然执行计算的情况下接收新数据?

即使Worker在其他线程上工作,而不是在主页上工作,因此可以在不阻塞UI的情况下连续运行,但它仍然在单个线程上运行。

即使您从该工作人员内部使用了其他工作人员,问题也将是相同的。

唯一的解决办法是使用一种 generator function

但是这样做会大大降低排序算法的速度。

为了使它更好,您可以尝试通过MessageChannel对象钩住每个事件循环:您在一个端口中通话,并在下一个事件循环中接收消息。如果您再次与另一个端口通信,那么每个事件循环都有自己的钩子。

现在,最好是在每个事件循环中运行一个好的批处理,但对于演示,我只调用生成器函数的一个实例(我从 this Q/A )

const worker = new Worker(getWorkerURL());
worker.onmessage = draw;

onclick = e =>     worker.postMessage(0x0000FF/0xFFFFFF); // add a red pixel

// every frame we request the current state from Worker
function requestFrame() {
  worker.postMessage('gimme a frame');
  requestAnimationFrame(requestFrame);
}
requestFrame();

// drawing part
const ctx = canvas.getContext('2d');
const img = ctx.createImageData(50, 50);
const data = new Uint32Array(img.data.buffer);
ctx.imageSmoothingEnabled = false;

function draw(evt) {
  // converts 0&1 to black and white pixels
  const list = evt.data;
  list.forEach((bool, i) =>
    data[i] = (bool * 0xFFFFFF) + 0xFF000000
  );
  ctx.setTransform(1,0,0,1,0,0);
  ctx.clearRect(0,0,canvas.width,canvas.height);
  ctx.putImageData(img,0,0);
  // draw bigger
  ctx.scale(5,5);
  ctx.drawImage(canvas, 0,0);
}

function getWorkerURL() {
  const script = document.querySelector('[type="worker-script"]');
  const blob = new Blob([script.textContent]);
  return URL.createObjectURL(blob);
}

body{
  background: ivory;
}

<script type="worker-script">
// our list
const list = Array.from({length: 2500}).map(_=>+(Math.random()>.5));
// our sorter generator
let sorter = bubbleSort(list);
let done = false;
/* inner messaging channel */
const msg_channel = new MessageChannel();
// Hook to every Event loop
msg_channel.port2.onmessage = e => {
  // procede next step in sorting algo
  // could be a few thousands in a loop
  const state = sorter.next();
  // while running
  if(!state.done) {
    msg_channel.port1.postMessage('');
    done = false;
  }
  else {
    done = true;
  }
}
msg_channel.port1.postMessage("");

/* outer messaging channel (from main) */
self.onmessage = e => {
  if(e.data === "gimme a frame") {
    self.postMessage(list);
  }
  else {
    list.push(e.data);
    if(done) { // restart the sorter
      sorter = bubbleSort(list);
      msg_channel.port1.postMessage('');
    }
  }
};

function* bubbleSort(a) { // * is magic
  var swapped;
  do {
    swapped = false;
    for (var i = 0; i < a.length - 1; i++) {
      if (a[i] > a[i + 1]) {
        var temp = a[i];
        a[i] = a[i + 1];
        a[i + 1] = temp;
        swapped = true;
        yield swapped; // pause here
      }
    }
  } while (swapped);
}
</script>
<pre> click to add red pixels</pre>
<canvas id="canvas" width="250" height="250"></canvas>

worker.onmessage=draw;

onclick=e=>worker.postMessage(0x0000FF/0xFFFFFF);//添加一个红色像素

函数requestFrame(){
worker.postMessage(“给我一个框架”);
}
requestFrame();

//绘图部件
常量数据=新的UINT32阵列(img.data.buffer);
ctx.imageSmoothingEnabled=假;

函数绘制(evt){
常数列表=evt.data;
list.forEach((bool,i)=>
数据[i]=(bool*0xFFFFFF)+0xFF000000
setTransform(1,0,0,1,0,0);
clearRect(0,0,canvas.width,canvas.height);
ctx.putImageData(img,0,0);
//画大一点
ctx.刻度(5,5);
ctx.drawImage(画布,0,0);

const script=document.querySelector('[type=“worker script”]');
const blob=新blob([script.textContent]);
返回URL.createObjectURL(blob);

身体{
背景:象牙;
}

<script type="worker-script">
// our list
const list = Array.from({length: 2500}).map(_=>+(Math.random()>.5));
// our sorter generator
let done = false;


/* outer messaging channel (from main) */
self.onmessage = e => {
  if(e.data === "gimme a frame") {
    self.postMessage(list);
  }
  else {
    list.push(e.data);
    if(done) { // restart the sorter
      bubbleSort(list);
    }
  }
};

async function bubbleSort(a) { // async is magic
  var swapped;
  do {
    swapped = false;
    for (var i = 0; i < a.length - 1; i++) {
      if (a[i] > a[i + 1]) {
        const temp = a[i];
        a[i] = a[i + 1];
        a[i + 1] = temp;
        swapped = true;
      }
      if( i % 50 === 0 ) { // by batches of 50?
        await waitNextTask(); // pause here
      }
    }
  } while (swapped);
  done = true;
}

function waitNextTask() {
  return new Promise( (resolve) => {
    const channel = waitNextTask.channel ||= new MessageChannel();
    channel.port1.addEventListener("message", (evt) => resolve(), { once: true });
    channel.port2.postMessage("");
    channel.port1.start();
  });
}

bubbleSort(list);
</script>
<pre> click to add red pixels</pre>
<canvas id="canvas" width="250" height="250"></canvas>

setTimeout 功能中断。例如,如果没有一个附加线程,就不可能并行执行2个函数,而是使用 设置超时 函数中断技巧我们可以按如下方式执行:

函数的并行执行示例

var count_0 = 0,
    count_1 = 0;

function func_0()
{
    if(count_0 < 3)
        setTimeout(func_0, 0);//the same: setTimeout(func_0);

    console.log('count_0 = '+count_0);
    count_0++
}

function func_1()
{
    if(count_1 < 3)
        setTimeout(func_1, 0);

    console.log('count_1 = '+count_1)
    count_1++
}

func_0();
func_1();

您将获得以下输出:

计数0=0
计数1=0
计数0=1
计数1=1
计数0=2


计数1=3

为什么可能?因为 设置超时 函数需要一些时间来执行。这一次甚至足够执行下面代码中的某些部分。

你的解决方案

对于这种情况,您必须编写自己的数组排序函数(或者您也可以使用我提供的以下函数),因为我们不能中断本机排序 sort 作用在这个函数中,你必须使用这个 设置超时 函数中断技巧。你可以收到你的 message 事件通知。

自定义排序函数中断的示例

var numbers = [4, 2, 1, 3, 5];

// this is my bubble sort function with interruption
/**
 * Sorting an array. You will get the same, but sorted array.
 * @param {array[]} arr – array to sort
 * @param {number} dir – if dir = -1 you will get an array like [5,4,3,2,1]
 *                 and if dir = 1 in opposite direction like [1,2,3,4,5]
 * @param {number} passCount – it is used only for setTimeout interrupting trick.
 */
function sortNumbersWithInterruption(arr, dir, passCount)
{
    var passes = passCount || arr.length,
        halfOfArrayLength = (arr.length / 2) | 0; // for ex. 2.5 | 0 = 2

    // Why we need while loop: some values are on
    // the end of array and we have to change their
    // positions until they move to the first place of array.
    while(passes--)
    {
        if(!passCount && passes == halfOfArrayLength)
        {
            // if you want you can also not write the following line for full break of sorting
            setTimeout(function(){sortNumbersWithInterruption(arr, dir, passes)}, 0);
            /*
                You can do here all what you want. Place 1
            */
            break
        }

        for(var i = 0; i < arr.length - 1; i++)
        {
            var a = arr[i],
                b = arr[i+1];

            if((a - b) * dir > 0)
            {
                arr[i] = b;
                arr[i+1] = a;
            }
        }

        console.log('array is: ' + arr.join());
    }

    if(passCount)
        console.log('END sring is: ' + arr.join());
}

sortNumbersWithInterruption(numbers, -1); //without passCount parameter
/*
    You can do here all what you want. Place 2
*/
console.log('The execution is here now!');

数组为:4,2,3,5,1
数组为:4,3,5,2,1
死刑现在就在这里!
数组为:4,5,3,2,1
数组为:5,4,3,2,1
末端行程为:5,4,3,2,1

以下是想法:

2. 工作进程只接收元素,而不是整个阵列

3. 您的工作人员将任何接收到的元素正确地插入数组中的正确位置

4. 如果当前数组在最后一个事件后发生更改,则工作进程每隔n秒会引发一条带有当前数组的消息。(如果愿意,可以在每次插入时发送数组,但以某种方式缓冲会更有效)

最终,您将获得整个数组,如果添加了任何项,您将收到要更新的数组。

注意:由于数组总是排序的,所以可以使用二进制搜索将其插入到正确的位置。这是非常有效的。

有两个不错的选择。

备选案文1: Worker.terminate()

第一种方法是杀死你现有的网络工作者,然后开始一个新的工作。你可以用 Worker.terminate() .

这个 terminate() 方法 Worker 工人

• 你失去了所有的工人。如果您必须为请求将大量数据复制到其中,那么您必须再次执行此操作。
• 它涉及到线程的创建和销毁,这并不像大多数人想象的那么慢,但是如果你终止web workers,那么 这可能会引起一些问题。

如果这两个都不是问题,那么这可能是最简单的选择。

就我而言,我有很多州。我的工作人员正在渲染图像的一部分,当用户平移到其他区域时,我希望它停止正在执行的操作并开始渲染新区域。但是渲染图像所需的数据非常庞大。

在你的情况下,你有你不想使用的(大概是巨大的)列表的状态。

选择2:屈服

第二种选择基本上是协同多任务处理。您可以正常运行计算,但偶尔会暂停(屈服)并说“我应该停止吗?”,就像这样(这是为了一些无意义的计算,而不是排序)。

let requestId = 0;

onmessage = event => {
  ++requestId;
  sortAndSendData(requestId, event.data);
}

function sortAndSendData(thisRequestId, data) {
  let isSorted = false;
  let total = 0;

  while (data !== 0) {
    // Do a little bit of computation.
    total += data;
    --data;

    // Check if we are still the current request ID.
    if (thisRequestId !== requestId) {
      // Data was changed. Cancel this sort.
      return;
    }
  }

  postMessage(total);
}

但这行不通,因为 sortAndSendData() 运行到完成并阻止web worker的事件循环。我们需要一些方法在这之前让步 thisRequestId !== requestId . 不幸的是,Javascript没有很好的应用程序 yield 方法它确实有 async await 所以我们可以试试这个:

let requestId = 0;

onmessage = event => {
  console.log("Got event", event);
  ++requestId;
  sortAndSendData(requestId, event.data);
}

async function sortAndSendData(thisRequestId, data) {
  let isSorted = false;
  let total = 0;

  while (data !== 0) {
    // Do a little bit of computation.
    total += data;
    --data;

    await Promise.resolve();

    // Check if we are still the current request ID.
    if (thisRequestId !== requestId) {
      console.log("Cancelled!");
      // Data was changed. Cancel this sort.
      return;
    }
  }

  postMessage(total);
}

不幸的是,它不起作用。我想是因为 异步的 / 等候 使用“微任务”急切地执行任务,如果可能,在挂起“宏任务”(我们的web worker消息)之前执行这些任务。

等候 要成为一个宏任务,可以使用 setTimeout(0)

let requestId = 0;

onmessage = event => {
  console.log("Got event", event);
  ++requestId;
  sortAndSendData(requestId, event.data);
}

function yieldToMacrotasks() {
  return new Promise((resolve) => setTimeout(resolve));
}

async function sortAndSendData(thisRequestId, data) {
  let isSorted = false;
  let total = 0;

  while (data !== 0) {
    // Do a little bit of computation.
    total += data;
    --data;

    await yieldToMacrotasks();

    // Check if we are still the current request ID.
    if (thisRequestId !== requestId) {
      console.log("Cancelled!");
      // Data was changed. Cancel this sort.
      return;
    }
  }

  postMessage(total);
}

这管用!但是它非常慢。 await yieldToMacrotasks() 设置超时(0)

幸运的是,另一位用户向我指出 a quicker way . 基本上是在另一台计算机上发送消息 MessageChannel 也会屈服于事件循环,但不受最小延迟的影响,如 是此代码有效,每个循环只需约0.04 ms,这应该可以。

let currentTask = {
  cancelled: false,
}

onmessage = event => {
  currentTask.cancelled = true;
  currentTask = {
    cancelled: false,
  };
  performComputation(currentTask, event.data);
}

async function performComputation(task, data) {
  let total = 0;

  let promiseResolver;

  const channel = new MessageChannel();
  channel.port2.onmessage = event => {
    promiseResolver();
  };

  while (data !== 0) {
    // Do a little bit of computation.
    total += data;
    --data;

    // Yield to the event loop.
    const promise = new Promise(resolve => {
      promiseResolver = resolve;
    });
    channel.port1.postMessage(null);
    await promise;

    // Check if this task has been superceded by another one.
    if (task.cancelled) {
      return;
    }
  }

  // Return the result.
  postMessage(total);
}

我对此并不完全满意——这取决于 postMessage() 事件以FIFO顺序处理,我怀疑这是有保证的。我怀疑你可以重写代码使它工作,即使那不是真的。