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

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

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

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

Модуль worker_threads появился в Node.js 10.5.0 и позволяет создавать отдельные потоки с собственным V8-движком. В отличие от кластеризации (процессы), Worker Threads разделяют память и могут обмениваться данными через SharedArrayBuffer.

Создание и запуск воркера

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

function runWorker(data) {
  return new Promise((resolve, reject) => {
    const worker = new Worker('./worker.js', { workerData: data });
    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 () => {
  const result = await runWorker({ numbers: [1, 2, 3, 4, 5] });
  console.log('Результат:', result); // { sum: 15 }
})();

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

const sum = workerData.numbers.reduce((acc, val) => acc + val, 0);
parentPort.postMessage({ sum });

Важно: Воркеры не имеют доступа к DOM или модулям, требующим основного потока. Используйте их для тяжелых вычислений, обработки изображений или шифрования.

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

SharedArrayBuffer позволяет нескольким воркерам читать и записывать данные в одну область памяти. Для синхронизации доступа используется объект Atomics — он гарантирует атомарные операции.

Пример: параллельная обработка массива

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

const buffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 10);
const array = new Int32Array(buffer);

// Заполняем массив
for (let i = 0; i < 10; i++) array[i] = i;

const worker1 = new Worker('./worker_atomics.js', { workerData: { start: 0, end: 5 } });
const worker2 = new Worker('./worker_atomics.js', { workerData: { start: 5, end: 10 } });

worker1.on('message', () => console.log('Worker 1 завершил'));
worker2.on('message', () => console.log('Worker 2 завершил'));

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

const buffer = new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT * 10);
const array = new Int32Array(buffer);

// Атомарное увеличение каждого элемента
for (let i = workerData.start; i < workerData.end; i++) {
  Atomics.add(array, i, 10);
}

parentPort.postMessage('done');

Примечание: Используйте Atomics.wait() и Atomics.notify() для синхронизации потоков (аналог мьютексов). Это критично для предотвращения состояний гонки.

3. Кластеризация: масштабирование через процессы

Модуль cluster создает несколько процессов Node.js (воркеров), каждый из которых обрабатывает запросы. Это стандартный способ для масштабирования HTTP-серверов на многоядерных системах.

Пример: кластеризованный HTTP-сервер

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  console.log(`Master ${process.pid} is running`);

  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  cluster.on('exit', (worker, code, signal) => {
    console.log(`Worker ${worker.process.pid} died`);
    cluster.fork(); // Автоматически перезапускаем
  });
} else {
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('Hello from worker ' + process.pid);
  }).listen(8000);

  console.log(`Worker ${process.pid} started`);
}

Важно: Каждый воркер — отдельный процесс с собственной памятью. Для обмена данными используйте Redis, RabbitMQ или файловую систему. Кластеризация идеальна для REST API, веб-сокетов и прокси.

4. Продвинутые паттерны: пул воркеров и очереди

Для эффективного управления ресурсами создавайте пул воркеров. Это позволяет ограничить количество одновременно работающих потоков и избежать перегрузки системы.

Реализация пула воркеров

const { Worker } = require('worker_threads');
const { EventEmitter } = require('events');

class WorkerPool extends EventEmitter {
  constructor(workerPath, poolSize = 4) {
    super();
    this.workers = [];
    this.queue = [];
    this.activeJobs = 0;

    for (let i = 0; i < poolSize; i++) {
      const worker = new Worker(workerPath);
      worker.on('message', (msg) => this._handleResult(worker, msg));
      worker.on('error', (err) => this.emit('error', err));
      this.workers.push({ worker, busy: false });
    }
  }

  runTask(data) {
    return new Promise((resolve, reject) => {
      const task = { data, resolve, reject };
      const freeWorker = this.workers.find(w => !w.busy);

      if (freeWorker) {
        this._assignTask(freeWorker, task);
      } else {
        this.queue.push(task);
      }
    });
  }

  _assignTask(workerObj, task) {
    workerObj.busy = true;
    workerObj.worker.postMessage(task.data);
    workerObj.currentTask = task;
    this.activeJobs++;
  }

  _handleResult(worker, msg) {
    const workerObj = this.workers.find(w => w.worker === worker);
    workerObj.busy = false;
    workerObj.currentTask.resolve(msg);
    this.activeJobs--;

    if (this.queue.length > 0) {
      this._assignTask(workerObj, this.queue.shift());
    }
  }
}

module.exports = WorkerPool;

Использование:

const pool = new WorkerPool('./worker.js', 4);

for (let i =