工人池

工人池 提供一种简单的方式,为动态卸载计算和管理一批有献身精神的工人创建了一批工人. 工人池 基本上执行 线程池模式. 有一批工人来执行任务。 新任务被排入队列 。 工人一次执行一个任务,完成后从队列中选择一个新的任务. 工人可以通过一个自然的、以承诺为基础的代理进入,似乎他们可以在主要应用中直接获得.

工人池 在Node.js和浏览器上运行.

特征

• 易于使用
• 在浏览器中运行, Node.js, and 阿宝
• 向工人动态卸载功能
• 工作人员通过代理
• 取消运行中的任务
• 设定任务超时
• 倒车工人
• 小型: 9 kB 微调和克齐普 (JS 构建)
• 支持可转让对象( 仅针对网络工作者和工作者 线索)
• 类型脚本构建 o(1) 循环缓冲队列(比 JS Array. shift快)
• 可选 WASM 模块 多工作者共享内存设想(见 WASM 限制(中文(简体) )
• 高级波尔 智能工人日程安排(工人选择策略、偷活、任务亲和)
• Bun 兼容 自动运行时间检测和优化配置

为什么

JavaScript是基于一个单个事件循环,一次处理一个事件. 杰里米·爱泼斯坦 说清楚点数字 :

在Node.js中,一切都平行运行, 除了你的代码. 这意味着你在Node.js中写的所有I/O代码都是无阻的, 而(反之)您在 Node.js 中写入的所有非 I/O 代码被屏蔽 .

这意味着CPU繁重的任务会阻碍其他任务被执行. 如果出现浏览器环境,浏览器在执行CPU密集任务(浏览器"吊")时不会对鼠标点击等用户事件作出反应. 如果出现节点.js服务器,服务器在执行单一而沉重的请求时不会响应任何新的请求.

对于前端进程来说,这不是一个理想的情况. 因此,CPU密集任务应从主事件循环卸入专用 劳动者在浏览器环境中, 网络工人 可以使用。 在节点.js, 儿童过程 财务报告和审定财务报表 工作线条( t) 备有。 应用程序应当被分拆成独立的,去相接的部分,这些部分可以互相平行运行. 这实际上形成了一种通过孤立的过程和传递信息实现通融的结构.

安装

通过 npm 安装 :

npm install workerpool

装入

将工人池装入节点应用程序(主要应用程序以及工人):

const workerpool = require('workerpool');

在浏览器中加载工人池 :

<script src="workerpool.js"></script>

要在浏览器中的网络工人中加载工人池:

importScripts('workerpool.js');

使用React或webpack5建立工人池需要额外的配置步骤,如本报告所述 网页包装5 部分。 。 。 .

Bun 支持

类型脚本构建( T)workerpool/modern)提供头等舱支持 阿宝 运行时间 :

// Using the TypeScript build with Bun
const workerpool = require('workerpool/modern');

// Option 1: Use optimalPool() for automatic best configuration
const pool = workerpool.optimalPool();

// Option 2: Explicitly use thread workers (recommended for Bun)
const pool = workerpool.pool({ workerType: 'thread' });

// Check runtime information
console.log(workerpool.getRuntimeInfo());
// { runtime: 'bun', version: '1.3.4', recommendedWorkerType: 'thread', ... }

// Check worker type support
console.log(workerpool.getWorkerTypeSupport());
// { thread: true, process: false, web: false, auto: true }

Bun 的重要注释 :

• 使用 workerType: 'thread' 或者说 workerType: 'auto' (两种工作都正确)
• 避免 workerType: 'process' 因为它知道IPC的问题与 Bun's child_process.fork()
• 库自动选择 'thread' 当在 Bun 中运行时 'auto' 或默认设置
• 所有533 TypeScript 测试通过 Bun

见 文件/附录 用于详细兼容信息.

使用

动态卸载函数

在以下例子中有一个函数 add,它被动态地卸载给一个要被执行的工人,用于一组特定的论证.

我的书

const workerpool = require('workerpool');
const pool = workerpool.pool();

function add(a, b) {
  return a + b;
}

pool
  .exec(add, [3, 4])
  .then(function (result) {
    console.log('result', result); // outputs 7
  })
  .catch(function (err) {
    console.error(err);
  })
  .then(function () {
    pool.terminate(); // terminate all workers when done
  });

