企业官网建设流程全解析

• 并发编程

导读

Piscina 是一个基于 Node.js Worker Threads 的高性能线程池实现。本文聚焦于Piscina类暴露的全部只读实例属性，逐一说明其语义、取值来源、在源码中的实现位置以及真实使用场景。读完本文，你将掌握如何用completed、duration、queueSize、idleThreads监控池的健康状态，如何用runTime、waitTime直方图定位性能瓶颈，如何用needsDrain配合drain/needsDrain事件实现优雅的背压（backpressure），以及如何计算并解读utilization这一池利用率指标。

本文所有属性均为只读（readonly），由实例内部状态派生，调用方不能直接赋值。对应实现可查阅 src/index.ts 与 src/histogram.ts。

一、基础状态属性：completed、duration、options

这三个属性回答的是最朴素的三个问题：完成了多少任务？池子运行了多久？当前用的是什么配置？

1.1completed（只读，number）

当前已完成的任务总数，是累计值，只增不减。

• 在源码中，ThreadPool维护completed : number = 0计数器（src/index.ts）；
• 每次任务回调执行时递增this.completed++（src/index.ts）；
• 对外通过get completed ()直接暴露（src/index.ts）。

无论任务成功还是失败，只要被 Worker 处理并返回结果/错误，都会计入completed。测试 test/post-task.test.ts 展示了任务完成后pool.completed变为1。

典型用途：与runTime.mean、utilization配合估算池子的吞吐总量。

1.2duration（只读，number）

自该Piscina实例创建以来经过的毫秒数。

实现非常直接（src/index.ts）：

