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

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

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

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

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

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

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

Класс: Worker

• Расширяет: <EventEmitter>

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

Событие: 'disconnect'

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

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

Событие: 'error'

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

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

Событие: 'exit'

• code <число> Код завершения, если завершение было нормальным.
• signal <строка> Название сигнала (например, 'SIGHUP'), который привел к завершению процесса.

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

MJS модули

import cluster from 'cluster';

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('cluster');

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 модули

import cluster from 'cluster';

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

CJS модули

const cluster = require('cluster');

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

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

Событие: 'message'

• message <Объект>
• handle <неопределено> | <Объект>

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

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

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

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

MJS модули

import cluster from 'cluster';
import http from 'http';
import { cpus } from 'os';
import process from '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 = cpus().length;
  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('cluster');
const http = require('http');
const process = require('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('os').cpus().length;
  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
});

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

worker.disconnect()

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

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

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

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

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

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

В процессе работника 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('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
    }
  });
}

worker.exitedAfterDisconnect

• <логическое значение>

Это свойство имеет значение true если работник завершился из-за .kill() или .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();

worker.id

• <число>

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

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

worker.isConnected()

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

worker.isDead()

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

MJS модули

import cluster from 'cluster';
import http from 'http';
import { cpus } from 'os';
import process from 'process';

const numCPUs = cpus().length;

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('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;
const process = require('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. В процессе работника — отключением канала и выходом с кодом 0.

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

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

Этот метод переименован как 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 <Handle>
• 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);
  });
}

Событие: 'disconnect'

• worker <cluster.Worker>

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

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

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

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

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

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

Событие addressType может быть одним из:

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

Событие: 'message'

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

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

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

Событие: 'online'

• worker <cluster.Worker>

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

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

Событие: '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 разветвлённых процессов. Поскольку модуль cluster зависит от IPC для работы, эта настройка должна содержать запись 'ipc'. При указании этой опции она переопределяет silent.
  • 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 '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('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 '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('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 'cluster';

// Go through all workers
function eachWorker(callback) {
  for (const id in cluster.workers) {
    callback(cluster.workers[id]);
  }
}
eachWorker((worker) => {
  worker.send('big announcement to all workers');
});

Модули CJS

const cluster = require('cluster');

// Go through all workers
function eachWorker(callback) {
  for (const id in cluster.workers) {
    callback(cluster.workers[id]);
  }
}
eachWorker((worker) => {
  worker.send('big announcement to all workers');
});

Использование уникального идентификатора рабочего процесса — самый простой способ его найти.

socket.on('data', (id) => {
  const worker = cluster.workers[id];
});