В async есть parallelLimit, но тут писать всего несколько строк кода

общая очередь делается примерно так:

// последовательная обработка очереди
function runWorker(func, getNext) {
	return new Promise(function (resolve) {
		(function runItem() {
			var item = getNext();
			if (item == null) {
				resolve();
			} else {
				func(item).then(runItem, runItem);
			}
		})();
	});
}

// использование

// функция, которая что-то делает асинхронно
var func = function(a) {
	var w = this;
    console.log('worker ' + w + ': start ' + a);
    return new Promise(function(r) {
         setTimeout(function() { console.log('worker ' + w + ': end ', a); r(); }, a * 1000);
    });
};

// очередь параметров
var data = [4, 3, 8, 2, 5, 6, 12];
var getNext = data.shift.bind(data);

console.log('queue: ' + data.join(' '));

// два параллельных потока
Promise.all([
    runWorker(func.bind(0), getNext),
    runWorker(func.bind(1), getNext)
]).then(console.log.bind(console, 'end all'));

если это выполнить с открытой консолью, то можно увидеть, каким работником какой пункт обрабатывается. "Свободная касса!" и всё такое

Сегодня баловался самой констуркцией. Пришел к вот такому рабочему варианту:

"use strict";
'esversion: 6';
(async function(){
 
  function procData(val, i){
    return new Promise((resolve, reject) => setTimeout(() => {
      var ret = i + "\t" + val + " #";
      console.log(ret);
      resolve(ret)
    }, 3000));
  }
 
 
  var arr = [3, 5, "asdf", 8, 0];
  var retArr = [];
 
  for(let n in arr){
    retArr.push(await procData(arr[n],n));
  }
 
  Promise.all(retArr).then((ret)=>{console.log(ret)});
}())

Оно работает, но я по прежнему не понимаю, почему все работает внутри безымянной функции с пометкой async, которая ещё и сразу вызывается, после объявления. Почему нельзя поставить async перед функцией внутри которой создается промис, почему нельзя поставить await перед самим промисом? Одни вопросы. Прочитал несколько статей на хабре, но материал, можно сказать, не усвоил. Но при этом с калбеками вроде более менее разобрался и ужасную лапшу из них строить могу :) А вот промисы и async/await не даются.

Спасибо за совет. Но я работаю с сервисом, который уже имеет некоторую настраиваемую многопоточность запросов, соединяюсь с ним через готовый модуль, который вроде имеет пул соединений (но это не точно - надо перепроверить), но всё равно я объемом тестовых данных кладу эту структуру. А если я ещё свой слой абстракции буду добавлять, то это только усложнить всю конструкцию и сделает её ещё более глючной (учитывая мои навыки).

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

for(let n in arr){
    retArr.push(await procData(arr[n],n));
  }
 
  Promise.all(retArr).then((ret)=>{console.log(ret)});

Это излишне.
Достаточно либо:

for(let n in arr){
    retArr.push(procData(arr[n],n));
  }
 
  Promise.all(retArr).then((ret)=>{console.log(ret)});

наполняем массив промисами, затем ждём выпонениния всех.
Либо:

for(let n in arr){
    retArr.push(await procData(arr[n],n));
  }
 
  console.log(retArr)

наполняем массив ожидая выполнения каждого промиса по очереди.

Потому что await работает только внутри async функции, и в простом коде ничего await нельзя. Как я сказал выше - это лишь обёртка над промисами, в сам язык вписанная, имхо, не слишком красиво.

Можно. Только если вызывать async функцию не поставив перед самим вызовом await, то ждать её вырлнения код не будет, и, хотя сам асинронный код в функции выполнится, но отдельно, не возвращая результата и не откладывая выполнения остального кода.

С коллбэков переходите на промисы. Самый постой вариант понять, как это работает - написать свой класс Promise, что абсолютно реализуемо классическими методами. (Если не получается - погуглить "промис полифил" и разобраться что да как)

Вот это вообще шикарно! Не уверен, что мне нужны отдельные воркеры (что модуль node который я использую не завалиться на них), но шикарно!

Спасибо за развернутое объяснение. С собственным классом промисов уже пытался. С большим количеством подсказок получилость сделать что-то рабочее, но так как полноценно не тестировал - использовать в собственном проекте не стал. У меня в части моих модулей модуль под названием promise, а в части bluebird :)
До сих пор не могу определиться что оставить )))

значит, там нет такого механизма, который бы умел работать с очередью. Не предусмотрен лимит количества одновременно выполняемых запросов. Точнее, предусмотрен, только в виде падения системы :)

хотя, не исключаю, что он там есть, только надо его как-то задействовать.

