Конкуренция воркеров

Оглавление серии

Один worker может ждать сетевой операции и одновременно обрабатывать другую задачу. Такой режим называют concurrency — числом одновременно выполняемых processor-вызовов данного исполнителя. Увеличим его до трёх и рассмотрим конкуренцию разных поступлений. Благодаря атомарному applyShipment отдельные прибавки не должны теряться, а повтор одного batchId по-прежнему возвращает сохранённый результат.

Десятое состояние stages/10 меняет только предел конкурентной работы и сохраняет прежний договор малого CSV. Пример not_run: несколько workers, измерения скорости и нагрузка не запускались. Значение 3 выбрано для объяснения, не как оптимальная настройка машины. Приготовленный код не доказывает линейное ускорение или определённое количество импортов в секунду.

Concurrency относится к процессу

Опция worker concurrency равна трём. Это позволяет иметь до трёх активных задач внутри этого worker. Если запустить ещё один процесс с тем же значением, общий предел станет до шести, если нет отдельного глобального ограничения. Поэтому локальную опцию нельзя использовать как обещание общего лимита всего сервиса при масштабировании. Имя очереди и prefix связывают процессы, а число мест вычисляется из их конфигурации.

Следующий фрагмент находится в объекте опций Worker в worker.js. Connection создан с режимом worker, prefix общий с producer, а параметры блокировки сохранены из предыдущего урока.

{
  connection,
  prefix,
  lockDuration: 30000,
  stalledInterval: 30000,
  maxStalledCount: 1,
  concurrency: 3
}

Конкурентность не превращает обычный JavaScript в три параллельных CPU-потока. Асинхронные ожидания дают процессу возможность обслуживать другие задачи, но тяжёлые синхронные вычисления продолжают занимать event loop. Для IO-bound работы увеличение может помочь использовать время ожидания; для CPU-bound задачи нужен другой способ выполнения и измерение ресурсов. Число processor-вызовов и число вычислительных потоков обозначают разные свойства.

Запись должна выдерживать пересечение

Вернёмся к ошибке раннего worker. Два процесса могли прочитать остаток лампы 8, вычислить свои прибавки и записать конкурирующие результаты HSET. Последняя запись уничтожала результат предыдущей. Увеличение concurrency делало бы ошибку вероятнее, но сама причина существовала и при одном месте в каждом из двух процессов. Правильность нельзя связывать только с текущим числом исполнителей, если архитектура допускает несколько процессов.

Теперь Lua читает и пишет пакет без вмешательства другой команды. Если пакет A прибавляет 12, а пакет B тоже прибавляет 12, они последовательно изменят 8 в 20 и затем в 32. При обратном порядке итог тот же, поскольку предметная операция является сложением. Для назначения абсолютного значения или изменения цены порядок мог бы влиять на результат, и тогда потребовался бы другой договор конфликтов, версий или приоритетов.

Ожидаемый результат двух разных batchId приведён ниже. Это расчёт, а не фактически измеренный конкурентный запуск.

Начало: Лампа=8
A: +12
B: +12
После обоих разных пакетов: Лампа=32
Повтор A: Лампа остаётся32

Если A и B на самом деле имеют одинаковый batchId, скрипт применит только первый логический пакет и второй получит прежний result. В этой серии dataset неизменен, поэтому такой повтор имеет одинаковое содержимое. При разных входных файлах production-договор обязан сравнивать содержимое и возвращать конфликт. Конкуренция делает важность идентичности действия особенно заметной: технически разные job-вызовы могут относиться к одному бизнес-событию.

Порядок завершения не равен порядку постановки

Даже если задачи добавлены A, затем B, B может завершиться раньше, когда её вычисление короче или A ждёт зависимость. Интерфейс должен показывать результат каждой по собственному jobId, а не считать последнюю полученную консольную строку ответом на последний HTTP-запрос. Прикладная связь batchId и результата сохраняется независимо от порядка завершений. Очередь задаёт получение работы, но не обещает строгий порядок окончания конкурентных операций.

Импортёр читает небольшой локальный файл для каждой задачи. Одновременное чтение допустимо, потому что данные не меняются. Если приложение принимало бы изменяемый путь к файлу, две задачи могли бы прочитать разные версии под одним именем. Поэтому producer должен ссылаться на неизменное сохранённое содержимое или версию объекта. Concurrency не является причиной этой проблемы, но чаще выявляет зависимость от скрытого изменяемого входа.

Полное заполнение отсутствующих продуктов через HSETNX сохраняется. Несколько processors могут одновременно проверить начальную модель, однако существующее поле не заменяется. Это небольшое согласованное действие отличается от раннего read-modify-write количества. Нельзя сделать вывод, что все операции с hash автоматически безопасны: свойство зависит от команды и расположения чтения с записью, а не от одного используемого типа данных.

Ограничение выбирается по зависимости

Слишком высокая concurrency может перегрузить Redis, файловую систему или внешнее API. Она также увеличивает число активных задач, для которых необходимо продлевать блокировки и хранить контекст. Поэтому увеличение не является бесплатной оптимизацией. Обычно выбирают предел по ресурсу с наименьшей допустимой параллельностью и наблюдают время ожидания, ошибки и завершения. При общих внешних ограничениях нужен общий rate limit обработки, отдельный от HTTP-лимита постановки.

В нашем сценарии Lua выполняется Redis последовательно независимо от числа workers. Увеличение мест может ускорить ожидание чтения, но не заставит три Lua-скрипта одновременно менять один Redis. Если именно скрипт становится узким местом, бесконечное добавление исполнителей лишь увеличит очереди ожидания и нагрузку на подключения. Сначала определите, какой участок потребляет время, затем выбирайте способ масштабирования соответствующего участка.

Возвращённый result относится к пакету, а не к общей версии каталога после всех конкурентных задач. Другой пакет мог уже увеличить generation к моменту чтения клиентом. Если интерфейсу нужна текущая сводка, он должен выполнить GET каталога. Если нужен документ о конкретном поступлении, result должен содержать собственные значения этого пакета. Смешение текущего состояния и результата действия создаёт непонятные отчёты при любом пересечении работы.

Для самостоятельного анализа измените мысленно операцию с прибавления на абсолютное присваивание остатка. Теперь A=20 и B=15 дадут итог последней записи, а не сумму. Это другой предметный договор, который нельзя исправить повышением числа попыток или мест worker. Пример показывает, почему выбор атомарной операции следует из смысла данных и требуемой обработки конфликтов, а не только из возможностей очереди.

Локальная конкурентность описана в BullMQ Concurrency. После десятого состояния работа допускает несколько processors, сохраняя атомарное принятие каждого малого пакета. Следующий урок покажет, как наблюдать состояние и прогресс, не делая вывод о завершении только по скорости появления HTTP-ответа или консольных строк.