get duration () : number { return performance.now() - this.#pool.start; }

其中start在构造时由performance.now()记录（src/index.ts）。duration是utilization计算中的核心分母，也是判断“池子冷启动后是否已经稳定运行”的简便依据。测试 test/simple-test.test.ts 断言运行若干任务后worker.duration > 0。

1.3options（只读，FilledOptions）

当前实例实际生效的配置副本。文档原话：与传给构造函数的 options 对象具有完全相同的属性。

要点（源码佐证，src/index.ts）：

this.options = withNullPrototype({ ...kDefaultOptions, ...options, filename, maxQueue: 0, });

• 默认值在kDefaultOptions中定义（src/index.ts）：minThreads默认max(floor(cpuParallelism/2), 1)，maxThreads默认cpuParallelism * 1.5，maxQueue默认Infinity，concurrentTasksPerWorker默认1，recordTiming默认true，closeTimeout默认30000；
• 构造时会做归一化：maxQueue === 'auto'时被换算为maxThreads ** 2（src/index.ts），且minThreads/maxThreads相互钳制。因此读取options拿到的是归一化后的最终值，例如'auto'已变成具体数字；
• 使用withNullPrototype（src/common.ts）创建一个无原型对象，避免原型链污染。

典型用途：程序化地检查options.maxQueue、options.maxThreads是否满足当前负载预期；文档也明确建议在配置maxQueue: 'auto'后通过options.maxQueue查看换算结果（docs/docs/api-reference/class.md）。

二、线程与队列视角：threads、idleThreads、queueSize

这三个属性描绘了线程池“有多少人在干活、多少人在排队”的动态画面。

2.1threads（只读，Worker[]）

由线程池当前使用的Node.jsWorker实例数组（来自node:worker_threads）。实现直接遍历资源池中的WorkerInfo，收集其底层worker（src/index.ts）。数组元素是原生Worker，可直接调用worker.threadId、worker.terminate()等原生 API。

注意：threads返回的是新数组（每次 getter 调用都会重建），不要依赖数组引用相等性判断变化，应比较长度或元素。

2.2idleThreads（只读，number）

当前已完成预热（ready）且没有在跑任何任务的 Worker 数量。

实现细节（src/index.ts）：

get idleThreads () : number { let count = 0; for (const workerInfo of this.#pool.workers.readyItems) { if (workerInfo.currentUsage() === 0) count++; } return count; }

• readyItems来自资源池AsynchronouslyCreatedResourcePool（src/worker_pool/base.ts），Worker 完成启动预热后由pendingItems移入；
• currentUsage()返回该 Worker 当前承接的任务数；若正在运行带 abort 信号的任务则返回Infinity（src/worker_pool/index.ts）；
• 测试 test/thread-count.test.ts 完整演示了minThreads: 2时idleThreads在2 → 1 → 0 → 2之间的变化，且destroy()后归零。

典型用途：判断池子是否还有空闲计算力，避免盲目新增任务。

2.3queueSize（只读，number）

当前等待分配给 Worker 的任务数，即排队中的任务量。

实现（src/index.ts）：

get queueSize () : number { const pool = this.#pool; return Math.max((pool.taskQueue.size + pool.skipQueue.length) - pool.pendingCapacity(), 0); }

• taskQueue是主任务队列，skipQueue用于临时存放被跳过（如等待 abort 信号处理）的任务（src/index.ts）；
• pendingCapacity()是“正在启动中、尚未 ready 的 Worker 可承接的任务容量”（src/index.ts），从排队数中扣除，因此队列越满、越接近maxQueue，queueSize越能真实反映排队压力；
• 结果是Math.max(..., 0)，保证不会出现负值。

测试 test/task-queue.test.ts 展示了maxThreads: 3时，任务从直接分发到排队的临界行为：当第 4 个任务提交时queueSize变为1。

关键联动：queueSize与drain/needsDrain事件直接相关——drain在queueSize归零时发出，needsDrain在容量满载时发出（docs/docs/api-reference/event.md）。

三、背压与容量：needsDrain

3.1 语义

needsDrain（只读，boolean）表示池子是否需要“排空”后才能继续处理新任务。

文档定义：当没有足够容量在不超出maxQueue或maxThreads的前提下处理更多任务时，该值为true。其计算综合了线程数（当前及待启动线程）、maxQueue大小以及当前在途（in-flight）任务数。

3.2 源码实现

_needsDrain标志位在ThreadPool中维护，对外由get needsDrain ()暴露（src/index.ts）。核心逻辑在_maybeDrain()（src/index.ts）：

if (maxCapacity === currentUsage && queueSize === maxQueueSize) { this._needsDrain = true; queueMicrotask(() => this.publicInterface.emit('needsDrain')); }

即当“所有 Worker 的在途任务数之和（maxCapacity = maxThreads * concurrentTasksPerWorker）”与“队列长度”同时达到上限时，置位并异步发出needsDrain事件；当queueSize重新归零时复位并发出drain事件（src/index.ts）。

3.3 使用要点与注意事项

• 用途：这是实现背压的“信号灯”。上游生产者看到pool.needsDrain === true时，应当暂停提交任务，等待'drain'事件后再继续；
• 重要限制（文档原注）：Piscina不会在maxQueue阈值之上缓冲任务，所以needsDrain并不是“池子一定会接受更多任务”的保证。超过maxQueue的提交会直接以错误拒绝，相关错误码见 docs/docs/api-reference/error-codes.md：
  • PISCINA_ERR_TASK_QUEUE_LIMIT：任务队列已达上限；
  • PISCINA_ERR_NO_TASK_QUEUE_AVAILABLE：队列被禁用且所有 Worker 已满。
• 测试 test/post-task.test.ts 验证了needsDrain事件触发时pool.needsDrain === true且pool.queueSize === 3（maxQueue: 3）。

设计哲学：_maybeDrain()的注释（src/index.ts）指出，目标正是让使用者把“等待数”维持在 0——不给任务源提交永远无法立即执行的任务，从而自然形成背压。这与 Node.js Writable stream 的drain行为对齐。

四、性能观测：runTime与waitTime直方图

4.1 属性概览

runTime与waitTime是直方图摘要对象（histogram summary），所有值单位均为毫秒：

• runTime：已完成任务的实际执行耗时分布；
• waitTime：任务在队列中等待被分配到 Worker 的耗时分布（从任务创建taskInfo.created到分发开始taskInfo.started）。

两者结构完全一致，均包含：

4.2 百分位字段p{N}的含义

凡符合p{N}模式（N 为数字，如p1、p99）的属性都代表耗时的百分位分布。文档示例：p99表示 99% 的观测耗时都小于或等于该值。本实现提供的分位点（src/histogram.ts）为：

p0_001、p0_01、p0_1、p1、p2_5、p10、p25、p50（中位数）、p75、p90、p97_5、p99、p99_9、p99_99、p99_999

尾部分位点（p99.9以上）特别适合发现“长尾慢任务”——平均值会被极端值拉高，但p99_999能精确暴露偶发的尖峰延迟。

文档中的典型输出示例（运行耗时，毫秒）：

{ average: 1880.25, mean: 1880.25, stddev: 1.93, min: 1877, max: 1882.0190887451172, p0_001: 1877, p0_01: 1877, p0_1: 1877, p1: 1877, p2_5: 1877, p10: 1877, p25: 1877, p50: 1881, p75: 1881, p90: 1882, p97_5: 1882, p99: 1882, p99_9: 1882, p99_99: 1882, p99_999: 1882 }

4.3 底层实现：基于node:perf_hooks的直方图

Piscina 使用 Node.js 内置的perf_hooks.createHistogram()（RecordableHistogram）作为底层数据结构（src/histogram.ts）：

• PiscinaHistogramHandler内部维护#runTime与#waitTime两个直方图（src/histogram.ts）；
• 记录值统一先由toHistogramIntegerNano()把毫秒转为整数纳秒（Math.max(1, Math.trunc(ms * 1000))，src/histogram.ts），再record()写入；
• 摘要输出时再统一/ 1000转回毫秒（src/histogram.ts），包括用histogram.percentile(x)计算各分位点；
• runTime记录发生在任务完成回调处：recordRunTime(performance.now() - taskInfo.started)（src/index.ts）；
• waitTime记录发生在任务分发成功时：recordWaitTime(now - task.created)（src/index.ts）；
• 对外暴露的histogram对象还带有resetRunTime()与resetWaitTime()两个重置方法（src/index.ts）。

4.4 前置条件与开关

• 默认开启：recordTiming默认为true，因此默认就会收集 run/wait 直方图；
• 若构造时传入recordTiming: false，ThreadPool不创建直方图处理器（src/index.ts），此时histogram.runTime/histogram.waitTime为undefined，utilization恒为0（见下文）；
• 测试 test/histogram.test.ts 分别验证了默认收集、recordTiming: false时不收集、以及resetRunTime()/resetWaitTime()后均值/标准差变为NaN、max归零等行为；
• 此外还有workerHistogram选项（默认false）：开启后每个 Worker 单独记录耗时直方图（src/worker_pool/index.ts），每个PiscinaWorker通过histogram属性暴露自己的摘要，测试见 test/histogram.test.ts。

4.5 快速上手示例

import { Piscina } from 'piscina'; const pool = new Piscina({ filename: new URL('./worker.js', import.meta.url).href, }); await Promise.all([ pool.run({ a: 1, b: 2 }), pool.run({ a: 3, b: 4 }), pool.run({ a: 5, b: 6 }), ]); console.log('平均执行耗时:', pool.histogram.runTime.average, 'ms'); console.log('P99 执行耗时:', pool.histogram.runTime.p99, 'ms'); console.log('P99 排队等待:', pool.histogram.waitTime.p99, 'ms'); console.log('标准差:', pool.histogram.runTime.stddev); // 观测结束后重置，开启新一轮统计 pool.histogram.resetRunTime(); pool.histogram.resetWaitTime(); await pool.destroy();

五、池利用率：utilization

5.1 定义与公式

utilization（只读，0~1 的小数）是某一时间点上，已完成任务的近似总执行耗时与线程池理论总容量之比。

文档给出的计算逻辑（源码实现见 src/index.ts）：

1. 理论容量（capacity）=duration（池子已运行毫秒数）×options.maxThreads。这代表了池子在此时间段内绝对理论上的最大聚合计算时间；
2. 近似总执行耗时（totalMeanRuntime）= 已完成任务的平均执行耗时 × 已完成任务总数，近似表示池子实际“干活”的总时间；
3. 利用率= 近似总执行耗时 ÷ 理论容量。

对应源码：

const capacity = this.duration * this.#pool.options.maxThreads; const totalMeanRuntime = (this.#pool.histogram.runTimeSummary.mean / 1000) * count; return totalMeanRuntime / capacity;

5.2 边界行为与注意事项

• 若recordTiming为false（没有直方图），或尚无任务完成（count === 0），utilization返回0（src/index.ts）；
• 因为容量是按maxThreads计算的“绝对上限”，而实际线程数通常在minThreads与maxThreads之间伸缩，所以利用率在实际使用中几乎不可能达到 1；
• 测试 test/simple-test.test.ts 断言：任务未运行时utilization === 0，跑完若干任务后0 < utilization < 1；
• 语义上这是一个“点值快照”（point-in-time ratio），适合做趋势采样而非精确计量。把utilization与duration组合成时间序列（例如按秒采样），可以画出池子的负载曲线，辅助判断是否需要调整minThreads/maxThreads。

5.3 使用场景

• 容量规划：长期utilization接近容量上限，说明maxThreads可能成为瓶颈，应考虑增大上限或拆分任务；
• 冷启动判断：池子刚创建时（duration小、任务少）utilization会偏低，可据此判断池子是否已进入稳定工作状态；
• 与queueSize交叉分析：utilization高 +queueSize持续增长，说明“线程不够用”；utilization低 +queueSize高，说明任务分发或单个 Worker 处理存在异常。

六、属性总览与监控实践

6.1 快速参考表

6.2 一个完整的监控片段

import { Piscina } from 'piscina'; const pool = new Piscina({ filename: new URL('./worker.js', import.meta.url).href, maxThreads: 8, maxQueue: 'auto', }); // 背压：满了先停手，排空再继续 pool.on('needsDrain', () => console.log('队列已满，暂停提交')); pool.on('drain', () => console.log('队列已清空，可以继续')); setInterval(() => { console.log({ completed: pool.completed, idleThreads: pool.idleThreads, queueSize: pool.queueSize, needsDrain: pool.needsDrain, utilization: pool.utilization, runTimeP99: pool.histogram.runTime.p99, waitTimeP99: pool.histogram.waitTime.p99, }); }, 1000).unref();

这个片段同时演示了本文涉及的绝大多数属性：状态类（completed、duration）、队列类（queueSize、needsDrain、idleThreads）、直方图类（runTime、waitTime）与比率类（utilization）。将它们组合成一个心跳探针，即可对生产环境中的 Piscina 线程池做低成本、可观测的持续监控。

结语

Piscina 的只读属性虽然简单，却是线程池可观测性的全部入口：completed/duration提供时间与数量维度，threads/idleThreads/queueSize提供资源与队列维度，runTime/waitTime提供延迟分布维度，needsDrain提供背压控制信号，utilization则把前几者浓缩为一个 0~1 的利用率快照。理解它们的实现位置（src/index.ts、src/histogram.ts、src/worker_pool）与测试行为（test/simple-test.test.ts、test/histogram.test.ts、test/thread-count.test.ts、test/task-queue.test.ts），你就能在真实负载下做出有理有据的容量决策。

• 并发编程