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

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

Хотя основное применение модуля 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'), но специфично для данного работника.

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'), но специфично для данного работника.

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

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

Событие: 'message'

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

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

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

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

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

const cluster = require('cluster');
const http = require('http');

if (cluster.isMaster) {

  // 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 master 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.isMaster) {
  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.

const cluster = require('cluster');
const http = require('http');
const numCPUs = require('os').cpus().length;

if (cluster.isMaster) {
  console.log(`Master ${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 хранит этот объект.

См.: Модуль Child 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.isMaster) {
  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'), вызвавшего завершение процесса.

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

Когда создаётся новый рабочий процесс, модуль 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();
});

Событие: 'listening'

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

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

Обработчик событий выполняется с двумя аргументами: 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>

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

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

Событие: 'setup'

• settings <Object>

Вызывается каждый раз, когда вызывается .setupMaster().

Объект settings — это объект cluster.settings на момент вызова .setupMaster() и используется только для справок, так как вызовы .setupMaster() могут выполняться в одном тике.

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

cluster.disconnect([callback])

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

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

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

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

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

cluster.fork([env])

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

Запустить новый рабочий процесс.

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

cluster.isMaster

• <boolean>

True, если процесс — мастер. Это определяется по process.env.NODE_UNIQUE_ID. Если process.env.NODE_UNIQUE_ID undefined, то isMaster — true.

cluster.isWorker

• <boolean>

True, если процесс не мастер (это отрицание cluster.isMaster).

cluster.schedulingPolicy

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

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.
  • uid <число> Устанавливает идентификатор пользователя процесса. (См. setuid(2).)
  • gid <число> Устанавливает идентификатор группы процесса. (См. setgid(2).)
  • inspectPort <число> | <Функция> Устанавливает порт инспектора рабочего процесса. Может быть числом или функцией без аргументов, возвращающей число. По умолчанию каждый рабочий процесс получает свой собственный порт, наращиваемый от порта мастера process.debugPort.
  • windowsHide <логическое значение> Скрыть окно консоли разветвлённых процессов, которое обычно создаётся в системах Windows. По умолчанию: false.

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

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

cluster.setupMaster([settings])

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

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

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

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

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

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

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

cluster.worker

• <Объект>

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

const cluster = require('cluster');

if (cluster.isMaster) {
  console.log('I am master');
  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'.

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