Очереди и фоновые воркеры

01Зачем нужны очереди: продюсеры, консьюмеры и брокеры

Представьте планку для чеков на кухне ресторана. Официант (продюсер) прикалывает чек с заказом и возвращается к гостям; повара (консьюмеры) берут чеки, когда освобождаются. Если одновременно пришло десять заказов, чеки ждут на планке, а официант не застывает на месте. В программах эта планка — брокер: Redis, таблица в базе данных, RabbitMQ или SQS. Ваше веб-приложение — продюсер, процессы-воркеры — консьюмеры. Преимущества: быстрые ответы, сглаживание пиков трафика, изоляция сбоев медленных сервисов и масштабирование воркеров независимо от веб-серверов. Цена: работа становится асинхронной, и приходится думать о повторах, дубликатах, порядке и мониторинге.

• Задача (job): небольшое сообщение с типом и данными, в идеале только id, а не объекты целиком.
• Имена очередей разделяют приоритеты: critical, default, emails, reports.
• Конкурентность: сколько задач воркер выполняет одновременно.

02Повторы, backoff и dead-letter-очереди

Задачи падают: платёжный API отвечает по таймауту, в базе случается дедлок, сторонний сервис возвращает 503. Повторяйте, но аккуратно. Экспоненциальный backoff каждый раз ждёт дольше (например, 10 секунд, 1 минуту, 5 минут), давая другой стороне время восстановиться; jitter разносит повторы, чтобы они не ударили все разом. Ограничивайте число попыток. Различайте ошибки: таймаут стоит повторить, а ошибка валидации («клиент не существует») будет повторяться вечно, поэтому сразу помечайте задачу как проваленную. После последней попытки переносите задачу в dead-letter-очередь (в Laravel — таблица failed_jobs, в BullMQ — набор failed). Кто-то должен туда смотреть: настройте алерты на количество сбоев и сделайте способ просматривать и перезапускать задачи после исправления причины.

1class SyncOrderToErp implements ShouldQueue
2{
3    use Queueable;
4 
5    public int $tries = 5;
6    public int $timeout = 60;
7 
8    public function __construct(public int $orderId) {}
9 
10    public function backoff(): array
11    {
12        return [10, 60, 300, 900];
13    }
14 
15    public function handle(ErpClient $erp): void
16    {
17        $order = Order::findOrFail($this->orderId);
18        $erp->upsertOrder($order); // idempotent by order id on the ERP side
19    }
20}

03Идемпотентные задачи: безопасно выполнить дважды

Большинство очередей гарантируют доставку at-least-once: задача может выполниться дважды — например, если воркер упал после того, как сделал работу, но до подтверждения. Поэтому каждая задача должна быть идемпотентной, то есть повторный запуск не даёт дополнительного эффекта. Приёмы: использовать естественный уникальный ключ (создать счёт для заказа 123, только если его ещё нет), хранить отметку об обработке с уникальным ограничением, использовать upsert вместо insert и передавать платёжным провайдерам idempotency key. Проверяйте состояние перед действием («уже отправлено?»). Доставка exactly-once между системами — во многом миф; эффект exactly-once достигается идемпотентными обработчиками. И избегайте классической гонки: диспатчьте задачи только после коммита транзакции в базе, иначе воркер может взять задачу для строки, которой ещё не существует.

04BullMQ в Node

BullMQ — стандартная очередь на Redis для Node. Queue добавляет задачи, Worker обрабатывает их с заданной конкурентностью, а опции управляют числом попыток, backoff, задержками, приоритетом и очисткой. Задайте jobId для дедупликации: добавление задачи с уже существующим id игнорируется. Повторяющаяся работа использует job schedulers (upsertJobScheduler) с cron-паттерном или интервалом; в BullMQ v6 старую опцию repeat убрали в пользу планировщиков. Запускайте воркеры отдельными от API процессами, чтобы тяжёлые задачи никогда не замедляли запросы, и задавайте каждой очереди свою конкурентность. Redis должен быть настроен с maxmemory-policy noeviction, иначе он может молча удалять задачи.

