Как это работает

Рабочие процессы создаются с помощью метода child_process.fork(), благодаря чему они могут взаимодействовать с родительским процессом через IPC и передавать дескрипторы сервера в обоих направлениях.

Модуль cluster поддерживает два способа распределения входящих подключений.

Первый способ (и используемый по умолчанию на всех платформах, кроме Windows) — это циклическое распределение: основной процесс прослушивает порт, принимает новые подключения и распределяет их между рабочими процессами по очереди, используя встроенные механизмы, предотвращающие перегрузку рабочего процесса.

При втором способе основной процесс создает сокет прослушивания и передает его заинтересованным рабочим процессам. Затем рабочие процессы принимают входящие подключения напрямую.

Теоретически второй способ должен обеспечивать наилучшую производительность. Однако на практике распределение часто оказывается очень неравномерным из-за особенностей планировщика операционной системы. Наблюдались ситуации, когда более 70% всех подключений приходилось всего на два процесса из восьми.

Поскольку server.listen() передает большую часть работы основному процессу, есть три случая, когда поведение обычного процесса Node.js и рабочего процесса кластера различается:

1. server.listen({fd: 7}) Поскольку сообщение передается основному процессу, прослушиваться будет файловый дескриптор 7 в родительском процессе, а его дескриптор будет передан рабочему процессу, вместо того чтобы прослушивать то, на что, по мнению рабочего процесса, ссылается файловый дескриптор с номером 7.
2. server.listen(handle) Явное прослушивание дескрипторов приведет к тому, что рабочий процесс будет использовать переданный дескриптор, а не обращаться к основному процессу.
3. server.listen(0) Обычно это приводит к тому, что серверы прослушивают случайный порт. Однако в кластере каждый рабочий процесс будет получать один и тот же «случайный» порт при каждом вызове listen(0). По сути, порт будет случайным при первом вызове, но предсказуемым при последующих. Чтобы прослушивать уникальный порт, сгенерируйте номер порта на основе идентификатора рабочего процесса кластера.

Node.js не предоставляет логику маршрутизации. Поэтому важно спроектировать приложение так, чтобы оно не слишком зависело от объектов данных в памяти для таких задач, как управление сеансами и вход в систему.

Поскольку рабочие процессы являются отдельными процессами, их можно завершать или создавать заново в соответствии с потребностями программы, не затрагивая другие рабочие процессы. Пока остаются работающие рабочие процессы, сервер будет продолжать принимать подключения. Если работающих рабочих процессов не останется, существующие подключения будут разорваны, а новые — отклонены. Однако Node.js не управляет количеством рабочих процессов автоматически. Приложение должно самостоятельно управлять пулом рабочих процессов в соответствии со своими потребностями.

Хотя основным сценарием использования модуля node:cluster является работа с сетью, его можно применять и в других случаях, требующих рабочих процессов.

Класс: Worker

• Наследует: <EventEmitter>

Объект Worker содержит общедоступную информацию и методы рабочего процесса. В основном процессе его можно получить с помощью cluster.workers. В рабочем процессе его можно получить с помощью cluster.worker.

Событие: 'disconnect'

Аналогично событию cluster.on('disconnect'), но относится к этому рабочему процессу.

cluster.fork().on('disconnect', () => {
  // Worker has disconnected
}); copy

Событие: 'error'

Это событие совпадает с событием, предоставляемым child_process.fork().

В рабочем процессе также можно использовать process.on('error').

Событие: 'exit'

• code <number> Код завершения, если процесс завершился штатно.
• signal <string> Имя сигнала (например, 'SIGHUP'), вызвавшего завершение процесса.

Аналогично событию cluster.on('exit'), но относится к этому рабочему процессу.

import cluster from 'node:cluster';

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.on('exit', (code, signal) => {
    if (signal) {
      console.log(`worker was killed by signal: ${signal}`);
    } else if (code !== 0) {
      console.log(`worker exited with error code: ${code}`);
    } else {
      console.log('worker success!');
    }
  });
}

const cluster = require('node:cluster');

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.on('exit', (code, signal) => {
    if (signal) {
      console.log(`worker was killed by signal: ${signal}`);
    } else if (code !== 0) {
      console.log(`worker exited with error code: ${code}`);
    } else {
      console.log('worker success!');
    }
  });
}

Событие: 'listening'

• address <Object>

Аналогично событию cluster.on('listening'), но относится к этому рабочему процессу.

