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

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

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

1. Event Loop, микротаски и макротаски: фундамент конкурентности

Понимание Event Loop — ключ к управлению конкурентностью без блокировок. В Node.js цикл событий состоит из нескольких фаз: таймеры, I/O колбэки, idle, poll, check (setImmediate) и close. Важно различать микротаски (Promise, process.nextTick) и макротаски (setTimeout, setInterval, I/O).

Микротаски выполняются сразу после завершения текущей фазы, до следующей макротаски. Это может привести к «голоданию» Event Loop, если микротаски генерируются бесконечно.

// Пример: микротаски блокируют макротаски
setTimeout(() => console.log('Макротаска'), 0);

Promise.resolve().then(() => {
  // Бесконечная микротаска
  process.nextTick(() => {
    console.log('Микротаска');
    // Рекурсивный вызов заблокирует Event Loop
    process.nextTick(() => process.nextTick(() => {}));
  });
});
// Вывод: 'Микротаска' (бесконечно), 'Макротаска' никогда не выполнится

Для продвинутого контроля используйте queueMicrotask() и setImmediate(). Помните: process.nextTick имеет приоритет перед микротасками Promise.

2. Worker Threads: настоящая многопоточность в Node.js

Модуль worker_threads позволяет выполнять JavaScript в параллельных потоках. Каждый Worker имеет собственную V8-инстанцию и изолированную память. Обмен данными происходит через сообщения или SharedArrayBuffer.

Создание простого Worker:

// main.js
const { Worker } = require('worker_threads');

function runWorker(file, data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker(file, { workerData: data });
    worker.on('message', resolve);
    worker.on('error', reject);
    worker.on('exit', (code) => {
      if (code !== 0) reject(new Error(`Worker остановлен с кодом ${code}`));
    });
  });
}

// Запуск
runWorker('./worker.js', { iterations: 1e9 })
  .then(result => console.log('Результат:', result))
  .catch(err => console.error(err));

// worker.js
const { parentPort, workerData } = require('worker_threads');

function heavyComputation(n) {
  let sum = 0;
  for (let i = 0; i < n; i++) { sum += Math.sqrt(i); }
  return sum;
}

parentPort.postMessage(heavyComputation(workerData.iterations));

Важно: Worker не имеет доступа к DOM, глобальным объектам процесса (process.env) и модулям, загруженным в главном потоке. Передавайте только сериализуемые данные.

Пул Worker-ов

Создавать Worker под каждую задачу накладно. Используйте пул:

class WorkerPool {
  constructor(file, size = 4) {
    this.workers = [];
    this.queue = [];
    for (let i = 0; i < size; i++) {
      const worker = new Worker(file);
      worker.on('message', (result) => this._handleResult(worker, result));
      worker.on('error', (err) => this._handleError(worker, err));
      this.workers.push(worker);
    }
  }

  runTask(data) {
    return new Promise((resolve, reject) => {
      this.queue.push({ data, resolve, reject });
      this._processQueue();
    });
  }

  _processQueue() {
    if (this.queue.length === 0) return;
    const worker = this.workers.find(w => !w.busy);
    if (!worker) return;
    worker.busy = true;
    const task = this.queue.shift();
    worker.postMessage(task.data);
    worker._currentTask = task;
  }

  _handleResult(worker, result) {
    worker.busy = false;
    worker._currentTask.resolve(result);
    this._processQueue();
  }

  _handleError(worker, err) {
    worker.busy = false;
    worker._currentTask.reject(err);
    this._processQueue();
  }
}

3. SharedArrayBuffer и Atomics: разделяемая память без копирования

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

Пример: счётчик с использованием Atomics

// main.js
const { Worker } = require('worker_threads');

const sharedBuffer = new SharedArrayBuffer(4); // 4 байта для Int32
const counter = new Int32Array(sharedBuffer);
counter[0] = 0;

const worker = new Worker('./atomic-worker.js');
worker.postMessage(sharedBuffer); // передаём буфер

setTimeout(() => {
  console.log('Итоговое значение:', counter[0]);
  worker.terminate();
}, 2000);

// atomic-worker.js
const { parentPort } = require('worker_threads');

parentPort.on('message', (sharedBuffer) => {
  const counter = new Int32Array(sharedBuffer);
  for (let i = 0; i < 1000000; i++) {
    // Атомарное прибавление
    Atomics.add(counter, 0, 1);
  }
  parentPort.postMessage('done');
});

Ключевые методы Atomics:

• Atomics.add() — атомарное сложение
• Atomics.sub() — атомарное вычитание
• Atomics.load() и Atomics.store() — атомарное чтение/запись
• Atomics.wait() и Atomics.notify() — механизм ожидания и уведомления (futex)
• Atomics.compareExchange() — CAS-операция

Важно: SharedArrayBuffer требует определённых заголовков безопасности (COOP, COEP) при использовании в браузере, но в Node.js работает «из коробки».

4. Избежание race conditions и deadlock

При работе с многопоточностью возникают ти