Очереди в 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 Queues | Symfony Messenger |
|---|---|---|
| Единица работы | Job (message + handler) | Message + Handler (раздельно) |
| Сериализация | Модели через SerializesModels | Только примитивы (ID, строки) |
| DI | Через method injection в handle() | Через конструктор handler'а |
| Routing | Очередь указывается в Job | Настраивается в messenger.yaml |
| Множественные handlers | Нет (1 job = 1 handler) | Да (1 message → N handlers) |
| Middleware | Job middleware | Middleware на шине |
| Сложность | Низкая | Средняя-высокая |
Транспорты и драйверы
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-мире.
Сравнение транспортов
| Транспорт | Laravel | Symfony | Когда использовать |
|---|---|---|---|
| Redis | Да (основной) | Да | Скорость, простота, до ~50k msg/sec |
| RabbitMQ (AMQP) | Через пакет | Да (основной) | Routing, fan-out, гарантии доставки |
| Database | Да | Да (Doctrine) | Простота, не нужен Redis |
| Amazon SQS | Да | Да | AWS-инфраструктура, serverless |
| Beanstalkd | Да | Да | Легковесный, priority queues |
| Kafka | Через пакет | Через пакет | Event streaming, высокий throughput |
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/сек)
| Сценарий | Laravel Queues | Symfony Messenger |
|---|---|---|
| Простой job (запись в БД) | ~2,800 | ~2,400 |
| Email отправка | ~1,200 | ~1,100 |
| Тяжёлый job (API + DB + cache) | ~180 | ~170 |
| Сериализация большой модели | ~900 | ~1,500* |
*Symfony быстрее на сериализации, потому что передаёт только ID, а не всю модель.
Использование памяти (на воркер)
| Метрика | Laravel | Symfony |
|---|---|---|
| Базовый расход | ~40 MB | ~35 MB |
| После 1000 jobs | ~45 MB | ~38 MB |
| После 10000 jobs | ~55 MB | ~42 MB |
| Memory leak за 24ч | ~15 MB | ~8 MB |
Symfony чуть экономнее по памяти благодаря более строгому DI и отсутствию SerializesModels. Laravel компенсирует это --memory лимитом и автоматическим рестартом.
Latency (время от dispatch до начала обработки)
| Транспорт | Laravel | Symfony |
|---|---|---|
| Redis | 2-5 ms | 3-7 ms |
| Database | 50-200 ms | 50-200 ms |
| RabbitMQ | N/A (пакет) | 1-3 ms |
Когда что выбирать
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
Комментарии (0)