cluster.fork().on('listening', (address) => {
  // Worker is listening
});

cluster.fork().on('listening', (address) => {
  // Worker is listening
});

В рабочем процессе это событие не генерируется.

Событие: 'message'

• message <Object>
• handle <undefined> | <Object>

Аналогично событию 'message' объекта cluster, но относится к этому рабочему процессу.

В рабочем процессе также можно использовать process.on('message').

См. событие process: 'message'.

Ниже приведен пример использования системы сообщений. В нем основной процесс подсчитывает количество HTTP-запросов, полученных рабочими процессами:

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

if (cluster.isPrimary) {

  // Keep track of http requests
  let numReqs = 0;
  setInterval(() => {
    console.log(`numReqs = ${numReqs}`);
  }, 1000);

  // Count requests
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // Start workers and listen for messages containing notifyRequest
  const numCPUs = availableParallelism();
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  for (const id in cluster.workers) {
    cluster.workers[id].on('message', messageHandler);
  }

} else {

  // Worker processes have a http server.
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');

    // Notify primary about the request
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}

const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').availableParallelism();
const process = require('node:process');

if (cluster.isPrimary) {

  // Keep track of http requests
  let numReqs = 0;
  setInterval(() => {
    console.log(`numReqs = ${numReqs}`);
  }, 1000);

  // Count requests
  function messageHandler(msg) {
    if (msg.cmd && msg.cmd === 'notifyRequest') {
      numReqs += 1;
    }
  }

  // Start workers and listen for messages containing notifyRequest
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  for (const id in cluster.workers) {
    cluster.workers[id].on('message', messageHandler);
  }

} else {

  // Worker processes have a http server.
  http.Server((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');

    // Notify primary about the request
    process.send({ cmd: 'notifyRequest' });
  }).listen(8000);
}

Событие: 'online'

Аналогично событию cluster.on('online'), но относится к этому рабочему процессу.

cluster.fork().on('online', () => {
  // Worker is online
}); copy

В рабочем процессе это событие не генерируется.

worker.disconnect()

• Возвращает: <cluster.Worker> Ссылку на worker.

В рабочем процессе эта функция закроет все серверы, дождется события 'close' на этих серверах, а затем отключит канал IPC.

В основном процессе рабочему процессу отправляется внутреннее сообщение, которое заставляет его вызвать .disconnect() для самого себя.

Устанавливает значение .exitedAfterDisconnect.

После закрытия сервера он перестанет принимать новые подключения, но их смогут принимать любые другие прослушивающие рабочие процессы. Существующим подключениям будет позволено закрыться обычным образом. Когда подключений не останется, см. server.close(), канал IPC рабочего процесса закроется, что позволит ему завершиться корректно.

Описанное выше относится только к подключениям сервера. Рабочие процессы не закрывают клиентские подключения автоматически, а отключение не ожидает их закрытия перед завершением.

В рабочем процессе существует process.disconnect, но это не данная функция, а disconnect().

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

if (cluster.isPrimary) {
  const worker = cluster.fork();
  let timeout;

  worker.on('listening', (address) => {
    worker.send('shutdown');
    worker.disconnect();
    timeout = setTimeout(() => {
      worker.kill();
    }, 2000);
  });

  worker.on('disconnect', () => {
    clearTimeout(timeout);
  });

} else if (cluster.isWorker) {
  const net = require('node:net');
  const server = net.createServer((socket) => {
    // Connections never end
  });

  server.listen(8000);

  process.on('message', (msg) => {
    if (msg === 'shutdown') {
      // Initiate graceful close of any connections to server
    }
  });
} copy

worker.exitedAfterDisconnect

• <boolean>

Это свойство равно true, если рабочий процесс завершился из-за .disconnect(). Если рабочий процесс завершился по любой другой причине, свойство равно false. Если рабочий процесс еще не завершился, свойство равно undefined.

Логическое свойство worker.exitedAfterDisconnect позволяет отличить добровольное завершение от случайного; основное приложение может решить не создавать рабочий процесс заново, основываясь на этом значении.

cluster.on('exit', (worker, code, signal) => {
  if (worker.exitedAfterDisconnect === true) {
    console.log('Oh, it was just voluntary – no need to worry');
  }
});

// kill worker
worker.kill(); copy

worker.id

• <integer>

Каждому новому рабочему процессу присваивается уникальный идентификатор, который хранится в id.