请注意,功能和论据都必须是静态的和可弦的,因为它们需要以序列形式发送给工人。 如果存在大型功能或功能论证,将数据发送给工人的间接费用可能很大.

专职工人

专用的工人可以单独用文字创作,然后通过工人池来使用.

我的员工.js

const workerpool = require('workerpool');

// a deliberately inefficient implementation of the fibonacci sequence
function fibonacci(n) {
  if (n < 2) return n;
  return fibonacci(n - 2) + fibonacci(n - 1);
}

// create a worker and register public functions
workerpool.worker({
  fibonacci: fibonacci,
});

这个工人可以被一个工人池使用:

我的书

const workerpool = require('workerpool');

// create a worker pool using an external worker script
const pool = workerpool.pool(__dirname + '/myWorker.js');

// run registered functions on the worker via exec
pool
  .exec('fibonacci', [10])
  .then(function (result) {
    console.log('Result: ' + result); // outputs 55
  })
  .catch(function (err) {
    console.error(err);
  })
  .then(function () {
    pool.terminate(); // terminate all workers when done
  });

// or run registered functions on the worker via a proxy:
pool
  .proxy()
  .then(function (worker) {
    return worker.fibonacci(10);
  })
  .then(function (result) {
    console.log('Result: ' + result); // outputs 55
  })
  .catch(function (err) {
    console.error(err);
  })
  .then(function () {
    pool.terminate(); // terminate all workers when done
  });

工人还可以同步初始化:

my Asyncworker.js (英语)

define(['workerpool/dist/workerpool'], function (workerpool) {
  // a deliberately inefficient implementation of the fibonacci sequence
  function fibonacci(n) {
    if (n < 2) return n;
    return fibonacci(n - 2) + fibonacci(n - 1);
  }

  // create a worker and register public functions
  workerpool.worker({
    fibonacci: fibonacci,
  });
});

实例

实例目录中有:

https://github.com/josdejong/workerpool/tree/master/examples. 中国植物物种信息数据库

API 密码

工人池的API由两个部分组成: workerpool.pool 创建工人池和功能 workerpool.worker 来创造一个工人.

组合

使用此函数创建工人池 workerpool.pool数字 :

workerpool.pool([script: string] [, options: Object]) : Pool

当a script 提供论据,所提供的脚本将作为一个专职工作者开始。 否时 script 参数,一个默认的工人被启动,可以用来动态地通过 Pool.exec注意在节点上, script 必须是绝对文件路径, 如 __dirname + '/myWorker.js'在浏览器环境中, script 也可以像数据 URL 'data:application/javascript;base64,...'。这允许将工人捆绑的代码嵌入您的主要应用程序中。 见 examples/embeddedWorker 为演示.

有以下备选方案:

• minWorkers: number | 'max'必须初始化和保留的最低工人人数。 设为 'max' 将创建 maxWorkers 违约工人(见下文).
• maxWorkers: number。最大工作人的默认数字是CPU的倒数。 当无法确定 CPU 的数目时( 例如在更老的浏览器中), maxWorkers 设置为 3.
• maxQueueSize: number。允许排队的最大任务数量。 可用于防止内存耗尽. 如果超过最大值, 添加新任务会丢出错误 。 默认值为 Infinity。 。 。 .
• workerType: 'auto' | 'web' | 'process' | 'thread'。 。 。 .
  • 如果发生 'auto' (默认), 工人池会自动选择合适的工人类型: 当在浏览器环境中, 'web' 将被使用。 当在节点环境中, worker_threads 如果可用(Node.js QX11.7.0),则将使用其他方法 child_process 将被使用.
  • 如果发生 'web',将使用网络工作员。 仅在浏览器环境中可用 .
  • 如果发生 'process', (中文(简体) ), child_process 将被使用。 仅在节点.js环境中可用.
  • 如果发生 'thread', (中文(简体) ), worker_threads 将被使用。 若为 worker_threads 不可用,则出现出错。 仅在节点.js环境中可用.
