Laravel Queues vs Symfony Messenger: архитектура, retry, батчи, бенчмарки

Очереди в PHP: почему это критически важно

Любое веб-приложение рано или поздно упирается в задачи, которые нельзя выполнять в HTTP-запросе: отправка email, генерация PDF, обработка загруженных файлов, вызовы внешних API, ресайз изображений. Пользователь нажал кнопку — и ждёт 15 секунд, пока сервер отправит письмо через SMTP. Это неприемлемо.

Очереди решают проблему: вместо синхронного выполнения задача кладётся в очередь, воркер забирает её в фоне, а пользователь получает ответ мгновенно. В PHP-экосистеме два главных решения: Laravel Queues и Symfony Messenger. Оба зрелые, оба production-ready, но архитектурно они принципиально разные.

Я работал с обоими: Laravel Queues — на двух e-commerce проектах с десятками тысяч джобов в час, Symfony Messenger — на финтех-проекте с event-driven архитектурой. Вот детальное сравнение на основе реального опыта.

Архитектура: два подхода к одной задаче

Laravel Queues — Job-центричная модель

Laravel строит очереди вокруг концепции Job — класса, который инкапсулирует единицу работы:

// app/Jobs/SendWelcomeEmail.php
class SendWelcomeEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(
        public readonly User $user,
    ) {}

    public function handle(Mailer $mailer): void
    {
        $mailer->to($this->user->email)
            ->send(new WelcomeEmail($this->user));
    }

    public function failed(\Throwable $e): void
    {
        Log::error("Welcome email failed for user {$this->user->id}: {$e->getMessage()}");
    }
}

// Диспатч — одна строка
SendWelcomeEmail::dispatch($user);

Всё в одном месте: данные ($user), логика (handle), обработка ошибок (failed), конфигурация (очередь, задержка, retry). Job — это и message, и handler одновременно.

Symfony Messenger — Message + Handler

Symfony разделяет сообщение и обработчик:

// src/Message/SendWelcomeEmail.php
class SendWelcomeEmail
{
    public function __construct(
        public readonly int $userId,
    ) {}
}

// src/MessageHandler/SendWelcomeEmailHandler.php
#[AsMessageHandler]
class SendWelcomeEmailHandler
{
    public function __construct(
        private readonly UserRepository $userRepository,
        private readonly MailerInterface $mailer,
    ) {}

    public function __invoke(SendWelcomeEmail $message): void
    {
        $user = $this->userRepository->find($message->userId);
        if (!$user) {
            return;
        }

        $this->mailer->send(
            (new Email())
                ->to($user->getEmail())
                ->subject('Welcome!')
                ->htmlTemplate('emails/welcome.html.twig', ['user' => $user])
        );
    }
}

// Диспатч
$this->messageBus->dispatch(new SendWelcomeEmail($user->getId()));

Сообщение — простой DTO с данными. Обработчик — отдельный класс с зависимостями через DI-контейнер. Один message может иметь несколько handlers.

Ключевое архитектурное отличие

Транспорты и драйверы

Laravel: встроенные драйверы

// config/queue.php
'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => 'default',
        'retry_after' => 90,
        'block_for' => 5,
    ],

    'database' => [
        'driver' => 'database',
        'table' => 'jobs',
        'queue' => 'default',
        'retry_after' => 90,
    ],

    'sqs' => [
        'driver' => 'sqs',
        'key' => env('AWS_ACCESS_KEY_ID'),
        'secret' => env('AWS_SECRET_ACCESS_KEY'),
        'prefix' => env('SQS_PREFIX'),
        'queue' => env('SQS_QUEUE', 'default'),
        'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    ],

    'sync' => [
        'driver' => 'sync', // для тестов — выполняется сразу
    ],
],

Поддерживаемые драйверы из коробки: Redis, Database, Amazon SQS, Beanstalkd, Sync. Redis — самый популярный для production.

Symfony: транспорты через DSN

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                retry_strategy:
                    max_retries: 3
                    delay: 1000
                    multiplier: 2
                    max_delay: 60000

            async_priority_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: high

            failed:
                dsn: 'doctrine://default?queue_name=failed'

        routing:
            'App\Message\SendWelcomeEmail': async
            'App\Message\ProcessPayment': async_priority_high
            'App\Message\GenerateReport': async

