Cluster, worker pool, HTTP proxy

Финал серии: что делать, когда одного процесса Node не хватает. Три разные техники, каждая под свою задачу:

• cluster для горизонтальной нагрузки на одной машине.
• worker_threads для CPU-параллельности внутри процесса.
• HTTP proxy для управления входящим трафиком.

В каждой теме отдельный набор грабель: правильно ли ты решаешь проблему, или просто прячешь её.

1. cluster: pre-fork над одним портом

cluster запускает N child-процессов, делящих один TCP-порт. Технически, master биндит порт через SO_REUSEPORT (или раздаёт file-descriptor), и операционная система балансирует входящие соединения между worker-ами round-robin (по умолчанию).

import cluster from 'node:cluster';
import os from 'node:os';
import http from 'node:http';

if (cluster.isPrimary) {
  const cores = os.availableParallelism();
  for (let i = 0; i < cores; i += 1) cluster.fork();

  cluster.on('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} died (${signal || code})`);
    cluster.fork(); // перезапуск
  });
} else {
  const server = http.createServer((req, res) => {
    res.end(`hi from ${process.pid}\n`);
  });
  server.listen(3000);
}

os.availableParallelism() (Node 19+) даёт реальное число доступных ядер с учётом cgroups (в контейнерах это важно: os.cpus().length возвращает host-овые ядра, не контейнерные).

Что cluster даёт:

• N процессов на N ядер: один Node-процесс держит одно ядро, cluster масштабирует.
• Изоляция: если один worker крашнулся, остальные живут.
• Бесшовные deploy через graceful restart: master запускает новых, старые получают SIGTERM.

Что cluster НЕ даёт:

• Shared memory между процессами (для этого нужен worker_threads или Redis).
• Sticky session для WebSocket (round-robin рвёт длинные соединения, см. § 3).

2. Когда cluster не нужен

В современном проде cluster часто избыточен. Причина: уже есть reverse-proxy (nginx, envoy, kubernetes ingress) и/или несколько инстансов сервиса за load balancer-ом. Они уже балансируют между N процессами/инстансами.

Запуск cluster-а внутри pod-а с одним worker-ом, бессмысленно. Запуск cluster-а в pod-е с 4 worker-ами вместо replicas: 4, размывает изоляцию: в одном pod-е 4 процесса делят CPU/memory лимиты, GC одного влияет на других, OOM убивает всю реплику.

Когда cluster имеет смысл:

• Один большой VPS, без kubernetes/nomad/докер-compose. Cluster это простой способ задействовать все ядра.
• Локальная разработка для нагрузочных тестов.
• Bare-metal в проде, где зацепиться за systemd проще, чем за пол-стека контейнеров.

Когда лучше N инстансов вместо cluster:

• В Kubernetes, Nomad, ECS, любом managed-runtime.
• За load balancer-ом, который уже балансит.
• Когда нужна изоляция падений и memory limit на инстанс.

3. Sticky session: WebSocket и cluster

Round-robin балансировка ломает WebSocket: handshake (HTTP upgrade) попадает в worker-а A, а следующее сообщение, в worker-а B, у которого нет state о соединении.

Решения:

1. Не использовать cluster для WebSocket. На уровне выше (envoy, nginx) делать sticky-балансировку по IP-у/cookie, отдавать тебе один процесс на соединение.
2. Sticky cluster scheduling через хеш IP (модуль @socket.io/sticky для Socket.io, или ручная реализация).

// упрощённая sticky-балансировка
import cluster from 'node:cluster';
import { createHash } from 'node:crypto';
import http from 'node:http';
import os from 'node:os';

if (cluster.isPrimary) {
  const workers = Array.from({ length: os.availableParallelism() }, () => cluster.fork());

  const server = http.createServer();
  server.on('connection', (socket) => {
    const ip = socket.remoteAddress ?? '0.0.0.0';
    const hash = createHash('sha1').update(ip).digest().readUInt32BE(0);
    const worker = workers[hash % workers.length];
    worker.send('sticky-connection', socket);
  });
  server.listen(3000);
}

Это упрощённая схема, реальная боевая sticky-балансировка делается на proxy-уровне.

4. worker_threads: настоящие потоки JS

cluster делает процессы. worker_threads делает потоки внутри процесса. Каждый worker, отдельный V8-isolate с собственным heap, своим event loop.

// worker.ts
import { parentPort } from 'node:worker_threads';
import { scryptSync } from 'node:crypto';

parentPort?.on('message', (input: { password: string; salt: Buffer }) => {
  const result = scryptSync(input.password, input.salt, 64);
  parentPort?.postMessage(result);
});

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

const worker = new Worker(new URL('./worker.ts', import.meta.url));
worker.postMessage({ password: 'hunter2', salt: Buffer.from('...') });
worker.on('message', (result) => {
  console.log('hashed', result);
});