Пока рабочий процесс работает, этот идентификатор используется в качестве ключа для него в cluster.workers.

worker.isConnected()

Эта функция возвращает true, если рабочий процесс подключен к основному процессу через канал IPC, и false в противном случае. Рабочий процесс подключается к основному процессу после создания. Отключение происходит после генерации события 'disconnect'.

worker.isDead()

Эта функция возвращает true, если процесс рабочего процесса завершился (в результате выхода или получения сигнала). В противном случае она возвращает false.

import cluster from 'node:cluster';
import http from 'node:http';
import { availableParallelism } from 'node:os';
import process from 'node:process';

const numCPUs = availableParallelism();

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

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

  cluster.on('fork', (worker) => {
    console.log('worker is dead:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker is dead:', worker.isDead());
  });
} else {
  // Workers can share any TCP connection. In this case, it is an HTTP server.
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`Current process\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}

const cluster = require('node:cluster');
const http = require('node:http');
const numCPUs = require('node:os').availableParallelism();
const process = require('node:process');

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

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

  cluster.on('fork', (worker) => {
    console.log('worker is dead:', worker.isDead());
  });

  cluster.on('exit', (worker, code, signal) => {
    console.log('worker is dead:', worker.isDead());
  });
} else {
  // Workers can share any TCP connection. In this case, it is an HTTP server.
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end(`Current process\n ${process.pid}`);
    process.kill(process.pid);
  }).listen(8000);
}

worker.kill([signal])

• signal <string> Имя сигнала завершения, отправляемого процессу рабочего процесса. По умолчанию: 'SIGTERM'

Эта функция завершает рабочий процесс. В основном процессе она отключает worker.process, а после отключения завершает процесс с помощью signal. В рабочем процессе она завершает процесс с помощью signal.

Функция kill() завершает процесс рабочего процесса, не дожидаясь корректного отключения; ее поведение совпадает с поведением worker.process.kill().

Для обратной совместимости этот метод также доступен под именем worker.destroy().

В рабочем процессе существует process.kill(), но это не данная функция, а kill().

worker.process

• <ChildProcess>

Все рабочие процессы создаются с помощью child_process.fork(); объект, возвращенный этой функцией, сохраняется в .process. В рабочем процессе сохраняется глобальный объект process.

См.: модуль дочерних процессов.

Рабочие процессы вызывают process.exit(0), если для process происходит событие 'disconnect', а .exitedAfterDisconnect не равно true. Это защищает от случайного отключения.

worker.send(message[, sendHandle[, options]][, callback])

• message <Object>
• sendHandle <Handle>
• options <Object> Аргумент options, если он присутствует, представляет собой объект, используемый для настройки отправки определенных типов дескрипторов. options поддерживает следующие свойства:
  • keepOpen <boolean> Значение, которое можно использовать при передаче экземпляров net.Socket. Если true, сокет остается открытым в отправляющем процессе. По умолчанию: false.
• callback <Function>
• Возвращает: <boolean>

Отправляет сообщение рабочему или основному процессу, при необходимости вместе с дескриптором.

В основном процессе отправляет сообщение указанному рабочему процессу. Функция идентична ChildProcess.send().

В рабочем процессе отправляет сообщение основному процессу. Функция идентична process.send().

В этом примере все сообщения от основного процесса отправляются обратно ему:

if (cluster.isPrimary) {
  const worker = cluster.fork();
  worker.send('hi there');

} else if (cluster.isWorker) {
  process.on('message', (msg) => {
    process.send(msg);
  });
} copy

Событие: 'disconnect'

• worker <cluster.Worker>

Генерируется после отключения канала IPC рабочего процесса. Это может произойти при корректном завершении рабочего процесса, его принудительном завершении или ручном отключении (например, с помощью worker.disconnect()).

Между событиями 'disconnect' и 'exit' может пройти некоторое время. Эти события можно использовать, чтобы определить, завис ли процесс при очистке или существуют ли длительные подключения.

cluster.on('disconnect', (worker) => {
  console.log(`The worker #${worker.id} has disconnected`);
}); copy

Событие: 'exit'

• worker <cluster.Worker>
• code <number> Код завершения, если процесс завершился штатно.
• signal <string> Имя сигнала (например, 'SIGHUP'), вызвавшего завершение процесса.

При завершении любого рабочего процесса модуль cluster генерирует событие 'exit'.

Это событие можно использовать для перезапуска рабочего процесса, повторно вызвав .fork().

