JavaScript Worker池实现教程

// worker.js - Worker执行的代码
self.onmessage = function(e) {
  const { taskId, payload } = e.data;
  
  // 执行任务的逻辑
  const result = executeTask(payload);
  
  // 返回结果
  self.postMessage({
    taskId,
    result
  });
};

function executeTask(payload) {
  // 这里是工作线程的实际逻辑
  // 示例：简单的计算任务
  if (payload.type === 'factorial') {
    return calculateFactorial(payload.number);
  } else if (payload.type === 'fibonacci') {
    return calculateFibonacci(payload.number);
  }
  return null;
}

function calculateFactorial(n) {
  if (n === 0 || n === 1) return 1;
  let result = 1;
  for (let i = 2; i <= n; i++) {
    result *= i;
  }
  return result;
}

function calculateFibonacci(n) {
  if (n <= 1) return n;
  
  let a = 0, b = 1;
  for (let i = 2; i <= n; i++) {
    const temp = a + b;
    a = b;
    b = temp;
  }
  return b;
}

// workerPool.js - Worker池的主要实现
class WorkerPool {
  constructor(workerPath, size) {
    this.workerPath = workerPath;
    this.size = size;
    this.workers = [];
    this.availableWorkers = [];
    this.taskQueue = [];
    this.taskCallbacks = new Map();
    this.taskIdCounter = 0;
    
    this.init();
  }
  
  init() {
    // 创建指定数量的worker
    for (let i = 0; i < this.size; i++) {
      const worker = new Worker(this.workerPath);
      
      worker.onmessage = (e) => {
        const { taskId, result } = e.data;
        
        // 调用对应任务的回调
        const { resolve } = this.taskCallbacks.get(taskId);
        this.taskCallbacks.delete(taskId);
        
        // 将worker标记为可用
        this.availableWorkers.push(worker);
        
        // 如果队列中有等待的任务，则分配给空闲worker
        this.processQueue();
        
        // 完成任务的回调
        resolve(result);
      };
      
      worker.onerror = (error) => {
        console.error('Worker error:', error);
      };
      
      this.workers.push(worker);
      this.availableWorkers.push(worker);
    }
  }
  
  processQueue() {
    // 有等待的任务且有可用worker时执行任务
    if (this.taskQueue.length > 0 && this.availableWorkers.length > 0) {
      const task = this.taskQueue.shift();
      const worker = this.availableWorkers.pop();
      
      worker.postMessage(task);
    }
  }
  
  executeTask(payload) {
    return new Promise((resolve, reject) => {
      const taskId = this.taskIdCounter++;
      
      // 创建任务对象
      const task = {
        taskId,
        payload
      };
      
      // 存储任务回调
      this.taskCallbacks.set(taskId, { resolve, reject });
      
      // 如果有可用worker，直接分配任务
      if (this.availableWorkers.length > 0) {
        const worker = this.availableWorkers.pop();
        worker.postMessage(task);
      } else {
        // 否则将任务加入队列
        this.taskQueue.push(task);
      }
    });
  }
  
  terminate() {
    // 终止所有worker
    this.workers.forEach(worker => worker.terminate());
    this.workers = [];
    this.availableWorkers = [];
    this.taskQueue = [];
    this.taskCallbacks.clear();
  }
}

// 使用示例
const pool = new WorkerPool('worker.js', 4); // 创建包含4个worker的池

// 使用worker池执行任务
async function runTasks() {
  console.time('Tasks execution');
  
  // 并行执行多个任务
  const results = await Promise.all([
    pool.executeTask({ type: 'factorial', number: 10 }),
    pool.executeTask({ type: 'fibonacci', number: 30 }),
    pool.executeTask({ type: 'factorial', number: 15 }),
    pool.executeTask({ type: 'fibonacci', number: 25 }),
    pool.executeTask({ type: 'factorial', number: 20 }),
    pool.executeTask({ type: 'fibonacci', number: 35 })
  ]);
  
  console.log('All task results:', results);
  console.timeEnd('Tasks execution');
}

runTasks().finally(() => {
  // 所有任务完成后，可以选择终止worker池
  // pool.terminate();
});