Что отличает worker_threads от cluster:

	cluster	worker_threads
Изоляция	процесс	поток внутри процесса
Память	каждый, свои ~50 MB	один heap на процесс, ~5 MB на worker
Shared memory	нет	через SharedArrayBuffer
Падение	один умер, другие живут	один пытается развалить процесс
Старт	секунды	сотни ms
Когда нужен	scale-out, изоляция	CPU-параллельность внутри запроса

worker_threads идеален для:

• Тяжёлая CPU-операция в одном запросе: hashing, image processing, CSV parsing. Запрос пришёл, отправили в worker, дождались ответа.
• Background-задачи: indexer, periodic computation, без блокировки основного event loop.

worker_threads плохо ложится на:

• Лёгкие, частые операции: накладные на сериализацию сообщений больше, чем сама работа.
• Задачи с большим IPC-объёмом: каждое сообщение копируется (если не передавать через Transferable или SharedArrayBuffer).

5. Worker pool своими руками

Создавать worker под каждую задачу дорого (~50 ms старт). На production-нагрузке нужен пул переиспользуемых worker-ов.

Авторская реализация (~80 строк):

import { Worker } from 'node:worker_threads';
import { EventEmitter } from 'node:events';

type Task<TIn, TOut> = {
  input: TIn;
  resolve: (value: TOut) => void;
  reject: (reason: unknown) => void;
};

export class WorkerPool<TIn, TOut> extends EventEmitter {
  private workers: Worker[] = [];
  private idle: Worker[] = [];
  private queue: Task<TIn, TOut>[] = [];
  private busy = new Map<Worker, Task<TIn, TOut>>();

  constructor(
    private readonly script: URL,
    private readonly size: number,
    private readonly maxQueue = 1024,
  ) {
    super();
    for (let i = 0; i < size; i += 1) {
      this.spawnWorker();
    }
  }

  private spawnWorker(): void {
    const worker = new Worker(this.script);
    worker.on('message', (output: TOut) => {
      const task = this.busy.get(worker);
      if (task === undefined) return;
      this.busy.delete(worker);
      task.resolve(output);
      this.idle.push(worker);
      this.dispatch();
    });
    worker.on('error', (error) => {
      const task = this.busy.get(worker);
      if (task !== undefined) {
        this.busy.delete(worker);
        task.reject(error);
      }
      this.replaceWorker(worker);
    });
    this.workers.push(worker);
    this.idle.push(worker);
  }

  private replaceWorker(dead: Worker): void {
    this.workers = this.workers.filter((w) => w !== dead);
    this.idle = this.idle.filter((w) => w !== dead);
    void dead.terminate();
    this.spawnWorker();
  }

  private dispatch(): void {
    while (this.idle.length > 0 && this.queue.length > 0) {
      const worker = this.idle.shift()!;
      const task = this.queue.shift()!;
      this.busy.set(worker, task);
      worker.postMessage(task.input);
    }
  }

  async run(input: TIn): Promise<TOut> {
    if (this.queue.length >= this.maxQueue) {
      throw new Error(`worker pool queue is full (${this.maxQueue})`);
    }
    return new Promise<TOut>((resolve, reject) => {
      this.queue.push({ input, resolve, reject });
      this.dispatch();
    });
  }

  async close(): Promise<void> {
    await Promise.all(this.workers.map((w) => w.terminate()));
    this.workers = [];
    this.idle = [];
    this.busy.clear();
  }
}

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

const pool = new WorkerPool<{ password: string; salt: Buffer }, Buffer>(
  new URL('./scrypt-worker.ts', import.meta.url),
  4, // 4 воркера
  500, // лимит очереди
);

const hash = await pool.run({ password: 'hunter2', salt: Buffer.from('...') });

Что важно:

• Лимит очереди (maxQueue). Без него под нагрузкой queue растёт до OOM. С лимитом, запрос быстро падает в overload, оркестратор добавит реплику.
• Замена упавшего worker-а. Без replaceWorker пул редеет с каждой ошибкой.
• Чистое завершение. pool.close() зовётся в graceful shutdown.

В реальном проекте проще взять piscina, она делает то же самое плюс дополнительные фичи (idle timeout, abort, transferList). Своя реализация, для понимания механики.

producer

8

/ s

queue

0/16

0 completed0 dropped0 ms avg0 tpsrunning

Подними producer выше суммарной мощности воркеров, и очередь наполняется до maxQueue, после чего новые задачи падают в dropped. Без лимита очередь росла бы до OOM, с лимитом сервис быстро ловит overload, и оркестратор поднимает реплику. Добавь воркеров, очередь снова пустеет, латентность падает.