# .env
MESSENGER_TRANSPORT_DSN=redis://localhost:6379/messages
# или
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages
# или
MESSENGER_TRANSPORT_DSN=doctrine://default

Поддерживаемые транспорты: AMQP (RabbitMQ), Doctrine (database), Redis, Amazon SQS, Beanstalkd, In-memory. AMQP — предпочтительный для production в Symfony-мире.

Сравнение транспортов

Retry и обработка ошибок

Laravel: retry на уровне Job

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

    public $tries = 5;           // максимум попыток
    public $maxExceptions = 3;   // максимум исключений (не считая timeout)
    public $timeout = 120;       // секунд на выполнение
    public $backoff = [10, 30, 60, 120, 300]; // задержка между попытками

    public function __construct(
        public readonly Order $order,
    ) {}

    public function handle(PaymentGateway $gateway): void
    {
        $result = $gateway->charge(
            $this->order->total,
            $this->order->payment_method_id,
        );

        if (!$result->success) {
            // Условный retry — только для recoverable ошибок
            if ($result->isRetryable()) {
                $this->release($this->backoff[$this->attempts() - 1] ?? 300);
                return;
            }

            $this->fail(new PaymentFailedException($result->error));
        }

        $this->order->markPaid($result->transaction_id);
    }

    public function failed(\Throwable $e): void
    {
        $this->order->markPaymentFailed($e->getMessage());
        Notification::route('slack', config('services.slack.payments'))
            ->notify(new PaymentFailedNotification($this->order, $e));
    }

    public function retryUntil(): DateTime
    {
        // Не пытаться дольше 24 часов
        return now()->addHours(24);
    }
}

Laravel позволяет настраивать retry на уровне каждого Job: количество попыток, backoff-стратегию, таймауты, максимальное время жизни. Метод failed() вызывается после исчерпания всех попыток.

Symfony: retry через middleware и stamps

// src/Message/ProcessPayment.php
class ProcessPayment
{
    public function __construct(
        public readonly int $orderId,
        public readonly string $paymentMethodId,
    ) {}
}

// src/MessageHandler/ProcessPaymentHandler.php
#[AsMessageHandler]
class ProcessPaymentHandler
{
    public function __construct(
        private readonly OrderRepository $orderRepository,
        private readonly PaymentGateway $gateway,
        private readonly NotifierInterface $notifier,
    ) {}

    public function __invoke(ProcessPayment $message): void
    {
        $order = $this->orderRepository->find($message->orderId);

        $result = $this->gateway->charge(
            $order->getTotal(),
            $message->paymentMethodId,
        );

        if (!$result->isSuccess()) {
            if ($result->isRetryable()) {
                throw new RecoverableMessageHandlingException($result->getError());
            }

            $order->markPaymentFailed($result->getError());
            throw new UnrecoverableMessageHandlingException($result->getError());
        }

        $order->markPaid($result->getTransactionId());
    }
}

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_payments:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: payments
                retry_strategy:
                    max_retries: 5
                    delay: 10000        # 10 секунд
                    multiplier: 3       # экспоненциальный backoff
                    max_delay: 300000   # максимум 5 минут
                    # service: App\Retry\PaymentRetryStrategy  # кастомная стратегия

        failure_transport: failed

        routing:
            'App\Message\ProcessPayment': async_payments

Symfony использует RecoverableMessageHandlingException для retry и UnrecoverableMessageHandlingException для немедленного перемещения в failure transport. Retry-стратегия настраивается на уровне транспорта, не handler'а.

Failed Jobs: мониторинг и replay

Laravel:

# Список failed jobs
php artisan queue:failed

# Retry конкретного job
php artisan queue:retry 5

# Retry всех
php artisan queue:retry all

# Удалить failed job
php artisan queue:forget 5

# Очистить все failed
php artisan queue:flush

// Кастомная таблица failed_jobs (миграция создаётся автоматически)
// database/migrations/xxxx_create_failed_jobs_table.php

// Программный доступ к failed jobs
$failedJobs = DB::table('failed_jobs')
    ->where('failed_at', '>=', now()->subDay())
    ->orderByDesc('failed_at')
    ->get();

Symfony:

# Список failed messages
php bin/console messenger:failed:show

# Retry конкретного
php bin/console messenger:failed:retry 5

