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

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

Современные веб-приложения всё чаще сталкиваются с необходимостью обработки больших объёмов данных и выполнения ресурсоёмких операций без блокировки пользовательского интерфейса. TypeScript, будучи надстройкой над JavaScript, наследует его однопоточную модель выполнения, но при этом предоставляет мощные инструменты для организации конкурентности и многопоточности. В этой статье мы разберём, как эффективно использовать эти возможности в ваших проектах.

Понимание конкурентности и многопоточности

Прежде чем углубляться в код, важно разграничить два ключевых понятия:

• Конкурентность — способность системы обрабатывать несколько задач в перекрывающиеся промежутки времени, но не обязательно одновременно. Это достигается за счёт быстрого переключения контекста.
• Многопоточность — параллельное выполнение нескольких потоков на разных ядрах процессора. В Node.js это реализуется через Worker Threads.

TypeScript работает в среде Node.js, где основной поток обрабатывает события через Event Loop. Для тяжёлых вычислений это может стать узким местом.

Асинхронность и Promise: основа конкурентности

Базовый уровень конкурентности в TypeScript строится на async/await и Promise. Эти механизмы позволяют не блокировать основной поток во время ожидания I/O операций.

// Пример асинхронной загрузки данных
interface User {
  id: number;
  name: string;
}

async function fetchUser(id: number): Promise<User> {
  const response = await fetch(`https://api.example.com/users/${id}`);
  return response.json();
}

// Параллельный запуск нескольких запросов
async function loadUsers(): Promise<User[]> {
  const userIds = [1, 2, 3, 4, 5];
  const promises = userIds.map(id => fetchUser(id));
  return Promise.all(promises);
}

loadUsers().then(users => {
  console.log('Загружено пользователей:', users.length);
});

Promise.all запускает все промисы одновременно, но они выполняются в одном потоке. Это идеально для I/O-задач, но не подходит для CPU-интенсивных операций.

Worker Threads: настоящая многопоточность

Для выполнения тяжёлых вычислений без блокировки основного потока используйте Worker Threads. Это встроенный модуль Node.js, который создаёт отдельные потоки с собственной памятью.

Создадим пример вычисления чисел Фибоначчи в отдельном потоке:

// fibonacci.worker.ts
import { parentPort } from 'worker_threads';

function fibonacci(n: number): number {
  if (n <= 1) return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

parentPort?.on('message', (n: number) => {
  const result = fibonacci(n);
  parentPort?.postMessage(result);
});

// main.ts
import { Worker } from 'worker_threads';
import path from 'path';

function runFibonacciInWorker(n: number): Promise<number> {
  return new Promise((resolve, reject) => {
    const worker = new Worker(path.resolve(__dirname, 'fibonacci.worker.ts'));
    
    worker.postMessage(n);
    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) reject(new Error(`Worker stopped with exit code ${code}`));
    });
  });
}

async function main() {
  console.time('fibonacci');
  const result = await runFibonacciInWorker(40);
  console.timeEnd('fibonacci');
  console.log(`Fibonacci(40) = ${result}`);
}

main();

Обратите внимание: для запуска TypeScript-воркеров потребуется компиляция в JavaScript. Можно использовать ts-node или предварительную компиляцию.

SharedArrayBuffer: разделяемая память между потоками

Когда потокам нужно обмениваться большими объёмами данных, копирование через postMessage может быть неэффективным. SharedArrayBuffer позволяет потокам работать с общей областью памяти.

// shared.worker.ts
import { parentPort } from 'worker_threads';

parentPort?.on('message', (buffer: SharedArrayBuffer) => {
  const view = new Int32Array(buffer);
  // Модифицируем данные в разделяемой памяти
  for (let i = 0; i < view.length; i++) {
    view[i] = view[i] * 2;
  }
  parentPort?.postMessage('done');
});

// main.ts
import { Worker } from 'worker_threads';
import path from 'path';

const buffer = new SharedArrayBuffer(1024);
const view = new Int32Array(buffer);

// Заполняем массив данными
for (let i = 0; i < view.length; i++) {
  view[i] = i;
}

const worker = new Worker(path.resolve(__dirname, 'shared.worker.ts'));
worker.postMessage(buffer);
worker.on('message', () => {
  console.log('Модифицированные данные:', view.slice(0, 10));
  worker.terminate();
});

При работе с SharedArrayBuffer обязательно используйте Atomics для синхронизации доступа, чтобы избежать состояния гонки:

// Безопасная запись с использованием Atomics
Atomics.store(view, 0, 42);
const value = Atomics.load(view, 0);
console.log(value); // 42

Пул воркеров для эффективного управления ресурсами

Создание нового воркера для каждой задачи — дорогостоящая операция. Лучше использовать пул воркеров, который переиспользует потоки:

import { Worker } from 'worker_threads';
import path from 'path';

class WorkerPool {
  private workers: Worker[] = [];
  private queue: Array<{ data: unknown; resolve: Function; reject: Function }> = [];

  constructor(private size: number, private workerPath: string) {
    for (let i = 0; i < size; i++) {
      this.workers.push(this.createWorker());
    }
  }

  private createWorker(): Worker {
    const worker = new Worker(path.resolve(__dirname, this.workerPath));
    worker.on('message', (result) => {
      const task = this.queue.shift();
      task?.resolve(result);
    });
    worker.on('error', (err) => {
      const task = this.queue.shift();
      task?.reject(err);
    });
    return worker;
  }

  execute(data: unknown): Promise<unknown> {
    return new Promise((resolve, reject) => {
      this.queue.push({ data, resolve, reject });
      const worker = this.workers[this.queue.length % this.size];
      worker.postMessage(data);
    });
  }

  terminate(): void {
    this.workers.forEach(w => w.terminate());
  }
}

// Использование
const pool = new WorkerPool(4, 'fibonacci.worker.ts');
const