Кластер

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

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

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

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

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'), который привёл к завершению процесса.

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

После вызова 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'

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

См. событие 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' заключается в том, что '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.

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

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

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

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)
  • silent <логическое> указывает, следует ли отправлять вывод в stdio родительского процесса. По умолчанию: false
  • stdio <Массив> Настраивает stdio разветвлённых процессов. Так как модуль cluster полагается на IPC для работы, эта настройка должна содержать 'ipc' запись. При указании этого параметра, он переопределяет silent.
  • uid <число> Устанавливает идентификатор пользователя процесса. (см. setuid(2))
  • gid <число> Устанавливает идентификатор группы процесса. (см. setgid(2))

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

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

cluster.setupMaster([settings])

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

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