# Retry всех
php bin/console messenger:failed:retry --force

# Удалить
php bin/console messenger:failed:remove 5

Приоритеты и множественные очереди

Laravel: очереди с приоритетами

// Job с указанием очереди
class SendWelcomeEmail implements ShouldQueue
{
    public $queue = 'emails';
}

class ProcessPayment implements ShouldQueue
{
    public $queue = 'payments';
}

class GenerateReport implements ShouldQueue
{
    public $queue = 'reports';
}

// Динамическое назначение очереди
SendWelcomeEmail::dispatch($user)->onQueue('urgent-emails');

# Воркер обрабатывает очереди с приоритетом
# payments первый, потом emails, потом reports
php artisan queue:work --queue=payments,emails,reports

# Выделенные воркеры
php artisan queue:work --queue=payments --timeout=120
php artisan queue:work --queue=emails --timeout=60
php artisan queue:work --queue=reports --timeout=600 --memory=512

# Supervisor конфигурация
[program:laravel-worker-payments]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/artisan queue:work redis --queue=payments --sleep=3 --tries=5 --timeout=120
autostart=true
autorestart=true
numprocs=4
redirect_stderr=true
stdout_logfile=/var/log/worker-payments.log

[program:laravel-worker-emails]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/artisan queue:work redis --queue=emails --sleep=3 --tries=3 --timeout=60
autostart=true
autorestart=true
numprocs=2
redirect_stderr=true
stdout_logfile=/var/log/worker-emails.log

[program:laravel-worker-reports]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/artisan queue:work redis --queue=reports --sleep=10 --tries=1 --timeout=600 --memory=512
autostart=true
autorestart=true
numprocs=1
redirect_stderr=true
stdout_logfile=/var/log/worker-reports.log

Symfony: routing и множественные consumers

# config/packages/messenger.yaml
framework:
    messenger:
        transports:
            async_high:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: high_priority

            async_low:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: low_priority

            async_reports:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                options:
                    queue_name: reports

        routing:
            'App\Message\ProcessPayment': async_high
            'App\Message\SendWelcomeEmail': async_high
            'App\Message\GenerateReport': async_reports
            'App\Message\UpdateSearchIndex': async_low
            'App\Message\CleanupTempFiles': async_low

# Отдельные consumers для каждого транспорта
php bin/console messenger:consume async_high --limit=1000 --time-limit=3600
php bin/console messenger:consume async_low --limit=500 --time-limit=3600
php bin/console messenger:consume async_reports --limit=10 --memory-limit=512M

# Один consumer для нескольких транспортов (с приоритетом)
php bin/console messenger:consume async_high async_low --limit=1000

Middleware: до и после обработки

Laravel: Job Middleware

// app/Jobs/Middleware/RateLimited.php
class RateLimited
{
    public function handle(object $job, callable $next): void
    {
        $key = 'job-rate:' . get_class($job);

        if (RateLimiter::tooManyAttempts($key, 100)) {
            $job->release(60); // вернуть в очередь через 60 секунд
            return;
        }

        RateLimiter::hit($key, 60);
        $next($job);
    }
}

// app/Jobs/Middleware/WithoutOverlapping.php
class WithoutOverlapping
{
    public function __construct(
        private readonly string $lockKey,
        private readonly int $releaseAfter = 300,
    ) {}

    public function handle(object $job, callable $next): void
    {
        $lock = Cache::lock("job-lock:{$this->lockKey}", $this->releaseAfter);

        if (!$lock->get()) {
            $job->release(30);
            return;
        }

        try {
            $next($job);
        } finally {
            $lock->release();
        }
    }
}

// Использование в Job
class SyncInventory implements ShouldQueue
{
    public function middleware(): array
    {
        return [
            new RateLimited(),
            new WithoutOverlapping("inventory-sync-{$this->warehouseId}"),
        ];
    }

    public function handle(): void
    {
        // ...
    }
}

Symfony: Middleware на Message Bus

// src/Middleware/LoggingMiddleware.php
class LoggingMiddleware implements MiddlewareInterface
{
    public function __construct(
        private readonly LoggerInterface $logger,
    ) {}

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $message = $envelope->getMessage();
        $class = get_class($message);

        $this->logger->info("Processing message: {$class}");
        $start = microtime(true);

