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

Рабочие процессы запускаются с помощью метода 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 <число> Код выхода, если выход нормальный.
• signal <строка> Название сигнала (например, 'SIGHUP'), который привел к завершению процесса.

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

MJS модули

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!');
    }
  });
}

CJS модули

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 <Объект>

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

MJS модули

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

CJS модули

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

Не испускается в рабочем процессе.

Событие: 'message'

• message <Объект>
• handle <undefined> | <Объект>

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

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

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

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

MJS модули

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);
}

CJS модули

const cluster = require('node:cluster');
const http = require('node:http');
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
  const numCPUs = require('node:os').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);
}

Событие: '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

• <логическое>

Это свойство 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

• <целое>

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

Пока работник жив, это ключ, который индексирует его в cluster.workers.

worker.isConnected()

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

worker.isDead()

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

MJS модули

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);
}

CJS модули

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 <строка> Имя сигнала завершения для отправки процессу работника. По умолчанию: '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) , если событие 'disconnect' произойдёт на process и .exitedAfterDisconnect не true. Это защищает от случайного отключения.

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

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

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

В основном процессе это отправляет сообщение определенному работнику. Это идентично 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>

Когда новый работник разветвляется (fork), модуль 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' генерируется на сервере, событие '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>

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

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

Событие: 'online'

• worker <cluster.Worker>

После создания нового работника, работник должен отправить сообщение о готовности. Когда основной процесс получает это сообщение, он генерирует это событие. Разница между 'fork' и 'online' заключается в том, что 'online' генерируется, когда основной процесс создаёт работника, а '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>

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

cluster.isWorker

• <boolean>

Истина, если процесс не является основным (это отрицание 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

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

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

Этот объект не предназначен для изменения или ручного задания.

cluster.setupMaster([settings])

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

cluster.setupPrimary([settings])

• settings <Объект> См. cluster.settings.

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

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

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

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

Модули MJS

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

Модули CJS

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

• <Объект>

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

Модули MJS

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}`);
}

Модули CJS

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

• <Объект>

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

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

Модули MJS

import cluster from 'node:cluster';

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

Модули CJS

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

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