Кластер

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

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

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

if (cluster.isMaster) {
  // Fork workers.
  for (var 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);
}

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

$ NODE_DEBUG=cluster node server.js
23521,Master Worker 23524 online
23521,Master Worker 23526 online
23521,Master Worker 23523 online
23521,Master Worker 23528 online

Обратите внимание, что в Windows пока нельзя настроить сервер именованной трубы в рабочем процессе.

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

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

Класс: Рабочий процесс

Объект 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 <неопределено> | <Объект>

Аналогично событию cluster.on('message'), но специфично для данного рабочего процесса. В рабочем процессе также можно использовать process.on('message').

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

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

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

if (cluster.isMaster) {

  // Keep track of http requests
  var 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 (var i = 0; i < numCPUs; i++) {
    cluster.fork();
  }

  Object.keys(cluster.workers).forEach((id) => {
    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()

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

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

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

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

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

Обратите внимание, что в рабочем процессе process.disconnect существует, но это не эта функция, это disconnect.

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

if (cluster.isMaster) {
  var worker = cluster.fork();
  var 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');
  var 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.id

• <Число>

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

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

worker.isConnected()

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

worker.isDead()

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

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

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

Эта функция убьёт рабочий процесс. В мастер-процессе это делается путём отключения worker.process, а затем убийством с помощью signal. В рабочем процессе это делается путём отключения канала и затем завершения с кодом 0.

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

Этот метод переименован в worker.destroy() для совместимости со старыми версиями.

Обратите внимание, что в рабочем процессе process.kill() существует, но это не эта функция, это kill.

worker.process

• <ChildProcess>

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

См.: Модуль Child Process

Обратите внимание, что рабочие процессы вызовут process.exit(0) если событие 'disconnect' произойдёт на process и .suicide не true. Это предотвращает случайное отключение.

worker.send(message[, sendHandle][, callback])

• message <Объект>
• sendHandle <Дескриптор>
• callback <Функция>
• Возвращает: Булево значение

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

В мастер-процессе это отправляет сообщение конкретному рабочему процессу. Это идентично ChildProcess.send().

В рабочем процессе это отправляет сообщение мастер-процессу. Оно идентично process.send().

Этот пример будет эхо-отражать все сообщения от мастер-процесса:

if (cluster.isMaster) {
  var worker = cluster.fork();
  worker.send('hi there');

} else if (cluster.isWorker) {
  process.on('message', (msg) => {
    process.send(msg);
  });
}

worker.suicide

• <Булево>

Устанавливается вызовом .kill() или .disconnect(), до тех пор, пока не будет установлено другое значение undefined.

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

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

// kill worker
worker.kill();

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

var 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'

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

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

Инициируется каждый раз, когда вызывается .setupMaster().

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

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

cluster.disconnect([callback])

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

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

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

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

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

cluster.fork([env])

• env <Объект> Пара пары ключ/значение для добавления в среду рабочего процесса.
• return <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))
  • silent <Булево> отправлять ли вывод в stdio родительского процесса. (По умолчанию=false)
  • uid <Число> Устанавливает идентификатор пользователя процесса. (См. setuid(2).)
  • gid <Число> Устанавливает идентификатор группы процесса. (См. setgid(2).)

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

Не предполагается, что вы будете изменять или устанавливать этот объект вручную.

cluster.setupMaster([settings])

• settings <Объект>
  • exec <Строка> путь к файлу рабочего процесса. (По умолчанию=process.argv[1])
  • args <Массив> строковые аргументы, переданные рабочему процессу. (По умолчанию=process.argv.slice(2))
  • silent <Булево> отправлять ли вывод в stdio родительского процесса. (По умолчанию=false)

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 (var id in cluster.workers) {
    callback(cluster.workers[id]);
  }
}
eachWorker((worker) => {
  worker.send('big announcement to all workers');
});

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

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