        try {
            $envelope = $stack->next()->handle($envelope, $stack);
            $duration = round((microtime(true) - $start) * 1000);
            $this->logger->info("Message processed: {$class} ({$duration}ms)");

            return $envelope;
        } catch (\Throwable $e) {
            $this->logger->error("Message failed: {$class} - {$e->getMessage()}");
            throw $e;
        }
    }
}

// src/Middleware/UniqueMessageMiddleware.php
class UniqueMessageMiddleware implements MiddlewareInterface
{
    public function __construct(
        private readonly CacheInterface $cache,
    ) {}

    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $message = $envelope->getMessage();

        if ($message instanceof UniqueMessage) {
            $key = 'msg-unique:' . $message->getUniqueId();

            if ($this->cache->get($key)) {
                $this->cache->delete($key);
                return $envelope; // пропустить дубликат
            }
        }

        return $stack->next()->handle($envelope, $stack);
    }
}

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            messenger.bus.default:
                middleware:
                    - 'App\Middleware\LoggingMiddleware'
                    - 'App\Middleware\UniqueMessageMiddleware'
                    # встроенные middleware автоматически добавлены:
                    # - send_message
                    # - handle_message

Планирование и отложенные задачи

Laravel: Schedule + delayed dispatch

// Отложенный dispatch
SendReminderEmail::dispatch($user)->delay(now()->addHours(24));
SendFollowUp::dispatch($order)->delay(now()->addDays(3));

// Schedule (app/Console/Kernel.php или routes/console.php)
Schedule::job(new CleanupTempFiles)->daily();
Schedule::job(new SyncInventory)->everyFiveMinutes();
Schedule::job(new GenerateDailyReport)->dailyAt('06:00');
Schedule::job(new PruneExpiredTokens)->hourly();

// Условное выполнение
Schedule::job(new SyncExternalApi)
    ->everyTenMinutes()
    ->withoutOverlapping()
    ->runInBackground()
    ->when(fn() => config('services.api.sync_enabled'));

Symfony: delayed messages + scheduler

// Отложенный dispatch через stamp
$this->messageBus->dispatch(
    new SendReminderEmail($userId),
    [new DelayStamp(86400000)] // миллисекунды — 24 часа
);

// Symfony Scheduler (с 6.3+)
// src/Scheduler/DefaultScheduleProvider.php
#[AsSchedule('default')]
class DefaultScheduleProvider implements ScheduleProviderInterface
{
    public function getSchedule(): Schedule
    {
        return (new Schedule())
            ->add(
                RecurringMessage::every('5 minutes', new SyncInventory())
            )
            ->add(
                RecurringMessage::cron('0 6 * * *', new GenerateDailyReport())
            )
            ->add(
                RecurringMessage::every('1 hour', new PruneExpiredTokens())
            );
    }
}

# Запуск scheduler
php bin/console messenger:consume scheduler_default

Батчевая обработка

Laravel: Job Batches

// Создание батча
$batch = Bus::batch([
    new ProcessImage($images[0]),
    new ProcessImage($images[1]),
    new ProcessImage($images[2]),
])
->then(function (Batch $batch) {
    // Все jobs завершились успешно
    Notification::send($user, new ImagesProcessedNotification());
})
->catch(function (Batch $batch, \Throwable $e) {
    // Один из jobs упал
    Log::error("Batch {$batch->id} failed: {$e->getMessage()}");
})
->finally(function (Batch $batch) {
    // Батч завершён (успешно или нет)
    CleanupTempFiles::dispatch($batch->id);
})
->allowFailures()
->name('process-user-images')
->onQueue('images')
->dispatch();

// Мониторинг прогресса
$batch = Bus::findBatch($batchId);
echo $batch->progress(); // 67 (процент)
echo $batch->processedJobs(); // 2
echo $batch->totalJobs; // 3
echo $batch->failedJobs; // 0

// Добавление jobs в существующий батч
$batch->add([
    new ProcessImage($newImage),
]);

Symfony: группировка через stamps

// Symfony не имеет встроенных батчей как Laravel
// Но можно реализовать через stamps и middleware

// src/Stamp/BatchStamp.php
class BatchStamp implements StampInterface
{
    public function __construct(
        public readonly string $batchId,
        public readonly int $totalMessages,
    ) {}
}

