Коротко — пример кода и обсуждение проблем + способы их решения.
1) Реализация с контролем степени параллелизма и сохранением порядка результатов:
async function processParallel(items, concurrency = 5) {
const results = new Array(items.length);
let index = 0;
async function worker() {
while (true) {
const i = index++;
if (i >= items.length) return;
try {
results[i] = await doWork(items[i]);
} catch (err) {
// проброс ошибки вверх (fail-fast)
throw err;
}
}
}
const workers = Array.from({ length: concurrency }, () => worker());
await Promise.all(workers);
return results;
}
Пример вызова: await processParallel(items, 5).
2) Вариант «не прерывать при ошибках» (сбор ошибок, продолжение):
async function processParallelAll(items, concurrency = 5) {
const results = new Array(items.length);
const errors = new Array(items.length).fill(null);
let index = 0;
async function worker() {
while (true) {
const i = index++;
if (i >= items.length) return;
try {
results[i] = await doWork(items[i]);
} catch (err) {
errors[i] = err;
}
}
}
const workers = Array.from({ length: concurrency }, () => worker());
await Promise.all(workers);
return { results, errors };
}
3) Простая альтернатива — библиотека p-limit (коротко):
const limit = pLimit(concurrency);
await Promise.all(items.map((it, i) => limit(() => doWork(it))));
Обсуждение проблем и способы решения
- Ограничения ресурсов:
- CPU/потоки: для CPU-bound задач параллелизм ограничен числом ядер; для Node.js — используйте worker_threads или внешние процессы.
- Сеть/БД/коннекты: слишком большой параллелизм приведёт к исчерпанию соединений/протоколам. Решение: устанавливать разумный лимит по эмпирике, использовать пул соединений, rate limiting.
- Память: хранение всех результатов в массиве вызывает рост памяти. Решение: обрабатывать результаты по мере готовности (stream/`for await`/callback), не собирать всё в память.
- Порядок результатов:
- В примере выше порядок сохраняется, потому что мы записываем результат по индексу.
- Если порядок не важен — можно записывать результаты по готовности (увеличивает throughput).
- Если нужен порядок, но при этом минимизировать задержки, можно отдавать результаты по готовности и реассамблировать потоком по индексам или возвращать объект {index, value}.
- Исключения и отказоустойчивость:
- Fail-fast (первая ошибка отменяет всю обработку): просто пробрасывать ошибку из worker и использовать Promise.all — остальные стартованные задачи продолжат выполняться, но вы получите отклонённый промис. Чтобы действительно остановить дальнейшую работу, используйте флаг отмены и передавайте AbortSignal в doWork.
- Собрать все ошибки: использовать конструкцию с errors[] и вернуть их вместе с результатами или применить Promise.allSettled.
- Повторы/рестарт: на ошибки сетевого характера применять retry с экспоненциальным бэкофом и лимитом повторов.
- Атомарность/транзакции: при постобработке учитывайте частично выполненные операции (компенсирующие действия).
- Отмена/прерывание:
- Нативно используйте AbortController/AbortSignal: передавайте signal в doWork и проверяйте signal.aborted.
- Без поддержки отмены — можно установить флаг и больше не запускать новые задачи, но запущенные завершатся.
- Производительность/тюнинг:
- Подбирать параллелизм эмпирически: начните с малого (например, $5$–$20$ для сетевых запросов), измеряйте latency и ресурсное потребление.
- Используйте таймауты и пул соединений.
- Для очень больших потоков — переходите на потоковую обработку (streams, async iterators) вместо загрузки всего массива.
Резюме:
- Для I/O-bound задач параллелизм полезен; контролируйте его числом воркеров $concurrency$.
- Решайте порядок через хранение по индексам; исключения — через fail-fast, агрегирование или Promise.allSettled; отмену — через AbortController.
- Тестируйте и подбирайте значение параллелизма по нагрузке и доступным ресурсам.