Кластер

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

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

В рабочем процессе также можно использовать 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.

Устанавливает .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][, callback])

• message <Объект>
• sendHandle <Дескриптор>
• 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);
  });
}

worker.suicide

Псевдоним для worker.exitedAfterDisconnect.

Устанавливается при вызове .kill() или .disconnect(). До этого момента, значение undefined.

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

cluster.on('exit', (worker, code, signal) => {
  if (worker.suicide === true) {
    console.log('Oh, it was just voluntary – no need to worry');
  }
});

// kill worker
worker.kill();

Этот API существует только для обратной совместимости и будет удален в будущем.

Событие: '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 <число> Код завершения, если выход был нормальным.
• signal <строка> Название сигнала (например, '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 <Объект>

После вызова 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 (сокет доменной области)
• 'udp4' или 'udp6' (UDP v4 или v6)

Событие: 'message'

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

Срабатывает, когда мастер-процесс кластера получает сообщение от любого рабочего процесса.

См. событие 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>

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

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

Событие: 'setup'

• settings <Объект>

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

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

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

cluster.disconnect([callback])

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

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

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

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

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

cluster.fork([env])

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

Создать новый рабочий процесс.

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

cluster.isMaster

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

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

cluster.isWorker

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

Истина, если процесс не является мастером (это отрицание 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 разветвленных процессов. Поскольку модуль кластеризации использует 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];
});