// src/MessageHandler/BatchAwareHandler.php
#[AsMessageHandler]
class ProcessImageHandler
{
    public function __construct(
        private readonly CacheInterface $cache,
        private readonly MessageBusInterface $bus,
    ) {}

    public function __invoke(ProcessImage $message): void
    {
        // Обработка изображения
        $this->processImage($message->imagePath);

        // Трекинг прогресса батча
        $batchKey = "batch:{$message->batchId}:completed";
        $completed = $this->cache->get($batchKey, fn() => 0) + 1;
        $this->cache->set($batchKey, $completed, 3600);

        if ($completed >= $message->totalInBatch) {
            $this->bus->dispatch(new BatchCompleted($message->batchId));
        }
    }
}

Laravel здесь выигрывает однозначно: батчи — первоклассная фича с прогрессом, callbacks, мониторингом. В Symfony это приходится строить руками.

Event-driven архитектура

Laravel: Events + Listeners + Queues

// app/Events/OrderPlaced.php
class OrderPlaced
{
    use Dispatchable, SerializesModels;

    public function __construct(
        public readonly Order $order,
    ) {}
}

// app/Listeners/SendOrderConfirmation.php
class SendOrderConfirmation implements ShouldQueue
{
    public $queue = 'emails';

    public function handle(OrderPlaced $event): void
    {
        Mail::to($event->order->email)
            ->send(new OrderConfirmationMail($event->order));
    }
}

// app/Listeners/UpdateInventory.php
class UpdateInventory implements ShouldQueue
{
    public $queue = 'inventory';

    public function handle(OrderPlaced $event): void
    {
        foreach ($event->order->items as $item) {
            Product::where('id', $item->product_id)
                ->decrement('stock', $item->quantity);
        }
    }
}

// app/Listeners/NotifySalesTeam.php
class NotifySalesTeam implements ShouldQueue
{
    public $queue = 'notifications';

    public function handle(OrderPlaced $event): void
    {
        if ($event->order->total > 10000) {
            Notification::route('slack', config('services.slack.sales'))
                ->notify(new BigOrderNotification($event->order));
        }
    }
}

// Dispatch
event(new OrderPlaced($order));
// или
OrderPlaced::dispatch($order);

Symfony: Event Bus + Handlers

// src/Event/OrderPlaced.php
class OrderPlaced
{
    public function __construct(
        public readonly int $orderId,
    ) {}
}

// src/MessageHandler/SendOrderConfirmationHandler.php
#[AsMessageHandler]
class SendOrderConfirmationHandler
{
    public function __invoke(OrderPlaced $event): void
    {
        // отправка подтверждения
    }
}

// src/MessageHandler/UpdateInventoryHandler.php
#[AsMessageHandler]
class UpdateInventoryHandler
{
    public function __invoke(OrderPlaced $event): void
    {
        // обновление склада
    }
}

// src/MessageHandler/NotifySalesTeamHandler.php
#[AsMessageHandler]
class NotifySalesTeamHandler
{
    public function __invoke(OrderPlaced $event): void
    {
        // уведомление
    }
}

framework:
    messenger:
        routing:
            'App\Event\OrderPlaced': async

В Symfony один message автоматически попадает ко всем зарегистрированным handlers. Не нужно явно связывать events и listeners — DI-контейнер делает это через атрибуты.

Тестирование

Laravel: Queue::fake()

// tests/Feature/OrderTest.php
class OrderTest extends TestCase
{
    public function test_placing_order_dispatches_jobs(): void
    {
        Queue::fake();

        $user = User::factory()->create();
        $order = Order::factory()->create(['user_id' => $user->id]);

        $response = $this->actingAs($user)
            ->post("/orders/{$order->id}/place");

        $response->assertStatus(200);

        // Проверить что job был отправлен
        Queue::assertPushed(SendOrderConfirmation::class, function ($job) use ($order) {
            return $job->order->id === $order->id;
        });

        // Проверить что job НЕ был отправлен
        Queue::assertNotPushed(RefundOrder::class);

        // Проверить количество
        Queue::assertPushed(SendOrderConfirmation::class, 1);

        // Проверить очередь
        Queue::assertPushedOn('emails', SendOrderConfirmation::class);
    }