• workerTerminateTimeout: number数以毫秒计的超时等待工人在终止前清理资源。 默认值是 1000。 。 。 .
• abortListenerTimeout: number。在数以毫秒计的超时中,等待中止的听众在强力阻止它,触发了清理。 默认值是 1000。 。 。 .
• forkArgs: String[]用于 process 工人类型。 流经的数组 args 改为 子进程.fork
• forkOpts: Object用于 process 工人类型。 被传递为 options 改为 子进程.fork。关于可用的选项,请参见节点文档.
• workerOpts: Object用于 web 工人类型。 传递到 网络工人的构造者。见 工人备选规格 对于可用的选项.
• workerThreadOpts: Object用于 worker 工人类型。 传递到 工作线条. options。关于可用的选项,请参见节点文档.
• onCreateWorker: Function. 每当创建工人时,都使用回调。 例如,它可用于为每个工人分配资源。 回调被传递为具有下列属性的参数对象:
  • forkArgs: String[]页:1 forkArgs 此池的选项
  • forkOpts: Object页:1 forkOpts 此池的选项
  • workerOpts: Object页:1 workerOpts 此池的选项
  • script: string页:1 script 此池的选项 可选的是,这种回调可以返回包含以上一个或多个属性的对象. 所提供的财产将用来取代正在创建的工人的Pool财产.
• onTerminateWorker: Function当工人被终止时,就叫回调。 它可以用来释放可能分配给这一具体工人的资源。 回调是作为参数传递的,它是描述的对象 onCreateWorker,每套财产中,对工人的价值将被终止.
• emitStdStreams: boolean用于 process 或者说 thread 工人类型。 若为 true,工人会发射 stdout 财务报告和审定财务报表 stderr 事件而不是传递到母流。 默认值是 false。 。 。 .

重要说明 'workerType':在向工人发送和接收原始数据类型(Plane JSON)时,不同的工人类型('web', (中文(简体) ), 'process', (中文(简体) ), 'thread')可以互用. 然而,当使用更高级的数据类型如缓冲器时,API和返回的结果可能有所不同. 在这种情况下,最好不要使用 'auto' 设置但有固定 'workerType' 和良好的单位测试到位.

工人人才库具有以下功能:

• Pool.exec(method: Function | string, params: Array | null [, options: Object]) : Promise<any, Error>
  用给定的论据执行工人的职能.

  • 何时 method 是一个字符串,使用这个名称的方法必须存在于工人身上,必须注册才能通过池子进入。 该功能将在给定参数的工人身上执行.
  • 何时 method 是函数, 提供的函数 fn 将被字符串化,发送到工人, 并在那里执行 提供参数。 所提供的功能必须是静态的,决不能依赖于周围范围的变量.
  • 有以下备选方案:
    • on: (payload: any) => void一个事件听众,处理工人为这一处决而发送的事件。 见 活动 更多细节.
    • transfer: Object[]. 向工人发送的可转让物品清单。 未获得支持 process 工人类型。 见 实例 用于使用.

• Pool.proxy() : Promise<Object, Error>
  创建工人池的代理 。 代用词包含一个代用词,用于介绍工人可以使用的所有方法. 所有方法都返回承诺 解决方法的结果.

• Pool.stats() : Object
  检索关于工人的统计数据,以及积极和未完成的任务.

  返回包含以下属性的对象:

  {
    totalWorkers: 0,
    busyWorkers: 0,
    idleWorkers: 0,
    pendingTasks: 0,
    activeTasks: 0
  }
  

• Pool.terminate([force: boolean [, timeout: number]]) : Promise<void, Error>

  如果参数 force 工人将完成他们的工作,然后结束他们的工作。 任何未决任务都会被错误的“ Pool 已终止 ” 拒绝 。 何时 force 是真的,所有工人都立即被解雇,没有完成各项工作。 若为 timeout 规定工人在停工期满而工人尚未完成时被强制解雇.

职能 Pool.exec 全部返回 a Promise该承诺具有以下功能:

• Promise.then(fn: Function<result: any>) : Promise<any, Error>
  得到承诺的结果 一旦决心.
• Promise.catch(fn: Function<error: Error>) : Promise<any, Error>
  得到承诺的错误 当被拒绝.
• Promise.finally(fn: Function<void>)
  承诺执行时的逻辑 resolves 或者说 rejects
• Promise.cancel() : Promise<any, Error>
  运行中的任务可以取消 。 执行这项任务的工人被强制立即终止. 承诺将被拒绝 Promise.CancellationError。 。 。 .
• Promise.timeout(delay: number) : Promise<any, Error>
  当一个运行中的任务在给定的延迟以毫秒内没有得到解决或拒绝时,取消它。 计时器将在任务实际开始时开始,而不是在任务创建并排队时开始. 执行这项任务的工人被强制立即终止. 承诺将被拒绝 Promise.TimeoutError。 。 。 .

示例用法:

const workerpool = require('workerpool');

