TypeScript 中的并发与多线程：进阶篇

TypeScript中的并发与多线程：高级水平

在现代Web开发和基于TypeScript的服务器应用中，有效处理并发与多线程的能力变得至关重要。虽然JavaScript是单线程语言，但运行时环境（浏览器或Node.js）提供了强大的工具来并行执行任务。在本文中，我们将探讨TypeScript中并发处理的高级技术，包括Worker Threads、SharedArrayBuffer、Atomics以及同步模式。

1. 基础原理：Event Loop与async/await

在深入多线程之前，理解TypeScript中Event Loop的工作原理非常重要。单线程环境中的并发是通过非阻塞I/O和异步操作实现的。

来看一个使用Promise.allSettled()并发执行多个异步任务并处理错误的例子：

interface TaskResult<T> {  status: 'fulfilled' | 'rejected';  value?: T;  reason?: Error;}

async function runConcurrentTasks<T>(  tasks: Promise<T>[]): Promise<TaskResult<T>[]> {  const results = await Promise.allSettled(tasks);    return results.map(result => {    if (result.status === 'fulfilled') {      return { status: 'fulfilled', value: result.value };    } else {      return { status: 'rejected', reason: result.reason };    }  });}

// 使用示例const task1 = fetchData('/api/users');const task2 = fetchData('/api/posts');const task3 = fetchData('/api/comments');

const results = await runConcurrentTasks([task1, task2, task3]);

2. Node.js中的Worker Threads：真正的多线程

对于CPU密集型任务（例如图像处理、加密、复杂计算），需要真正的线程。在Node.js中，这通过Worker Threads实现。

让我们创建一个带类型定义的Worker：

// worker.tsimport { parentPort, workerData } from 'worker_threads';

interface WorkerInput {  data: number[];  operation: 'sum' | 'multiply';}

interface WorkerOutput {  result: number;  duration: number;}

if (parentPort) {  const input: WorkerInput = workerData;  const start = performance.now();    let result: number;    if (input.operation === 'sum') {    result = input.data.reduce((acc, val) => acc + val, 0);  } else {    result = input.data.reduce((acc, val) => acc * val, 1);  }    const output: WorkerOutput = {    result,    duration: performance.now() - start  };    parentPort.postMessage(output);}

现在创建一个管理器来管理多个Worker：

// threadPool.tsimport { Worker } from 'worker_threads';import path from 'path';

class ThreadPool<TInput, TOutput> {  private workers: Worker[] = [];  private taskQueue: Array<{    input: TInput;    resolve: (value: TOutput) => void;    reject: (reason?: any) => void;  }> = [];

  constructor(    private workerScript: string,    private poolSize: number = 4  ) {    this.initialize();  }

  private initialize(): void {    for (let i = 0; i < this.poolSize; i++) {      const worker = new Worker(this.workerScript);            worker.on('message', (result: TOutput) => {        const task = this.taskQueue.shift();        if (task) {          task.resolve(result);        }        worker.postMessage(this.taskQueue[0]?.input);      });

      worker.on('error', (error) => {        const task = this.taskQueue.shift();        if (task) {          task.reject(error);        }      });

      this.workers.push(worker);    }  }

  async execute(input: TInput): Promise<TOutput> {    return new Promise((resolve, reject) => {      this.ta