    public function test_job_sends_email(): void
    {
        Mail::fake();

        $order = Order::factory()->create();

        // Выполнить job синхронно
        (new SendOrderConfirmation($order))->handle(app(Mailer::class));

        Mail::assertSent(OrderConfirmationMail::class, function ($mail) use ($order) {
            return $mail->hasTo($order->email);
        });
    }

    public function test_batch_processing(): void
    {
        Bus::fake();

        $images = Image::factory(5)->create();

        $this->post('/images/process', ['ids' => $images->pluck('id')]);

        Bus::assertBatched(function (PendingBatch $batch) {
            return $batch->name === 'process-images'
                && $batch->jobs->count() === 5;
        });
    }
}

Symfony: InMemoryTransport

// tests/Functional/OrderTest.php
class OrderTest extends WebTestCase
{
    public function testPlacingOrderDispatchesMessages(): void
    {
        $client = static::createClient();

        $client->request('POST', '/orders/1/place');

        $this->assertResponseIsSuccessful();

        // Получить сообщения из in-memory транспорта
        $transport = $this->getContainer()
            ->get('messenger.transport.async');

        $messages = $transport->get();

        $this->assertCount(3, $messages); // 3 handlers

        // Проверить конкретное сообщение
        $orderPlaced = $messages[0]->getMessage();
        $this->assertInstanceOf(OrderPlaced::class, $orderPlaced);
        $this->assertEquals(1, $orderPlaced->orderId);
    }
}

# config/packages/test/messenger.yaml
framework:
    messenger:
        transports:
            async:
                dsn: 'in-memory://'

Laravel здесь удобнее: Queue::fake() — одна строка, и все assertions готовы. В Symfony нужно настроить in-memory транспорт и вручную доставать сообщения.

Мониторинг и observability

Laravel: Horizon (для Redis)

composer require laravel/horizon
php artisan horizon:install
php artisan horizon

Horizon — это полноценный дашборд для мониторинга Redis-очередей:

• Real-time метрики: throughput, runtime, failed jobs
• Графики нагрузки по очередям
• Просмотр и retry failed jobs
• Теги для группировки jobs
• Балансировка воркеров между очередями
• Notifications при длинных очередях

// config/horizon.php
'environments' => [
    'production' => [
        'supervisor-payments' => [
            'connection' => 'redis',
            'queue' => ['payments'],
            'balance' => 'auto',
            'minProcesses' => 2,
            'maxProcesses' => 10,
            'tries' => 5,
            'timeout' => 120,
        ],
        'supervisor-default' => [
            'connection' => 'redis',
            'queue' => ['default', 'emails', 'notifications'],
            'balance' => 'auto',
            'minProcesses' => 1,
            'maxProcesses' => 5,
            'tries' => 3,
            'timeout' => 60,
        ],
    ],
],

Symfony: нет встроенного дашборда

Symfony не имеет аналога Horizon. Мониторинг строится через:

• Prometheus + Grafana: кастомные метрики через prometheus/client_php
• Symfony Profiler: в dev-режиме показывает обработанные messages
• messenger:stats: команда для базовой статистики

// src/EventSubscriber/MessengerMetricsSubscriber.php
class MessengerMetricsSubscriber implements EventSubscriberInterface
{
    public function __construct(
        private readonly CollectorRegistry $registry,
    ) {}

    public static function getSubscribedEvents(): array
    {
        return [
            WorkerMessageHandledEvent::class => 'onMessageHandled',
            WorkerMessageFailedEvent::class => 'onMessageFailed',
        ];
    }

    public function onMessageHandled(WorkerMessageHandledEvent $event): void
    {
        $class = get_class($event->getEnvelope()->getMessage());
        $this->registry->getOrRegisterCounter(
            'messenger', 'messages_handled_total', 'Total handled messages', ['message_class']
        )->inc([$class]);
    }

    public function onMessageFailed(WorkerMessageFailedEvent $event): void
    {
        $class = get_class($event->getEnvelope()->getMessage());
        $this->registry->getOrRegisterCounter(
            'messenger', 'messages_failed_total', 'Total failed messages', ['message_class']
        )->inc([$class]);
    }
}

Производительность и бенчмарки

Тесты на одной машине (4 ядра, 16GB RAM), Redis 7, PHP 8.4, 4 воркера.

Throughput (jobs/сек)