cluster.on('exit', (worker, code, signal) => {
  console.log('worker %d died (%s). restarting...',
              worker.process.pid, signal || code);
  cluster.fork();
}); copy

См. событие child_process: 'exit'.

Событие: 'fork'

• worker <cluster.Worker>

При создании нового рабочего процесса модуль cluster генерирует событие 'fork'. Это можно использовать для регистрации активности рабочего процесса и создания пользовательского тайм-аута.

const timeouts = [];
function errorMsg() {
  console.error('Something must be wrong with the connection ...');
}

cluster.on('fork', (worker) => {
  timeouts[worker.id] = setTimeout(errorMsg, 2000);
});
cluster.on('listening', (worker, address) => {
  clearTimeout(timeouts[worker.id]);
});
cluster.on('exit', (worker, code, signal) => {
  clearTimeout(timeouts[worker.id]);
  errorMsg();
}); copy

Событие: 'listening'

• worker <cluster.Worker>
• address <Object>

После вызова listen() из рабочего процесса, когда на сервере генерируется событие 'listening', в основном процессе на cluster также генерируется событие 'listening'.

Обработчик события вызывается с двумя аргументами: worker содержит объект рабочего процесса, а объект address содержит следующие свойства подключения: address, port и addressType. Это очень полезно, если рабочий процесс прослушивает несколько адресов.

cluster.on('listening', (worker, address) => {
  console.log(
    `A worker is now connected to ${address.address}:${address.port}`);
}); copy

Значение addressType может быть одним из следующих:

• 4 (TCPv4)
• 6 (TCPv6)
• -1 (сокет домена Unix)
• 'udp4' или 'udp6' (UDPv4 или UDPv6)

Событие: 'message'

• worker <cluster.Worker>
• message <Object>
• handle <undefined> | <Object>

Генерируется, когда основной процесс кластера получает сообщение от любого рабочего процесса.

См. событие child_process: 'message'.

Событие: 'online'

• worker <cluster.Worker>

После создания нового рабочего процесса он должен отправить сообщение о переходе в состояние онлайн. Когда основной процесс получает такое сообщение, он генерирует это событие. Разница между 'fork' и 'online' заключается в том, что событие fork генерируется при создании рабочего процесса основным процессом, а 'online' — когда рабочий процесс уже запущен.

cluster.on('online', (worker) => {
  console.log('Yay, the worker responded after it was forked');
}); copy

Событие: 'setup'

• settings <Object>

Генерируется каждый раз при вызове .setupPrimary().

Объект settings — это объект cluster.settings на момент вызова .setupPrimary(); он носит рекомендательный характер, поскольку в рамках одного цикла событий может быть выполнено несколько вызовов .setupPrimary().

Если важна точность, используйте cluster.settings.

cluster.disconnect([callback])

• callback <Function> Вызывается, когда все рабочие процессы отключены и дескрипторы закрыты.

Вызывает .disconnect() для каждого рабочего процесса в cluster.workers.

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

Метод принимает необязательный аргумент обратного вызова, который будет вызван по завершении.

Этот метод можно вызывать только из основного процесса.

cluster.fork([env])

• env <Object> Пары ключ/значение, добавляемые в окружение процесса рабочего процесса.
• Возвращает: <cluster.Worker>

Создает новый процесс рабочего процесса.

Этот метод можно вызывать только из основного процесса.

cluster.isMaster

Устаревший псевдоним для cluster.isPrimary.

cluster.isPrimary

• <boolean>

Значение равно true, если процесс является основным. Это определяется переменной process.env.NODE_UNIQUE_ID. Если process.env.NODE_UNIQUE_ID не определена, то isPrimary равно true.

cluster.isWorker

• <boolean>

Значение равно true, если процесс не является основным (то есть противоположно cluster.isPrimary).

cluster.schedulingPolicy

Политика планирования: cluster.SCHED_RR для циклического распределения или cluster.SCHED_NONE, чтобы передать управление операционной системе. Это глобальная настройка, которая фактически фиксируется после создания первого рабочего процесса или вызова .setupPrimary() — в зависимости от того, что произойдет раньше.

SCHED_RR используется по умолчанию во всех операционных системах, кроме Windows. В Windows значение изменится на SCHED_RR, когда libuv сможет эффективно распределять дескрипторы IOCP без значительного снижения производительности.

