Конкурентность и многопоточность в TypeScript: продвинутый уровень

Конкурентность и многопоточность в TypeScript: продвинутый уровень

В современной веб-разработке и серверных приложениях на TypeScript умение эффективно работать с конкурентностью и многопоточностью становится критически важным. Хотя JavaScript является однопоточным языком, среда выполнения (браузер или Node.js) предоставляет мощные инструменты для параллельного выполнения задач. В этой статье мы рассмотрим продвинутые техники работы с конкурентностью в TypeScript, включая Worker Threads, SharedArrayBuffer, Atomics и паттерны синхронизации.

1. Фундаментальные основы: Event Loop и async/await

Прежде чем погружаться в многопоточность, важно понимать, как работает Event Loop в TypeScript. Конкурентность в однопоточной среде достигается за счет неблокирующего ввода-вывода и асинхронных операций.

Рассмотрим пример с использованием 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. Worker Threads в Node.js: настоящая многопоточность

Для CPU-интенсивных задач (например, обработка изображений, криптография, сложные вычисления) необходимы настоящие потоки. В Node.js для этого используются Worker Threads.

Создадим типизированный Worker с TypeScript:

// worker.ts
import { 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.ts
import { 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.taskQueue.push({ input, resolve, reject });
      
      const availableWorker = this.workers.find(
        w => w.threadId !== undefined
      );
      
      if (availableWorker) {
        availableWorker.postMessage(input);
      }
    });
  }

  async terminate(): Promise<void> {
    await Promise.all(
      this.workers.map(worker => worker.terminate())
    );
  }
}

3. SharedArrayBuffer и Atomics: разделяемая память

Для максимальной производительности можно использовать разделяемую память между потоками с помощью SharedArrayBuffer и атомарных операций Atomics.

// sharedMemory.ts
interface SharedCounter {
  buffer: SharedArrayBuffer;
  counter: Int32Array;
}

function createSharedCounter(initialValue: number = 0): SharedCounter {
  const buffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
  const counter = new Int32Array(buffer);
  counter[0] = initialValue;
  return { buffer, counter };
}

class AtomicCounter {
  private counter: Int32Array;

  constructor(shared: SharedCounter) {
    this.counter = shared.counter;
  }

  increment(): number {
    return Atomics.add(this.counter, 0, 1);
  }

  decrement(): number {
    return Atomics.sub(this.counter, 0, 1);
  }

  getValue(): number {
    return Atomics.load(this.counter, 0);
  }

  compareAndSwap(expected: number, newValue: number): boolean {
    const result = Atomics.compareExchange(
      this.counter, 0, expected, newValue
    );
    return result === expected;
  }
}

// Пример использования в Worker
// worker.ts
import { parentPort, workerData } from 'worker_threads';

const shared = workerData as SharedCounter;
const counter = new AtomicCounter(shared);

// Безопасное обновление счетчика
const oldValue = counter.increment();
parentPort?.postMessage({ oldValue, newValue: oldValue + 1 });

4. Продвинутые паттерны синхронизации

При работе с многопоточностью важны паттерны синхронизации. Рассмотрим реализацию мьютекса с использованием Atomics:

class Mutex {
  private locked: Int32Array;
  private static UNLOCKED = 0;
  private static LOCKED = 1;

  constructor(sharedBuffer?: SharedArrayBuffer) {
    if (sharedBuffer) {
      this.locked = new Int32Array(sharedBuffer);
    } else {
      const buffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT);
      this.locked = new Int32Array(buffer);
      this.locked[0] = Mutex.UNLOCKED;
    }
  }

  async lock(): Promise<void> {
    while (true) {
      const oldValue = Atomics.compareExchange(
        this.locked,
        0,
        Mutex.UNLOCKED,
        Mutex.LOCKED
      );
      
      if (oldValue === Mutex.UNLOCKED) {
        return;
      }
      
      // Ожидание с yield для эффективности
      await new Promise(resolve => setTimeout(res