Кластер

Исходный код: lib/cluster.js

Один экземпляр 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

• Расширяет: <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 хранится.

См.: Модуль процесса дочернего процесса.

Рабочие процессы вызовут 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 <число> Код завершения, если процесс завершился нормально.
• 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 (сокет доменной области Unix)
• 'udp4' или 'udp6' (UDP v4 или v6)

Событие: 'message'

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

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

См. 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 <Объект> Пара-значение для добавления в окружение процесса рабочего процесса.
• Возвращает: <cluster.Worker>

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

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

cluster.isMaster

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

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

cluster.isWorker

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

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