function add(a, b) {
  return a + b;
}

const pool1 = workerpool.pool();

// offload a function to a worker
pool1
  .exec(add, [2, 4])
  .then(function (result) {
    console.log(result); // will output 6
  })
  .catch(function (err) {
    console.error(err);
  });

// create a dedicated worker
const pool2 = workerpool.pool(__dirname + '/myWorker.js');

// supposed myWorker.js contains a function 'fibonacci'
pool2
  .exec('fibonacci', [10])
  .then(function (result) {
    console.log(result); // will output 55
  })
  .catch(function (err) {
    console.error(err);
  });

// send a transferable object to the worker
// supposed myWorker.js contains a function 'sum'
const toTransfer = new Uint8Array(2).map((_v, i) => i)
pool2
  .exec('sum', [toTransfer], { transfer: [toTransfer.buffer] })
  .then(function (result) {
    console.log(result); // will output 3
  })
  .catch(function (err) {
    console.error(err);
  });

// create a proxy to myWorker.js
pool2
  .proxy()
  .then(function (myWorker) {
    return myWorker.fibonacci(10);
  })
  .then(function (result) {
    console.log(result); // will output 55
  })
  .catch(function (err) {
    console.error(err);
  });

// create a pool with a specified maximum number of workers
const pool3 = workerpool.pool({ maxWorkers: 7 });

劳动者

工人的构成如下:

workerpool.worker([methods: Object<String, Function>] [, options: Object]) : void

参数 methods 是可选的,可以是具有工人可使用功能的对象。 登记的职能将通过工人人才库提供.

有以下备选方案:

• onTerminate: ([code: number]) => Promise<void> | void当工人被终止时,就叫回调。 它可以用来释放可能分配给这一具体工人的资源。 和游泳池的区别 onTerminateWorker 这种回调是在工人背景下运行的,而 onTerminateWorker 在主线上执行.

示例用法:

// file myWorker.js
const workerpool = require('workerpool');

function add(a, b) {
  return a + b;
}

function multiply(a, b) {
  return a * b;
}

// create a worker and register functions
workerpool.worker({
  add: add,
  multiply: multiply,
});

通过从工人的某个职能中返回一个承诺,可以处理同步的结果:

// file myWorker.js
const workerpool = require('workerpool');

function timeout(delay) {
  return new Promise(function (resolve, reject) {
    setTimeout(resolve, delay);
  });
}

// create a worker and register functions
workerpool.worker({
  timeout: timeout,
});

可转让对象可以使用 Transfer 帮助类 :

// file myWorker.js
const workerpool = require('workerpool');

function array(size) {
  var array = new Uint8Array(size).map((_v, i) => i);
  return new workerpool.Transfer(array, [array.buffer]);
}

// create a worker and register functions
workerpool.worker({
  array: array,
});

活动

您可以在工作执行时将数据从工人运回池中 workerEmit 函数 :

workerEmit(payload: any) : unknown

此功能只在工人内部工作 财务报告和审定财务报表 在任务期间.

示例

// file myWorker.js
const workerpool = require('workerpool');

function eventExample(delay) {
  workerpool.workerEmit({
    status: 'in_progress',
  });

  workerpool.workerEmit({
    status: 'complete',
  });

  return true;
}

// create a worker and register functions
workerpool.worker({
  eventExample: eventExample,
});

为了接收这些事件,你可以使用 on 池选项 exec 方法 :

pool.exec('eventExample', [], {
  on: function (payload) {
    if (payload.status === 'in_progress') {
      console.log('In progress...');
    } else if (payload.status === 'complete') {
      console.log('Done!');
    }
  },
});

劳动者 API

工人可使用 worker api 包含以下方法

• emit: (payload: unknown | Transfer): void
• addAbortListener: (listener: () => Promise<void>): void

添加自动收听器

工人的解雇可通过下列途径收回: abort listeners 登记通过 worker.addAbortListener如果所有注册的听众都下决心,那么工人将不会被解雇,在某些情况下允许工人再利用.

注:为了成功清理作业,工人应执行 自动同步如果工人的线被堵住,工人就会被杀死.

function asyncTimeout() {
  var me = this;
  return new Promise(function (resolve) {
    let timeout = setTimeout(() => {
        resolve();
    }, 5000);

    // Register a listener which will resolve before the time out
    // above triggers.
    me.worker.addAbortListener(async function () {
        clearTimeout(timeout);
        resolve();
    });
  });
}