Вот, как то тоже боролся с подобной задачей. - но тут есть нюанс, обработку эксепшенов берите на себя (я же делал промежуточный обработчик - некий виртуальный слушатель). Ещё это можно решить рекурсивно и при падении (эксепш в цикле) перезапускать рекурсию с индексом падения. Такие помощники как Promise.all() не помогут когда надо "все" обработать в не зависимости от результата. Во время рекурсии результаты можно скидывать в стек, некий диспетчер, будет раздавать ответы по коду. Т.е. вы не ждете ответа от промиса а подписываетесь на событие с неким кодом, промис - когда выполниться сообщит вам о завершении через диспетчера.

Проще всего представлять себе «промис» как контейнер. Не какой-то конкретный, а просто любой контейнер. Подойдёт и просто обёртка для ссылки на объект: это — как бы контейнер, в котором может быть только один элемент.

Для этого контейнера должна быть определена операция, которая на самом деле — просто операция создания контейнера. По понятным причинам она должна быть: иначе их было бы невозможно использовать. Фактически эта операция — статичный метод, создающий этот контейнер — метод `Promise.resolve`.

Ещё должна быть определена операция отображения одного контейнера на другой. Более точней — операция «связывания» одного контейнера с другим. Чтобы понять, почему именно оно, а не что-то другое, рассмотрим сначала операцию, которая чуть проще — метод `map` у массива Array.

Пусть у нас имеется массив с элементами 1, 2 и 3. При помощи метода `map` этого массива мы можем над каждым элементом произвести какую-то операцию и получить новый массив, в котором находятся преобразованные данной операцией элементы. Например, мы можем умножить каждый элемент списка целых чисел на пять.

const a = [1, 2, 3];
const newA = a.map(x => 5 * x);

console.log(newA);
// > [5, 10, 15]

В скобках после `map` написана функция, при помощи которой делается преобразования каждого элемента.

Важно то, что на выходе получается контейнер того же класса: был Array — новый тоже будет Array-ем. При этом тип значения в новом контейнере может поменяться. То есть, мы могли бы не умножать на 5 элементы контейнера, а, например, превращать их в строки. Тогда у нас из Array<Number> получился бы Array<String>.

const a = [1, 2, 3];
const newA = a.map(x => `*${x}*`);

console.log(newA);
// > ["*1*", "*2*", "*3*"]

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

Рассмотрим пример: нужен поиск чего-то где-то. Например, первого элемента в массиве строк, который удовлетворяет заданным условиям. Сигнатура функции поиска в этом случае будет примерно такая.

function find(/* Array<String> */ array, /* String => Boolean */ criteria) {}

Первый аргумент функции — массив строк. Второй — функция, при помощи которой будет проверяться каждый аргумент на соответствие условию. Первый найденный элемент будет возвращён. Просто!

Ну а что возвращать, когда элемент, удовлетворяющий условию, не найден? null?

Ну, в этом примере оно, конечно, более-менее подходит, однако в общем случае null может считаться корректным значением и лежать в массиве. Как его отличить от «ничего не найдено», и, наконец, как потом обрабатывать результат? Ведь можно просто вызвать нужный метод у найденного объекта позабыв о том, что в качестве ответа может вернуться null.

И вот тут в тему оказываются контейнеры с одним элементом, которое может быть не указано, а именно так устроены «промисы»: надо просто результат поиска оборачивать в такой контейнер.

Рассмотрим один из них, назовём его Option. Option имеет два варианта: Some, то есть «какой-то», и None — «ничего». Если мы что-то нашли, то мы возвращаем Some(найденное), если нет — возвращаем None().

class Option extends Promise {}

function Some(v) { return new Option((r, _) => r(v)); }
function None( ) { return new Option((_, r) => r( )); }

Если оборачивать результат операций, подобных поиску или открытию файлов (т. е. те, где можно ничего не найти, неудачно открыть и т. п.) в такие контейнеры, то вышеописанные проблемы устраняются. Исключение при попытке вызова метода у null возникнуть уже не может. Есть, что возвращать даже для примитивных типов, безо всяких там «магических» констант вида «-1 означает, что ничего не найдено», и т. д.

Чтобы понять, в чём прикол такого контейнера, рассмотрим пример с серией операций.

Предположим, что у нас есть некоторый довольно примитивный набор методов для обработки файлов. Файл можно открыть, после этого можно прочитать его содержимое, которое можно «распарсить». И наконец, файл можно закрыть, чтобы высвободить ресурс. Для краткости вместо реализации этих методов я сделаю заглушки, ведь интересует только их использование.

class ParsedData {}

class File {
	read() { return ""; }
	close() {}
}

function openFile(fileName) { return new File }
function parseString(str) { return new ParsedData }

Примерно так выглядит способ работы с таким интерфейсом...

const file = openFile("my-file.txt");
const parsed = parseString(file.read());
file.close();

