Кластер

Один экземпляр Node.js работает в одном потоке. Для использования многоядерных систем пользователю иногда требуется запустить кластер процессов Node.js для обработки нагрузки.

Модуль cluster позволяет легко создавать дочерние процессы, которые все совместно используют порты сервера.

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('exit', (worker, code, signal) => {
    console.log(`worker ${worker.process.pid} died`);
  });
} else {
  // Workers can share any TCP connection
  // In this case it is an HTTP server
  http.createServer((req, res) => {
    res.writeHead(200);
    res.end('hello world\n');
  }).listen(8000);

  console.log(`Worker ${process.pid} started`);
}

При запуске Node.js теперь порт 8000 будет совместно использоваться рабочими процессами:

$ node server.js
Master 3596 is running
Worker 4324 started
Worker 4520 started
Worker 6056 started
Worker 5644 started

Обратите внимание, что в Windows пока невозможно настроить сервер именованной очереди в рабочем процессе.

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

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

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

Аналогично событию '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

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

Устанавливается при вызове .kill() или .disconnect(). До этого момента он равен 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.

worker.kill([signal='SIGTERM'])

• signal <строка> Название сигнала завершения, который нужно отправить рабочему процессу.

Эта функция завершит рабочий процесс. В мастер-процессе это делается путем отключения 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][, callback])

• message <Object>
• sendHandle <Handle>
• callback <Function>
• Возвращает: <boolean>

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

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

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

До Node.js v6.0 это событие передавало только сообщение и дескриптор, но не объект рабочего процесса, вопреки тому, что было указано в документации.

Если требуется поддержка старых версий, но объект рабочего процесса не требуется, можно обойти эту проблему, проверив количество аргументов:

cluster.on('message', (worker, message, handle) => {
  if (arguments.length === 2) {
    handle = message;
    message = worker;
    worker = undefined;
  }
  // ...
});

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

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

cluster.isWorker

• <boolean>

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

cluster.schedulingPolicy

Политика планирования, либо cluster.SCHED_RR для циклического распределения, либо cluster.SCHED_NONE для переопределения операционной системой. Это глобальная настройка и она фактически замораживается после создания первого рабочего процесса или вызова cluster.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 (унаследован от родительского процесса).
  • silent <Булево> Отправлять ли вывод в stdio родительского процесса. По умолчанию: false.
  • stdio <Массив> Настраивает stdio разветвлённых процессов. Так как модуль cluster полагается на 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];
});