1import { Queue, Worker } from 'bullmq';
2const connection = { host: '127.0.0.1', port: 6379 };
3 
4export const webhooks = new Queue('webhooks', { connection });
5 
6await webhooks.add('deliver', { eventId }, {
7  jobId: 'evt-' + eventId,              // dedupe
8  attempts: 8,
9  backoff: { type: 'exponential', delay: 5_000 },
10  removeOnComplete: 1000,
11  removeOnFail: 5000,
12});
13 
14new Worker('webhooks', async (job) => deliverWebhook(job.data.eventId), {
15  connection,
16  concurrency: 10,
17});

05Расписание и запуск воркеров в продакшене

Работа по расписанию (ночные отчёты, истечение подписок, ежечасные синхронизации) запускается по таймеру. Классический cron выполняет команду в фиксированное время; планировщику Laravel нужна одна запись в cron, а остальное описывается в коде; планировщики BullMQ хранят расписания в Redis. Делайте задачи по расписанию безопасными при наложении или явно запрещайте наложение и запускайте их только на одном сервере. Воркеры — это долгоживущие процессы, поэтому их должен поддерживать живыми менеджер процессов: supervisor или systemd для воркеров Laravel, pm2 или systemd для Node, или по контейнеру на каждый тип воркера в Docker. Перезапускайте воркеры на каждом деплое, чтобы они загрузили новый код, и обеспечьте graceful shutdown, чтобы выполняющаяся задача успела завершиться.

1; /etc/supervisor/conf.d/app-worker.conf
2[program:app-worker]
3command=php /var/www/app/artisan queue:work redis --queue=critical,default --sleep=3 --tries=3 --max-time=3600
4process_name=%(program_name)s_%(process_num)02d
5numprocs=2
6user=app
7autostart=true
8autorestart=true
9stopwaitsecs=3600
10stdout_logfile=/var/log/app-worker.log

06Входящие и исходящие вебхуки

Входящие вебхуки (от платёжных провайдеров, ERP, служб доставки): проверьте подпись, сохраните сырое событие, быстро ответьте 200 и обработайте его в задаче. Провайдеры повторяют отправку, поэтому дедуплицируйте по их id события. Когда речь о деньгах, никогда не доверяйте одному лишь payload: заново запросите статус платежа через API провайдера. Исходящие вебхуки (вашим партнёрам): записывайте событие в таблицу outbox в той же транзакции, что и бизнес-изменение, пусть воркер его доставит, подписывайте тело через HMAC-SHA256 с временной меткой, чтобы получатели могли его проверить и отклонить повторы, повторяйте с backoff в течение суток или дольше и показывайте партнёру логи доставки. Этот паттерн outbox гарантирует, что вы никогда не уведомите о том, что было откачено, и не потеряете событие, которое было закоммичено.

07Как задача «резервируется»: visibility timeout, блокировки и зависшие задачи

Когда воркер берёт задачу, брокер её не удаляет. Он скрывает её на время и ждёт подтверждения. Если подтверждение не пришло вовремя, он считает, что воркер умер, и отдаёт задачу кому-то другому. В SQS это visibility timeout. В Laravel это retry_after в config/queue.php: если задача всё ещё выполняется спустя столько секунд, другой воркер запускает её заново, поэтому timeout задачи всегда должен быть меньше retry_after, иначе долгие задачи выполняются дважды параллельно. BullMQ использует блокировку, которую воркер постоянно продлевает во время обработки; если блокировка истекла, например потому что тяжёлый по CPU код заблокировал event loop, задача помечается как stalled и возвращается в ожидание, а после слишком большого числа таких случаев падает. Все три подхода дают доставку at-least-once, и именно поэтому идемпотентность обязательна.

1// config/queue.php
2'redis' => [
3    'driver' => 'redis',
4    'connection' => 'default',
5    'queue' => env('REDIS_QUEUE', 'default'),
6    'retry_after' => 960,     // longer than the slowest job's timeout
7    'block_for' => 5,
8    'after_commit' => true,   // push only after the DB transaction commits
9],
10 
11// Slow export job: public $timeout = 900; run it on a worker
12// started with --timeout=900, on its own queue.

08Управление конкурентностью: уникальность, перекрытие и rate limits