Всё cупер, кратко и лаконично. Однако есть много проблем. Каждый из этих методов может «не сработать». Файла с таким именем может не быть. Он может быть, но не читаться. Даже прочитавшись, он может содержать синтаксические ошибки и потому не «парситься». И закрытие файла тоже может завершиться с ошибкой.

Традиционный способ решения подразумевает использование try-catch. И вот во что превращаются эти простые три строки в данном случае.

let file, parsed;

try {
	file = openFile("my-file.txt");
	parsed = parseString(file.read());
} catch(error) {
	console.error(error);
} finally {
	try {
		if(file) file.close();
	} catch(error) {
		console.error(error);
	}
}

Это — ужасное, трудночитаемое и содержащее повторы нагромождение кода так разительно отличается от изначальных трёх строк...

Посмотрим, что нам даст вышерассмотренный контейнер с одним или нулём элементов — Option. Сначала переопределим методы для обработки файлов так, чтобы они не выкидывали исключений, а если возвращают результат, то возвращали бы его обёрнутым в Option.

class ParsedData {}

class File {
	read() { return Some(""); }
	close() { return Some() }
}

function openFile(fileName) { return Some(new File) }
function parseString(str) { return Some(new ParsedData) }

Воспользуемся тем, что Option — это контейнер. Причём он не просто контейнер, а контейнер с методом `then`, поскольку наследует от класса Promise. Это нам даёт следующее:

1. Если в контейнере нет элементов (None), то написанное в `then` не будет выполнено ни разу
2. Если в контейнере один элемент (Some), то написанное в `then` будет выполнено один раз — именно для того, что обёрнуто в этот Some
3. Нам вернётся результат выполнения, обёрнутый в Option: Some(результат), если всё прошло удачно, и None() — если нет.

Это позволяет нам написать вот так...

const file   = openFile("my-file.txt");
const parsed = file.then(f => f.read()).then(parseString);
file.then(f => f.close());

Пусть, с небольшими добавками, но это всё-таки почти те же исходные три строки. Почти столь же хорошо читаемые.

Теперь, что делать, если мы хотим не просто проигнорировать все ошибки, а уведомить о них пользователя?

Идея та же: контейнер с одним элементом. С той лишь разницей, что вместо None должно использоваться что-то типа OtherSome(exception), которая будет, подобно None, «путешествовать» от одной строки к другой и, в конце концов, вернётся к нам, если что-то пойдёт не так.

function OtherSome(exception) { return Promise.reject(exception); }

Ещё «промис» может выполнять роль контейнера Try — это такая версия контейнера с одним элементом, которая внутри себя перехватывает возникшие исключения. Но не просто их игнорирует, а запоминает внутри обёртки, которая трактуется так же, как в Option трактуется None: никаких преобразований совершать не надо, надо просто вернуть то же самое обёрнутое исключение.

function Success(v) { return Promise.resolve(v); }
function Failure(v) { return Promise.reject (v); }
function Try(fn) {
	return new Promise(r => r(fn()))
		.then(Success)
		.catch(Failure)
	;
}

// пример с вводом целого числа
Try(() => {
	const a = prompt("Insert number");
		if(/^\d+$/.test(a.trim()))
		return a;
	else
		throw "Not a Number";
}).then(console.log, alert);

В результате вышенаписанного в контейнере будет лежать либо Success(правильный ввод), либо Failure(то исключение, когда ввели не целое число).

Также «промис» можно представить контейнером — назовём его Future — над значением, которое может быть доступно в данный конкретный момент времени(resolved), или, возможно, будет доступно только в будущем(pending), или никогда не будет доступно(rejected)

function Future(fn) {
	return new Promise(fn);
}

В принципе роль функций Some, None, Success, Failure, Try, Future — вспомогательная, они возвращают «промис», который можно было бы создать напрямую в коде. Однако согласитесь, что наделяют большим смыслом то, что просходит в коде, чем просто Promise.resolve или Promise.reject.

Об отображении

Функция `Promise.prototype.then` работает так...

из объекта типа Promise<A>                    // обозначим его как p1
и функции типа (A => Promise<B>) или (A => B) // обозначим её  как f
делает Promise<B>                             // обозначим его как p2
т. е. p1.then(f) === p2

Для сравнения, функция `Array.prototype.map` работает так...

из объекта типа Array<A> // обозначим его как p1
и функции типа  (A => B) // обозначим её  как f
делает Array<B>          // обозначим его как p2
т. е. p1.map(f) === p2

Об упаковке в контейнер

Функция `Promise.resolve` работает так...

из объекта типа A // обозначим его как a
делает Promise<A> // обозначим его как p
т. е. Promise.resolve(a) === p

Для сравнения, функция `Array.of` работает так...

из объекта типа A // обозначим его как a
делает Array<A>   // обозначим его как p
т. е. Array.of(a) === p

Продолжение скоро...