*Symfony быстрее на сериализации, потому что передаёт только ID, а не всю модель.

Использование памяти (на воркер)

Symfony чуть экономнее по памяти благодаря более строгому DI и отсутствию SerializesModels. Laravel компенсирует это --memory лимитом и автоматическим рестартом.

Latency (время от dispatch до начала обработки)

Когда что выбирать

Laravel Queues — когда

• Уже на Laravel — нет смысла тянуть Symfony Messenger
• Нужны батчи с прогрессом и callbacks
• Нужен Horizon для мониторинга
• Команда привыкла к Jobs
• Простая архитектура: один handler на задачу
• Redis как основной broker

Symfony Messenger — когда

• Уже на Symfony — нет смысла тянуть Laravel
• Event-driven архитектура: один event → много handlers
• RabbitMQ с продвинутым routing (exchanges, bindings)
• CQRS: разделение command и query шин
• Строгое разделение message и handler (для тестируемости)
• Микросервисная архитектура с разными consumers

Ни то, ни другое — когда

• Меньше 100 jobs в день — достаточно pcntl_fork или cron
• Real-time с WebSocket — смотреть в сторону Centrifugo, Mercure
• Event streaming (миллионы событий) — Kafka
• Нужна exactly-once семантика — ни Laravel, ни Symfony не гарантируют, нужен Kafka/Pulsar

Dead Letter Queue и poison messages

Laravel: failed_jobs таблица

// Автоматическое перемещение в failed_jobs после исчерпания tries
// Кастомный обработчик poison messages
class PoisonPillDetector
{
    public function handle(object $job, callable $next): void
    {
        $payload = $job->payload();
        $attempts = $payload['attempts'] ?? 0;

        // Если job перезапускался > 50 раз за последний час — poison
        $key = 'poison:' . md5($payload['data']['command'] ?? '');
        $recentAttempts = (int) Cache::get($key, 0);

        if ($recentAttempts > 50) {
            Log::critical("Poison message detected", [
                'job' => get_class($job),
                'attempts' => $recentAttempts,
            ]);
            $job->delete();
            return;
        }

        Cache::increment($key);
        Cache::put($key, Cache::get($key), 3600);

        $next($job);
    }
}

Symfony: failure transport

framework:
    messenger:
        failure_transport: failed

        transports:
            failed:
                dsn: 'doctrine://default?queue_name=failed'

# Просмотр failed messages
php bin/console messenger:failed:show

# Retry с фильтрацией
php bin/console messenger:failed:retry --class='App\Message\ProcessPayment'

# Удалить старые (>7 дней)
php bin/console messenger:failed:remove --older-than='7 days'

Итоговый чеклист выбора и внедрения

Выбор системы очередей

• [ ] Определить стек: Laravel → Queues, Symfony → Messenger, смешанный → оценить оба
• [ ] Определить транспорт: Redis (простота), RabbitMQ (routing), SQS (AWS), Database (минимум инфры)
• [ ] Определить паттерн: Job-centric (Laravel) или Event-driven (Symfony)
• [ ] Оценить объём: < 1K jobs/час → database, < 100K → Redis, > 100K → RabbitMQ/Kafka

Проектирование

• [ ] Разделить очереди по приоритету: payments (critical), emails (high), reports (low)
• [ ] Настроить retry: экспоненциальный backoff, max_retries, dead letter queue
• [ ] Сделать jobs идемпотентными — повторный запуск не должен создавать дубликаты
• [ ] Не передавать большие объекты — только ID, загружать из базы в handler

Production

• [ ] Supervisor/systemd для воркеров — автоматический перезапуск при падении
• [ ] Memory limits: --memory=128 (Laravel), --memory-limit=128M (Symfony)
• [ ] Time limits: --timeout (Laravel), --time-limit (Symfony) — для graceful restart
• [ ] Мониторинг: Horizon (Laravel) или Prometheus + Grafana (Symfony)
• [ ] Алерты: failed jobs > порога, queue depth > порога, consumer lag

Тестирование

• [ ] Unit-тесты handlers: логика без очереди
• [ ] Integration-тесты: Queue::fake() / in-memory transport
• [ ] Load-тесты: throughput под нагрузкой
• [ ] Chaos-тесты: поведение при падении Redis/RabbitMQ