Повторы и параллельные воркеры создают новые проблемы: две синхронизации одного клиента соревнуются друг с другом, или всплеск задач упирается в rate limit партнёра. Laravel даёт строительные блоки. ShouldBeUnique игнорирует новую отправку, пока идентичная задача (по uniqueId) ещё ожидает. Job middleware WithoutOverlapping разрешает одновременно выполняться только одной задаче на ключ и возвращает остальные в очередь. RateLimited применяет к задачам именованный rate limiter, а ThrottlesExceptions отступает после повторяющихся ошибок, работая как circuit breaker для нестабильного API; сочетайте его с retryUntil, чтобы повторы ограничивало время, а не число попыток. Блокировкам уникальности и перекрытия нужно кэш-хранилище с атомарными блокировками, например Redis или база данных. В BullMQ опция deduplication игнорирует повторные задачи с тем же id, а limiter воркера ограничивает число задач за окно времени.

1class SyncCustomerToCrm implements ShouldQueue, ShouldBeUnique
2{
3    use Queueable;
4 
5    public function __construct(public int $customerId) {}
6    public function uniqueId(): string
7    {
8        return (string) $this->customerId;
9    }
10 
11    public function middleware(): array
12    {
13        return [
14            new WithoutOverlapping('crm-'.$this->customerId),
15            new ThrottlesExceptions(10, 5 * 60), // 10 errors, then pause 5 min
16        ];
17    }
18 
19    public function retryUntil(): DateTime { return now()->addHours(6); }
20}

09Workflows: цепочки, батчи и родительско-дочерние потоки

Реальная работа часто состоит из нескольких задач с зависимостями. Цепочка запускает задачи одну за другой и останавливается, если одна упала: сгенерировать счёт, затем отправить его, затем уведомить бухгалтера. Батч запускает много задач параллельно и вызывает колбэк, когда все они завершились, что подходит для импортов и экспортов, разбитых на тысячи маленьких задач; Laravel отслеживает прогресс в таблице job_batches, а задачи в батче используют трейт Batchable. В BullMQ FlowProducer создаёт деревья родитель–потомок, где родитель выполняется только после завершения всех потомков и может читать их результаты. Проектируйте каждый шаг идемпотентным и маленьким, чтобы повтор одного шага не переделывал весь workflow. Когда поздний шаг падает после того, как ранние уже изменили внешние системы, нужна явная компенсация, например отмена резервирования, потому что очередь не может откатить платёж.

1$batch = Bus::batch(
2    $orderIds->map(fn (int $id) => new ExportOrderRow($exportId, $id))->all()
3)
4    ->name('orders-export-'.$exportId)
5    ->allowFailures()
6    ->then(fn (Batch $batch) => BuildExportFile::dispatch($exportId))
7    ->catch(fn (Batch $batch, Throwable $e) => report($e))
8    ->finally(fn (Batch $batch) => CleanupExportChunks::dispatch($exportId))
9    ->onQueue('reports')
10    ->dispatch();
11 
12// Progress for the UI: Bus::findBatch($batch->id)->progress()

10Ёмкость и наблюдаемость: глубина, возраст и закон Литтла

Сколько воркеров вам нужно? Закон Литтла даёт первый ответ: среднее число задач в обработке равно скорости поступления, умноженной на среднее время обработки. Если в секунду приходит 30 задач и каждая занимает 200 мс, нужно минимум 6 параллельных слотов только чтобы успевать, и больше — для всплесков и повторов. Одна лишь глубина очереди вводит в заблуждение: 10,000 крошечных задач могут разойтись за секунды, а 50 медленных могут отставать на часы. Самая полезная единичная метрика — возраст самой старой ожидающей задачи в каждой очереди с алертом относительно цели, например «письма в течение 60 секунд». Отслеживайте также пропускную способность, долю ошибок и перцентили времени обработки. Horizon умеет автоматически балансировать воркеры между очередями, а php artisan queue:monitor может генерировать событие, когда очередь превышает порог. В мультитенантных системах следите за справедливостью: импорт одного тенанта не должен задерживать чеки всех остальных.

• Автомасштабируйте воркеры по возрасту самой старой задачи, а не по CPU.
• Разделяйте очереди по целевой задержке: critical, default, bulk.
• Ограничивайте конкурентность на тенант для тяжёлых типов задач.