// create a worker and register public functions
workerpool.worker(
  {
    asyncTimeout: asyncTimeout,
  },
  {
    abortListenerTimeout: 1000
  }
);

发射

事件也可能从 worker api通过 worker.emit

// file myWorker.js
const workerpool = require('workerpool');

function eventExample(delay) {
  this.worker.emit({
    status: "in_progress",
  });
  workerpool.workerEmit({
    status: 'complete',
  });

  return true;
}

// create a worker and register functions
workerpool.worker({
  eventExample: eventExample,
});

公用事业

以下属性可供使用:

• 平台: Javascript平台. 无论是 节点 或者说 浏览器
• 是主路径: 代码是否在主线程中运行( 工人)
• cpus (英语): 可用的CPU/分数

TypeScript API( 工作池/ 现代)

TypeScript构建为现代应用提供了全面的API. 根据您的需要从不同的入境点导入:

// Modern build (~20KB) - recommended for most use cases
import * as workerpool from 'workerpool/modern';

// Minimal build (~5KB) - core features only
import * as workerpool from 'workerpool/minimal';

// Full build (~34KB) - all features including WASM
import * as workerpool from 'workerpool/full';

平台检测

所有 TypeScript 构建导出平台检测工具 :

import {
  platform,           // 'node' or 'browser'
  isMainThread,       // boolean - true if running in main thread
  cpus,               // number - CPU count
  isNode,             // function - check if running in Node.js
  getPlatformInfo,    // function - get complete platform info
  hasWorkerThreads,   // boolean - worker_threads available
  hasSharedArrayBuffer, // boolean - SharedArrayBuffer available
  hasAtomics,         // boolean - Atomics available
} from 'workerpool/minimal';

// Get complete platform info
const info = getPlatformInfo();
// { platform, isMainThread, cpus, hasWorkerThreads, hasSharedArrayBuffer, hasAtomics, isBun, bunVersion, ... }

Bun 兼容性

TypeScript 架构包括 Bun 运行时检测 :

• 是本语: Boolean 表示是否在 Bun 运行时运行
• 包发: Bun 版本字符串如果在 Bun 中运行, 否则无效
• 建议的工人类型: 当前运行时间的最佳工人类型( Bun 的“ 线程 ” 、 Node.js 的“ 自动 ” 、 浏览器的“ web ” )
• 获取工人 Type 支持 (): 回返支助矩阵 { thread, process, web, auto } 所有工人类型
• 已支持工人类型:检查是否完全支持特定工人类型
• 最佳Pool( [script], [选项]): 创建一个具有当前运行时间最佳设置的池
• 获取运行时信息 (): 回归 { runtime, version, recommendedWorkerType, workerTypeSupport }

高级pool(工人池/现代和/完整)

AdvancedPool为有智慧的工人列表提供了多种工人选择策略,盗取工作,以及任务亲和性:

import {
  advancedPool,           // Create pool with advanced scheduling
  cpuIntensivePool,       // Optimized for CPU-bound tasks
  ioIntensivePool,        // Optimized for I/O-bound tasks
  mixedWorkloadPool,      // Balanced for mixed workloads
} from 'workerpool/full';

// Create an advanced pool with intelligent scheduling
const pool = advancedPool('./worker.js', {
  workerChoiceStrategy: 'least-busy',  // or 'round-robin', 'least-used', 'fair-share'
  enableWorkStealing: true,
  enableTaskAffinity: true,
});

// Execute with task affinity (tasks with same key go to same worker)
await pool.execWithAffinity('user-123', 'processData', [data]);

// Execute with task type hint (routes to best performer)
await pool.execWithType('image-processing', 'resize', [image]);

// Change strategy at runtime
pool.setWorkerChoiceStrategy('fair-share');

// Get advanced statistics
const stats = pool.stats();
console.log(stats.workStealingStats?.totalSteals);

工人选择战略:

• round-robin - 甚至是轮流分配
• least-busy - 积极工作最少的工人(最适合I/O级)
• least-used - 完成工作最少的工人
• fair-share - 按总执行时间的平衡(最多应付混合工作量)
• weighted-round-robin - 可配置的工人体重
• interleaved-weighted-round-robin - 更平滑的加权分布

工作盗窃: 从繁忙的工人到闲闲的工人,自动地重新平衡任务.

任务无限 : 前往同一工人的缓存位置的路线.

