export function setTimeout(callback: (err: any) => any, time: number) { const id = uuidv4(); const worker = runWorker( path.join(__dirname, './timeout-worker.js'), (err) => { if (!timeoutState[id]) { return null; } timeoutState[id] = null; if (err) { return callback(err); } callback(null); }, { time, }, ); timeoutState[id] = worker; return id; }
首先，我们使用 UUID 包为 worker 创建一个唯一的标识符，然后用先前定义的函数 runWorker 来获取 worker。我们还向 worker 传入一个回调函数，一旦 worker 发送了数据就会被触发。最后，把 worker 保存在状态中并返回 id。
在回调函数中，我们必须检查该 worker 是否仍然存在于该状态中，因为有可能会 cancelTimeout()，这将会把它删除。如果确实存在，就把它从状态中删除，并调用传给 setTimeout 函数的 callback。
cancelTimeout 函数使用 .terminate() 方法强制 worker 退出，并从该状态中删除该这个worker：
export function cancelTimeout(id: string) { if (timeoutState[id]) { timeoutState[id].terminate(); timeoutState[id] = undefined; return true; } return false; }
如果你有兴趣，我也实现了 setInterval，代码在，但因为它对线程什么都没做（我们重用setTimeout的代码），所以我决定不在这里进行解释。
我已经创建了一个短小的测试代码，目的是检查这种方法与原生方法的不同之处。你可以。这些是结果：
native setTimeout { ms: 7004, averageCPUCost: 0.1416 } worker setTimeout { ms: 7046, averageCPUCost: 0.308 }
我们可以看到 setTimeout 有一点延迟 - 大约40ms - 这时 worker 被创建时的消耗。平均 CPU 成本也略高，但没什么难以忍受的（CPU 成本是整个过程持续时间内 CPU 使用率的平均值）。
如果我们可以重用 worker，就能够降低延迟和 CPU 使用率，这就是要实现工作池的原因。
实现工作池
如上所述，工作池是给定数量的被事先创建的 worker，他们保持空闲并监听 message 事件。一旦 message 事件被触发，他们就会开始工作并发回结果。
为了更好地描述我们将要做的事情，下面我们来创建一个由八个 thread worker 组成的工作池：
const pool = new WorkerPool(path.join(__dirname, './test-worker.js'), 8);
如果你熟悉，那么你在这里看到的逻辑几乎相同，只是一个不同的用例。
如上面的代码片段所示，我们把指向 worker 的路径和要生成的 worker 数量传给了 WorkerPool 的构造函数。
export class WorkerPool<T, N> { private queue: QueueItem<T, N>[] = []; private workersById: { [key: number]: Worker } = {}; private activeWorkersById: { [key: number]: boolean } = {}; public constructor(public workerPath: string, public numberOfThreads: number) { this.init(); } }
这里还有其他一些属性，如 workersById 和 activeWorkersById，我们可以分别保存现有的 worker 和当前正在运行的 worker 的 ID。还有 queue，我们可以使用以下结构来保存对象：
type QueueCallback<N> = (err: any, result?: N) => void; interface QueueItem<T, N> { callback: QueueCallback<N>; getData: () => T; }
callback 只是默认的节点回调，第一个参数是错误，第二个参数是可能的结果。 getData 是传递给工作池 .run() 方法的函数（如下所述），一旦项目开始处理就会被调用。 getData 函数返回的数据将传给工作线程。
在 .init() 方法中，我们创建了 worker 并将它们保存在以下状态中：
private init() { if (this.numberOfThreads < 1) { return null; } for (let i = 0; i < this.numberOfThreads; i += 1) { const worker = new Worker(this.workerPath); this.workersById[i] = worker; this.activeWorkersById[i] = false; } }
为避免无限循环，我们首先要确保线程数 > 1。然后创建有效的 worker 数，并将它们的索引保存在 workersById 状态。我们在 activeWorkersById 状态中保存了它们当前是否正在运行的信息，默认情况下该状态始终为false。
现在我们必须实现前面提到的 .run() 方法来设置一个 worker 可用的任务。
public run(getData: () => T) { return new Promise<N>((resolve, reject) => { const availableWorkerId = this.getInactiveWorkerId(); const queueItem: QueueItem<T, N> = { getData, callback: (error, result) => { if (error) { return reject(error); } return resolve(result); }, }; if (availableWorkerId === -1) { this.queue.push(queueItem); return null; } this.runWorker(availableWorkerId, queueItem); }); }
在 promise 函数里，我们首先通过调用 .getInactiveWorkerId() 来检查是否存在空闲的 worker 可以来处理数据：
private getInactiveWorkerId(): number { for (let i = 0; i < this.numberOfThreads; i += 1) { if (!this.activeWorkersById[i]) { return i; } } return -1; }