Настройка распределённых Background Jobs (несколько воркеров)

Настройка распределённых Background Jobs (несколько воркеров)

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

Архитектура

[App Server 1]  [App Server 2]  [App Server 3]
      ↓                ↓               ↓
   dispatch         dispatch        dispatch
      ↓                ↓               ↓
         ┌─────────────────────────────┐
         │     Redis / RabbitMQ        │  ← централизованный брокер
         └─────────────────────────────┘
              ↓           ↓         ↓
        [Worker 1]  [Worker 2]  [Worker 3]   ← могут быть на разных серверах

Брокер — единственный компонент, который должен быть доступен всем серверам. Остальные узлы не общаются напрямую.

Требования к брокеру

Redis — стандартный выбор для Laravel. Требует phpredis или predis. Для высокой доступности — Redis Sentinel или Redis Cluster.

RabbitMQ — подходит для сложных routing-сценариев (fanout, topic exchanges). Laravel поддерживает через пакет vladimir-yuldashev/laravel-queue-rabbitmq.

Amazon SQS — управляемый сервис, не нужно обслуживать. Подходит при инфраструктуре на AWS.

Минимальная конфигурация Redis для production — отдельный сервер (не shared с основной БД), persistence включена (appendonly yes), maxmemory-policy настроена.

Конфигурация Laravel для distributed workers

// config/queue.php
'connections' => [
    'redis' => [
        'driver'       => 'redis',
        'connection'   => 'queue',   // отдельный Redis-коннект для очередей
        'queue'        => env('REDIS_QUEUE', 'default'),
        'retry_after'  => 90,        // секунды до повтора зависшей задачи
        'block_for'    => 5,         // блокирующий BLPOP вместо polling
        'after_commit' => true,      // диспатч только после commit транзакции БД
    ],
],

retry_after — ключевой параметр при распределённых воркерах: если воркер упал в процессе задачи, задача будет повторно видна другим воркерам через retry_after секунд. Должен быть больше timeout Job'а.

Горизонтальное масштабирование через Horizon

Horizon поддерживает запуск на нескольких серверах. Каждый сервер запускает свой экземпляр Horizon, они не координируются между собой напрямую — Redis выступает общим реестром.

На каждом сервере запускается одинаковый Supervisor-конфиг:

[program:horizon]
command=php /var/www/artisan horizon
autostart=true
autorestart=true
user=www-data
stdout_logfile=/var/log/horizon.log
stopwaitsecs=3600

Horizon автоматически балансирует воркеры внутри одного сервера. Для балансировки между серверами — ручная настройка числа процессов с учётом мощности каждого.

Конкурентный доступ и дедупликация

При нескольких воркерах одна задача может быть взята дважды, если воркер завис и не снял блокировку. Механизм Redis LPOP атомарен — задача берётся одним воркером. Но «невидимые» задачи (взятые, но не завершённые) возвращаются в очередь через retry_after.

Если задача должна выполняться строго один раз (idempotency) — проверяем это явно:

class ProcessPaymentJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(private string $paymentId) {}

    public function handle(): void
    {
        // Distributed lock через Redis — только один воркер обрабатывает платёж
        $lock = Cache::lock("payment:{$this->paymentId}", 120);

        if (!$lock->get()) {
            // Другой воркер уже обрабатывает
            $this->release(10); // вернуть в очередь через 10 секунд
            return;
        }

        try {
            $payment = Payment::find($this->paymentId);

            // Проверка идемпотентности
            if ($payment?->status !== 'pending') {
                return; // уже обработан
            }

            $this->processPayment($payment);
        } finally {
            $lock->release();
        }
    }
}

Cache::lock() использует Redis SET NX PX — атомарная операция, гарантирующая, что блокировку получит ровно один воркер.

Разделение воркеров по типу нагрузки

На разных серверах можно запускать воркеры для разных очередей, если задачи требуют специфических ресурсов:

[Server: API-1, API-2]     → воркеры для 'critical', 'default'
[Server: Media-1]          → воркеры для 'transcoding', 'media'
[Server: Worker-1]         → воркеры для 'batch', 'reports', 'low'

На медиа-сервере стоит GPU или мощный CPU для FFmpeg; на API-серверах — быстрые воркеры с малым timeout.

Supervisor на Media-сервере:

[program:media-worker]
command=php /var/www/artisan queue:work --queue=transcoding,media --timeout=3600 --max-jobs=1
numprocs=2
autostart=true
autorestart=true
user=www-data

--max-jobs=1 — воркер берёт одну задачу и перезапускается (освобождает память после тяжёлой операции).

Graceful shutdown

При деплое нужно дождаться завершения текущих задач, не убивая воркеры резко:

php artisan queue:restart

Эта команда ставит флаг в Redis — воркеры завершат текущую задачу и остановятся. Supervisor перезапустит их с новым кодом.

В Supervisor stopwaitsecs должен быть не меньше максимального timeout задачи:

stopwaitsecs=3600   # для сервера с транскодированием
stopwaitsecs=120    # для стандартных воркеров

Мониторинг распределённого состояния

Horizon агрегирует метрики всех серверов в одном дашборде. Ключевые показатели:

• Throughput (задач/минуту) по каждой очереди
• Wait time — среднее время ожидания задачи в очереди
• Runtime — среднее время выполнения
• Failed jobs — количество упавших задач

Автоматическое масштабирование воркеров (если инфраструктура на Kubernetes):

# HPA для масштабирования подов воркеров по метрике длины очереди
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: queue-workers
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: queue-worker
  minReplicas: 2
  maxReplicas: 20
  metrics:
    - type: External
      external:
        metric:
          name: redis_queue_depth
          selector:
            matchLabels:
              queue: default
        target:
          type: AverageValue
          averageValue: "50"  # масштабируем если > 50 задач на воркер

Кастомная метрика redis_queue_depth экспортируется через Prometheus Redis Exporter.

RabbitMQ как альтернатива

При необходимости сложной маршрутизации (разные типы событий → разные очереди, fanout рассылка) RabbitMQ даёт больше гибкости:

// config/queue.php
'rabbitmq' => [
    'driver'   => 'rabbitmq',
    'dsn'      => env('RABBITMQ_DSN', 'amqp://user:pass@localhost:5672/'),
    'queue'    => env('RABBITMQ_QUEUE', 'default'),
    'options'  => [
        'exchange' => [
            'name' => 'app-exchange',
            'type' => 'direct',
        ],
        'queue' => [
            'durable'     => true,
            'exclusive'   => false,
            'auto_delete' => false,
        ],
    ],
],

RabbitMQ Management UI (порт 15672) предоставляет детальный мониторинг: consumers, connections, channel-загрузку, message rates.

Сроки

Настройка Redis Sentinel/Cluster или RabbitMQ, конфигурация Horizon на нескольких серверах, Supervisor — 1 рабочий день. Distributed locks, idempotency-проверки в критичных Job'ах — 6–8 часов. Интеграция с Kubernetes HPA и Prometheus — отдельный проект на 1–2 дня.