Координация параллельного выполнения в node.js

Модель программирования node.js, управляемая событиями, несколько усложняет координацию выполнения программы.

Простое последовательное выполнение превращается во вложенные обратные вызовы, что достаточно легко (хотя и немного запутано для записи).

Но как насчет параллельного выполнения? Скажем, у вас есть три задачи A, B, C, которые могут выполняться параллельно, и когда они будут выполнены, вы хотите отправить их результаты в задачу D.

С моделью вилки / соединения это будет

• вилка А
• вилка B
• вилка C
• присоединиться к A, B, C, запустить D

Как мне написать это в node.js? Есть ли какие-нибудь передовые практики или кулинарные книги? Должен ли я каждый раз подбирать решение вручную или для этого есть какая-нибудь библиотека с помощниками?

В node.js нет ничего по-настоящему параллельного, поскольку он однопоточный. Однако можно запланировать и запустить несколько событий в последовательности, которую невозможно определить заранее. А некоторые вещи, такие как доступ к базе данных, на самом деле «параллельны» в том смысле, что сами запросы к базе данных выполняются в отдельных потоках, но по завершении повторно интегрируются в поток событий.

Итак, как запланировать обратный вызов для нескольких обработчиков событий? Что ж, это один из распространенных методов, используемых в анимации в javascript на стороне браузера: использование переменной для отслеживания завершения.

Это звучит как взлом, и это так, и это звучит потенциально беспорядочно, оставляя кучу глобальных переменных для отслеживания, и на меньшем языке это было бы. Но в javascript мы можем использовать замыкания:

function fork (async_calls, shared_callback) {
  var counter = async_calls.length;
  var callback = function () {
    counter --;
    if (counter == 0) {
      shared_callback()
    }
  }

  for (var i=0;i<async_calls.length;i++) {
    async_calls[i](callback);
  }
}

// usage:
fork([A,B,C],D);

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

Дополнительный ответ:

На самом деле, даже как есть, эта fork()функция уже может передавать аргументы асинхронным функциям, используя замыкание:

fork([
  function(callback){ A(1,2,callback) },
  function(callback){ B(1,callback) },
  function(callback){ C(1,2,callback) }
],D);

осталось только собрать результаты из A, B, C и передать их D.

Еще более дополнительный ответ:

Я не мог устоять. Продолжал думать об этом во время завтрака. Вот реализация, fork()которая накапливает результаты (обычно переданные в качестве аргументов функции обратного вызова):

function fork (async_calls, shared_callback) {
  var counter = async_calls.length;
  var all_results = [];
  function makeCallback (index) {
    return function () {
      counter --;
      var results = [];
      // we use the arguments object here because some callbacks 
      // in Node pass in multiple arguments as result.
      for (var i=0;i<arguments.length;i++) {
        results.push(arguments[i]);
      }
      all_results[index] = results;
      if (counter == 0) {
        shared_callback(all_results);
      }
    }
  }

  for (var i=0;i<async_calls.length;i++) {
    async_calls[i](makeCallback(i));
  }
}

Это было достаточно просто. Это имеет fork()довольно общее назначение и может использоваться для синхронизации нескольких неоднородных событий.

Пример использования в Node.js:

// Read 3 files in parallel and process them together:

function A (c){ fs.readFile('file1',c) };
function B (c){ fs.readFile('file2',c) };
function C (c){ fs.readFile('file3',c) };
function D (result) {
  file1data = result[0][1];
  file2data = result[1][1];
  file3data = result[2][1];

  // process the files together here
}

fork([A,B,C],D);

Обновить

Этот код был написан до появления таких библиотек, как async.js или различных библиотек, основанных на обещаниях. Я хотел бы верить, что async.js был вдохновлен этим, но у меня нет никаких доказательств этого. В любом случае ... если вы думаете об этом сегодня, взгляните на async.js или обещания. Просто рассмотрите ответ выше, хорошее объяснение / иллюстрацию того, как работают такие вещи, как async.parallel.

Для полноты картины вот как это сделать async.parallel:

var async = require('async');

async.parallel([A,B,C],D);

Обратите внимание, что это async.parallelработает точно так же, как forkреализованная выше функция. Основное отличие заключается в том, что он передает ошибку в качестве первого аргумента, Dа обратный вызов - в качестве второго аргумента в соответствии с соглашением node.js.

Используя обещания, мы бы написали это так:

// Assuming A, B & C return a promise instead of accepting a callback

Promise.all([A,B,C]).then(D);

Я считаю, что теперь модуль «async» обеспечивает эту параллельную функциональность и примерно такой же, как и функция fork выше.

В модуле Futures есть подмодуль join, который мне нравится использовать:

Объединяет асинхронные вызовы вместе аналогично тому, как pthread_joinработает для потоков.

Readme показывает несколько хороших примеров использования его вольным стилем или использования будущего подмодуля с использованием шаблона Promise. Пример из документов:

var Join = require('join')
  , join = Join()
  , callbackA = join.add()
  , callbackB = join.add()
  , callbackC = join.add();

function abcComplete(aArgs, bArgs, cArgs) {
  console.log(aArgs[1] + bArgs[1] + cArgs[1]);
}

setTimeout(function () {
  callbackA(null, 'Hello');
}, 300);

setTimeout(function () {
  callbackB(null, 'World');
}, 500);

setTimeout(function () {
  callbackC(null, '!');
}, 400);

// this must be called after all 
join.when(abcComplete);

Здесь возможно простое решение: http://howtonode.org/control-flow-part-ii перейдите к Параллельные действия. Другим способом было бы, чтобы все A, B и C использовали одну и ту же функцию обратного вызова, чтобы эта функция имела глобальный или, по крайней мере, инкрементор вне функции, если все три вызвали обратный вызов, тогда позвольте ему запустить D, конечно, вам также придется где-то хранить результаты A, B и C.

Другим вариантом может быть модуль Step для узла: https://github.com/creationix/step

Вы можете попробовать эту крошечную библиотеку: https://www.npmjs.com/package/parallel-io

Помимо популярных обещаний и async-библиотеки, есть 3-й элегантный способ - с помощью "wiring":

var l = new Wire();

funcA(l.branch('post'));
funcB(l.branch('comments'));
funcC(l.branch('links'));

l.success(function(results) {
   // result will be object with results:
   // { post: ..., comments: ..., links: ...}
});

https://github.com/garmoshka-mo/mo-wire