Значение cluster.schedulingPolicy также можно задать с помощью переменной окружения NODE_CLUSTER_SCHED_POLICY. Допустимые значения: 'rr' и 'none'.

cluster.settings

• <Object>
  • execArgv <string[]> Список строковых аргументов, передаваемых исполняемому файлу Node.js. По умолчанию: process.execArgv.
  • exec <string> Путь к файлу рабочего процесса. По умолчанию: process.argv[1].
  • args <string[]> Строковые аргументы, передаваемые рабочему процессу. По умолчанию: process.argv.slice(2).
  • cwd <string> Текущий рабочий каталог процесса рабочего процесса. По умолчанию: undefined (наследуется от родительского процесса).
  • serialization <string> Задает тип сериализации для отправки сообщений между процессами. Возможные значения: 'json' и 'advanced'. Подробнее см. в разделе «Расширенная сериализация для child_process». По умолчанию: false.
  • silent <boolean> Определяет, следует ли отправлять вывод в стандартные потоки ввода-вывода родительского процесса. По умолчанию: false.
  • stdio <Array> Настраивает стандартные потоки ввода-вывода создаваемых процессов. Поскольку для работы модуля cluster требуется IPC, эта конфигурация должна содержать элемент 'ipc'. Если задан этот параметр, он имеет приоритет над silent. См. описание параметра stdio метода child_process.spawn().
  • uid <number> Задает идентификатор пользователя процесса. (См. setuid(2).)
  • gid <number> Задает идентификатор группы процесса. (См. setgid(2).)
  • inspectPort <number> | <Function> Задает порт инспектора рабочего процесса. Это может быть число или функция без аргументов, возвращающая число. По умолчанию каждый рабочий процесс получает собственный порт, начиная с process.debugPort основного процесса и увеличивая его.
  • windowsHide <boolean> Скрывает окно консоли создаваемых процессов, которое обычно отображается в системах Windows. По умолчанию: false.

После вызова .setupPrimary() (или .fork()) этот объект настроек будет содержать параметры, включая значения по умолчанию.

Не следует изменять или задавать этот объект вручную.

cluster.setupMaster([settings])

Устаревший псевдоним для .setupPrimary().

cluster.setupPrimary([settings])

• settings <Object> См. cluster.settings.

setupPrimary используется для изменения поведения «fork» по умолчанию. После вызова настройки будут доступны в cluster.settings.

Изменения настроек влияют только на последующие вызовы .fork() и не влияют на уже работающие рабочие процессы.

Единственный атрибут рабочего процесса, который нельзя задать через .setupPrimary(), — это env, передаваемый в .fork().

Указанные выше значения по умолчанию применяются только к первому вызову; значениями по умолчанию для последующих вызовов становятся текущие значения на момент вызова cluster.setupPrimary().

import cluster from 'node:cluster';

cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true,
});
cluster.fork(); // https worker
cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'http'],
});
cluster.fork(); // http worker

const cluster = require('node:cluster');

cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'https'],
  silent: true,
});
cluster.fork(); // https worker
cluster.setupPrimary({
  exec: 'worker.js',
  args: ['--use', 'http'],
});
cluster.fork(); // http worker

Эту функцию можно вызывать только из основного процесса.

cluster.worker

• <Object>

Ссылка на объект текущего рабочего процесса. Недоступно в основном процессе.

import cluster from 'node:cluster';

if (cluster.isPrimary) {
  console.log('I am primary');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`I am worker #${cluster.worker.id}`);
}

const cluster = require('node:cluster');

if (cluster.isPrimary) {
  console.log('I am primary');
  cluster.fork();
  cluster.fork();
} else if (cluster.isWorker) {
  console.log(`I am worker #${cluster.worker.id}`);
}

cluster.workers

• <Object>

Хеш, в котором хранятся активные объекты рабочих процессов, индексированные по полю id. Это упрощает перебор всех рабочих процессов. Доступно только в основном процессе.

Рабочий процесс удаляется из cluster.workers после отключения и завершения. Заранее определить порядок этих двух событий невозможно. Однако гарантируется, что удаление из списка cluster.workers происходит до отправки последнего события 'disconnect' или 'exit'.

import cluster from 'node:cluster';

for (const worker of Object.values(cluster.workers)) {
  worker.send('big announcement to all workers');
}

const cluster = require('node:cluster');

for (const worker of Object.values(cluster.workers)) {
  worker.send('big announcement to all workers');
}