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

Рабочие процессы порождаются с помощью метода 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'), который привел к завершению процесса.

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

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>

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

См. 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');
}