Web Worker并行执行疑问：如何实现多线程并行处理？

核心结论 #

你当前的方案无法实现并行处理——单个Web Worker本质是一个独立的单线程环境，多次调用postMessage只会把任务加入该Worker的消息队列，任务会串行执行，不会同时运行多个任务。

为什么原方案不行？ #

Web Worker的消息处理机制是队列式的：即使你连续发送多条消息，Worker的onmessage回调也会按接收顺序依次处理每个任务。你的Worker代码中loadData是异步函数，但JavaScript单线程特性决定了，上一个任务的await操作会阻塞当前Worker线程，直到该任务完成后才会处理下一条消息。

正确实现多线程并行的方法：Worker池 #

要实现5-10个线程的并行，你需要创建多个Worker实例组成的Worker池，将任务分发给不同的Worker同时处理。

1. 主线程代码（创建Worker池并分发任务）

// 定义Worker数量（建议参考CPU核心数，或设置5-10）
const WORKER_COUNT = 10;
const workers = [];
const taskQueue = [];
let activeWorkerCount = 0;

// 初始化Worker池
for (let i = 0; i < WORKER_COUNT; i++) {
  const worker = new Worker("./worker/fetcher.worker.js");
  worker.onmessage = handleWorkerResponse;
  workers.push(worker);
}

// 处理Worker返回的数据
function handleWorkerResponse(e) {
  activeWorkerCount--;
  // 处理Worker返回的点位数据（示例：合并到主线程数组）
  if (Array.isArray(e.data)) {
    const [positions, colors] = e.data;
    // 这里加入你的数据处理逻辑，比如更新Three.js场景
    console.log("收到Worker返回的点位数据：", positions.length / 3, "个点");
  }
  // 从队列中取出下一个任务分配给空闲Worker
  if (taskQueue.length > 0) {
    const nextTask = taskQueue.shift();
    assignTask(nextTask);
  }
}

// 将任务分配给空闲的Worker
function assignTask(task) {
  const freeWorker = workers.find(() => activeWorkerCount < WORKER_COUNT);
  if (freeWorker) {
    activeWorkerCount++;
    freeWorker.postMessage(task);
  } else {
    // 所有Worker忙碌时，任务加入等待队列
    taskQueue.push(task);
  }
}

// 遍历任务列表，分发任务
for (let m = 0; m < keyCountMap.length; m += 2) {
  const task = [keyCountMap[m], keyCountMap[m + 1]];
  assignTask(task);
}

2. Worker代码修改（修复数据污染问题）

原Worker中的positions、colors是全局变量，会导致不同任务的数据互相污染，需要将这些变量移到loadData函数内部，保证每个任务的数据独立：

import { Copc, Key } from "copc";
import * as THREE from "three";

// 仅保留全局共享的初始化数据
let copc;
let x_min, y_min, z_min, x_max, y_max, z_max, width;
let nodePages;
const filename = "https://s3.amazonaws.com/data.entwine.io/millsite.copc.laz";

const readPoints = (id, getters, positions, colors) => {
  const returnPoint = getXyzi(id, getters);
  positions.push(
    returnPoint[0] - x_min - 0.5 * width,
    returnPoint[1] - y_min - 0.5 * width,
    returnPoint[2] - z_min - 0.5 * width
  );
  const vx = (returnPoint[3] / 65535) * 255;
  const color = new THREE.Color(); // 每个点创建独立Color实例，避免复用污染
  color.setRGB(vx, vx, vx);
  colors.push(color.r, color.g, color.b);
};

function getXyzi(index, getters) {
  return getters.map((get) => get(index));
}

async function load() {
  copc = await Copc.create(filename);
  [x_min, y_min, z_min, x_max, y_max, z_max] = copc.info.cube;
  width = Math.abs(x_max - x_min);
  const receivedData = await Copc.loadHierarchyPage(
    filename,
    copc.info.rootHierarchyPage
  );
  nodePages = receivedData.nodes;
  postMessage(200);
}

async function loadData(myRoot, pointCount) {
  // 每个任务独立维护点位和颜色数组
  const positions = [];
  const colors = [];
  const view = await Copc.loadPointDataView(filename, copc, myRoot);
  const getters = ["X", "Y", "Z", "Intensity"].map(view.getter);
  for (let j = 0; j < pointCount; j += 1) {
    readPoints(j, getters, positions, colors);
  }
  postMessage([positions, colors]);
}

load();

onmessage = function (message) {
  const [mapIndex, pointCount] = message.data;
  const myRoot = nodePages[mapIndex];
  loadData(myRoot, pointCount);
};

关键注意事项 #

• Worker数量：建议参考navigator.hardwareConcurrency获取CPU核心数，避免创建过多Worker导致系统资源耗尽。
• 任务队列：当所有Worker忙碌时，新任务会进入队列等待，保证任务不会丢失。
• 数据隔离：Worker内部必须避免全局变量存储任务相关数据，防止不同任务的数据互相覆盖。

内容的提问来源于stack exchange，提问作者Pravin Poudel