检测(工人库/现代和/完全)

探测和优化可转移物体的效用:

import {
  isTransferable,        // Check if value is transferable
  detectTransferables,   // Find all transferables in an object tree
  getTransferableType,   // Get the type of a transferable
  validateTransferables, // Validate transfer list before sending
} from 'workerpool/modern';

const buffer = new ArrayBuffer(1024);
console.log(isTransferable(buffer)); // true
console.log(getTransferableType(buffer)); // 'ArrayBuffer'

const data = { buffer, nested: { another: new Uint8Array(256) } };
const detected = detectTransferables(data);
// { transferables: [...], totalSize: 1280, hasLargeBuffers: false, warnings: [] }

数据结构(全部建筑)

所有 TypeScript 输出的高性能数据结构构建 :

import {
  CircularBuffer,        // Fixed-size circular buffer with O(1) operations
  GrowableCircularBuffer, // Circular buffer that grows when full
  TimeWindowBuffer,      // Time-based buffer with automatic pruning
  FIFOQueue,             // First-in-first-out queue
  LIFOQueue,             // Last-in-first-out queue (stack)
} from 'workerpool/minimal';

// CircularBuffer - evicts oldest when full
const buffer = new CircularBuffer(3);
buffer.push(1); buffer.push(2); buffer.push(3);
buffer.push(4); // Evicts 1
console.log(buffer.toArray()); // [2, 3, 4]

// GrowableCircularBuffer - grows instead of evicting
const growable = new GrowableCircularBuffer(2);
growable.push(1); growable.push(2); growable.push(3);
console.log(growable.size); // 3

// FIFOQueue and LIFOQueue for task management
const fifo = new FIFOQueue();
fifo.push(task1); fifo.push(task2);
fifo.pop(); // Returns task1 (first in)

const lifo = new LIFOQueue();
lifo.push(task1); lifo.push(task2);
lifo.pop(); // Returns task2 (last in)

平行阵列操作( 工作池/ 现代和/ 完整)

在工人之间分配工作的高性能平行阵列操作:

import { pool } from 'workerpool/modern';

// Basic operations (reduce, filter, find, etc.)
const sum = await pool.reduce(
  [1, 2, 3, 4, 5],
  (acc, x) => acc + x,
  (left, right) => left + right,
  { initialValue: 0 }
);

const evens = await pool.filter([1, 2, 3, 4, 5], x => x % 2 === 0);
// [2, 4]

const found = await pool.find(items, x => x.matches);

// Extended operations (count, partition, groupBy, etc.)
const evenCount = await pool.count([1, 2, 3, 4, 5], x => x % 2 === 0);
// 2

const [evens, odds] = await pool.partition([1, 2, 3, 4, 5], x => x % 2 === 0);
// evens = [2, 4], odds = [1, 3, 5]

const groups = await pool.groupBy(items, item => item.type);
// { typeA: [...], typeB: [...] }

const flattened = await pool.flatMap([1, 2, 3], x => [x, x * 2]);
// [1, 2, 2, 4, 3, 6]

const unique = await pool.unique([1, 2, 2, 3, 3, 3]);
// [1, 2, 3]

const hasThree = await pool.includes([1, 2, 3, 4, 5], 3);
// true

const index = await pool.indexOf([1, 2, 3, 4, 5], 3);
// 2

可用的并行操作 :

所有操作都支持取消、暂停/恢复,并进行分批执行,以优化平行.

计量收集(工作场所/现代和/完整)

import { MetricsCollector } from 'workerpool/modern';

const metrics = new MetricsCollector();
metrics.recordTaskLatency(150);
metrics.recordWorkerUtilization(0.75);
console.log(metrics.getSnapshot());
// { avgLatency, p95Latency, utilizationHistory, ... }

错误类( 全部构建)

import {
  CancellationError,  // Thrown when task is cancelled
  TimeoutError,       // Thrown when task times out
  TerminationError,   // Thrown when pool is terminated
} from 'workerpool/minimal';

优化公用事业(工人池/现代和/完全)

最佳平行处理的高性能公用事业:

import {
  // Function Compilation Cache - avoid repeated eval() overhead
  FunctionCache,        // LRU cache for compiled functions
  compileCached,        // Cache function compilation
  getGlobalFunctionCache,  // Access global cache
  clearGlobalFunctionCache, // Clear the cache

  // Worker Selection Bitmap - O(1) idle worker lookup
  WorkerBitmap,         // Fast worker state tracking
  SharedWorkerBitmap,   // Thread-safe version with Atomics

  // K-Way Merge - O(n log k) merge for parallel results
  kWayMerge,            // Generic k-way merge
  kWayMergeIndexed,     // Merge indexed items
  mergeFilterResults,   // Merge parallel filter results
  mergePartitionResults, // Merge parallel partition results
  twoWayMerge,          // Optimized 2-way merge
  adaptiveMerge,        // Auto-selects best algorithm

  // SIMD Operations - accelerated numeric processing
  SIMDProcessor,        // Unified SIMD interface
  hasSIMDSupport,       // Check SIMD availability
  simdSumF32,           // SIMD sum for Float32Array
  simdDotProductF32,    // SIMD dot product
  createNumericReducer, // Factory for sum/product/min/max

  // Auto-Transfer - zero-copy optimization
  AutoTransfer,         // Reusable transfer optimizer
  extractTransferables, // Find all transferable objects
  autoDetectTransfer,   // Intelligent transfer decisions
  wrapForTransfer,      // Prepare params with transfer list
  createTransferableChunks, // Split arrays for parallel transfer
} from 'workerpool/modern';

// Example: SIMD-accelerated numeric operations
const arr = new Float32Array(10000);
const sum = simdSumF32(arr);        // SIMD-accelerated sum
const dot = simdDotProductF32(a, b); // SIMD dot product

// Example: Function compilation caching
const cache = new FunctionCache({ maxEntries: 100, ttl: 60000 });
const fn = compileCached('(x) => x * 2');  // Cached after first compile
const fn2 = compileCached('(x) => x * 2'); // Returns cached version

// Example: K-way merge for parallel results
const merged = kWayMerge([sorted1, sorted2, sorted3], (a, b) => a - b);

// Example: Auto-transfer optimization
const autoTransfer = new AutoTransfer({ minTransferSize: 1024 });
const result = autoTransfer.prepare(largeBuffer);
if (result.shouldTransfer) {
  pool.exec('process', [data], { transfer: result.transferables });
}

完全构建外加( 工作池/ 全部)

整个大楼包括用于高级使用的更多水电设施:

import {
  // WASM support
  canUseWasm,
  WasmBridge,
  hasWasmSupport,
  hasFullWasmSupport,

  // Debug utilities
  LogLevel,
  enableDebug,
  disableDebug,

  // Worker management
  AdaptiveScaler,
  HealthMonitor,
  WorkerCache,

  // Protocol v2 - Error codes
  WorkerErrorCode,    // 1xxx: worker crashes, pool terminated
  ProtocolErrorCode,  // 2xxx: invalid messages, version mismatch
  TaskErrorCode,      // 3xxx: method not found, timeout, cancelled
  ResourceErrorCode,  // 4xxx: out of memory, SAB unavailable
  CommunicationErrorCode, // 5xxx: connection lost, channel closed
  getErrorMessage,
  isRetryableError,

  // Protocol v2 - Message priority
  MessagePriority,    // LOW=0, NORMAL=1, HIGH=2, CRITICAL=3
  PROTOCOL_VERSION,   // Current: 2

  // Heartbeat mechanism
  HeartbeatMonitor,
  createHeartbeatRequest,
  createHeartbeatResponse,

  // Binary protocol (WASM-ready)
  encodeTaskRequest,
  encodeHeartbeatResponse,
  validateHeader,
} from 'workerpool/full';

WASM 限制

重要业绩通知: WASM 模块是 4-5x 较慢 而不是由于 JavaScript-WASM 边界在上方交叉, 用于单行操作的 TypeScript 队列 .

基准结果(50K业务,Node.js v25):

何时使用WASM(queueStrategy: 'wasm':

• 通过共享箭头缓冲器共享单个任务队列的多池
• 跨工人线的无锁原子操作
• 偷盗或分配任务
• 需要主线和工人共享内存的情景

不使用 WASM 时( 默认使用) 'fifo' 相反:

• 有专职工作者的单一人才库(最常见的使用案例)
• 性能关键应用程序
• 没有共享箭头缓冲支持的环境
• 简单的任务排队, 不交叉线程共享

// Default (recommended for most use cases)
const pool = workerpool.pool({ queueStrategy: 'fifo' });  // TypeScript O(1) queue

// Only use WASM when you need shared memory across threads
const pool = workerpool.pool({ queueStrategy: 'wasm' });  // Requires SharedArrayBuffer

// Auto-fallback: tries WASM, falls back to FIFO if unavailable
const pool = workerpool.pool({ queueStrategy: 'wasm-auto' });

类型脚本构建 FIFOQueue 使用a GrowableCircularBuffer 与 O(1) 推/ pop 操作, 速度明显快于 JavaScript Array.shift() (这是O(n)),并且比WASM执行速度快,用于单行道使用.

路线图

• 执行并行处理的功能: map, (中文(简体) ), reduce, (中文(简体) ), forEach, (中文(简体) ), filter, (中文(简体) ), some, (中文(简体) ), every, (中文(简体) ), count, (中文(简体) ), partition, (中文(简体) ), groupBy, (中文(简体) ), flatMap, (中文(简体) ), unique, (中文(简体) ), includes, (中文(简体) ), indexOf, (中文(简体) ), reduceRight ✅ 已完成 - 可用类型脚本 API(workerpool/modern(中文(简体) )
• 在不支持 Web 工作的旧浏览器上执行优雅的退化 : 返回主应用程序中的处理任务. ✅ 已完成 - 说吧 MainThreadExecutor 财务报告和审定财务报表 createPoolWithFallback()
• 执行会话支持:能够通过一个 单工,可保持会话状态. ✅ 已完成 - 说吧 SessionManager 与 pool.createSession()

相关图书馆

• https://github.com/andywer/threads.js (中文(简体) )
• https://github.com/piscinajs/piscina (英语)
• https://github.com/learnboost/ cluble (中文(简体) )
• https://github.com/adambom/avil.js (中文(简体) )
• https://github.com/padolsey/ operation (英语)
• https://github.com/calvinmetcalf/catiline 页面存档备份,存于互联网档案馆
• https://github.com/Unitech/pm2 (中文(简体) )
• https://github.com/godaddy/node-cluster-service 页面存档备份,存于互联网档案馆
• https://github.com/ramesaliyev/ EasyWebWorker (中文(简体) )
• https://github.com/rvagg/node-worker-farm 互联网档案馆的存檔,存档日期2013-12-02

构建

首先从 github 复制工程 :

git clone git://github.com/josdejong/workerpool.git
cd workerpool

安装项目依赖性 :

npm install

双重构建系统

图书馆支持两个独立的建筑:

Java 脚本构建 (莱克西):

npm run build:js     # Build JavaScript bundles (src/js/ → dist/)

产出: dist/workerpool.js, (中文(简体) ), dist/workerpool.min.js, (中文(简体) ), dist/worker.js

类型脚本 + WASM 构建 (现代):

npm run build:wasm   # Build TypeScript + WASM (src/ts/ → dist/ts/)

产出: dist/ts/index.js, (中文(简体) ), dist/ts/full.js, (中文(简体) ), dist/ts/minimal.js, (中文(简体) ), dist/workerpool.wasm

TypeScript 构建提供了 O(1) 队列操作 GrowableCircularBuffer。可选的WASM模块只有利于多工作者共享内存(见 WASM 限制) (中文(简体) ).

基准制定

两者之间的业绩比较:

node benchmark.mjs

遗产建设

默认构建命令构建 JavaScript 版本 :

npm run build

这将建立来自来源的图书馆工人池.js和工人池.min.js 文档并将其放入文件夹 dist.

测试

要执行库的测试, 安装项目依赖性一次 :

npm install

然后,测试可以执行:

npm test

测试测试的密码范围:

npm run coverage

要查看覆盖结果, 请在浏览器中打开生成的报告 :

./coverage/index.html

发布

• 描述History.md中的变化.
• 在 package.json 中更新版本,运行 npm install 更新到 package-lock.json 我也一样.
• 推入克图.
• 通过 Npm 部署到 npm npm publish。 。 。 .
• 以版本编号添加 git 标签, 如 :

  git tag v1.2.3
  git push --tags
  

许可证

版权(C)2014-2025 Jos de Jong [email protected] (英语)

根据Apache许可证第2.0版("License"); 您不得使用此文件,除非符合许可证. 您可在下列地点获得许可证副本:

http://www.apache.org/licenses/LICENSE-2.0 (中文(简体) )

除非适用的法律要求或书面同意,否则软件 根据许可证发放的是在“AS IS”BASIS上发放的, 没有任何明示或暗示的警告或条件. 有关许可的特定语言的许可证 许可证的限制.