Очереди, события, планировщик

Очереди выносят долгие операции из HTTP-запроса: отправка писем, ресайз картинок, обращения к внешним API, генерация отчётов. Ответ пользователю — мгновенно, работа — асинхронно воркером.

php artisan make:job ProcessPodcast

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

    public function __construct(public Podcast $podcast) {}  // модель сериализуется как id

    public function handle(): void { /* тяжёлая работа */ }
}

ProcessPodcast::dispatch($podcast);
ProcessPodcast::dispatch($podcast)->delay(now()->addMinutes(10))->onQueue('slow');

Запуск обработчика:

php artisan queue:work --queue=high,default

Драйверы: database, redis, sqs, sync (синхронно, для dev/test). Конфиг — config/queue.php, QUEUE_CONNECTION в .env.

На проде воркеры держат под Supervisor; для Redis — панель Horizon. Воркеры долгоживущие: после деплоя — queue:restart.

Вместо десятков cron-записей — одна на сервере:

* * * * * cd /var/www/app && php artisan schedule:run >> /dev/null 2>&1

Задачи описываются в routes/console.php:

Schedule::command('reports:sync')->dailyAt('03:00');
Schedule::command('queue:prune-batches')->daily();
Schedule::job(new CleanupOldFiles)->hourly();
Schedule::call(fn () => DB::table('logs')->where('created_at', '<', now()->subMonth())->delete())
    ->weekly();

// Модификаторы:
->everyFiveMinutes() ->weekdays() ->between('8:00', '17:00')
->timezone('Europe/Moscow')
->withoutOverlapping()        // не запускать, пока предыдущий не завершился (lock)
->onOneServer()               // для кластера — только один сервер (нужен общий cache)
->runInBackground()
->emailOutputOnFailure('admin@example.com')

Отладка: php artisan schedule:list, schedule:test.

Преимущества над cron: расписание — в коде и под git, условия/локи/уведомления из коробки, единая точка входа.

class ProcessPodcast implements ShouldQueue
{
    public $tries = 5;                 // всего попыток
    public $backoff = [10, 60, 300];   // паузы между попытками (сек)
    public $maxExceptions = 3;         // max не-обработанных исключений
    public $timeout = 120;             // лимит времени выполнения

    public function retryUntil(): \DateTime    // альтернатива tries — дедлайн по времени
    {
        return now()->addHours(2);
    }

    public function failed(?Throwable $e): void   // хук после окончательного провала
    {
        Log::error('Job failed', ['e' => $e]);
    }
}

Провалившиеся задачи — в таблице failed_jobs:

php artisan queue:failed          # список
php artisan queue:retry 5         # повторить
php artisan queue:retry all
php artisan queue:flush           # очистить

Нюансы:

• идемпотентность: задача может выполниться больше одного раза (ShouldBeUnique — защита от дублей в очереди);
• модели сериализуются по id (SerializesModels) — если запись удалили до выполнения, джоба получит ModelNotFoundException (можно deleteWhenMissingModels);
• исключения в middleware() джобы, ->onConnection('redis'), цепочки Bus::chain([...]) и батчи Bus::batch([...]).

События развязывают код: «что-то произошло» → реакции в слушателях.

php artisan make:event OrderShipped
php artisan make:listener SendShipmentNotification --event=OrderShipped

class OrderShipped
{
    use Dispatchable;
    public function __construct(public Order $order) {}
}

class SendShipmentNotification
{
    public function handle(OrderShipped $event): void
    {
        Mail::to($event->order->user)->send(...);
    }
}

Регистрация (Laravel 11+ — автообнаружение по type-hint'ам; явно — в EventServiceProvider или Event::listen):

Event::listen(OrderShipped::class, SendShipmentNotification::class);
event(new OrderShipped($order));            // или OrderShipped::dispatch($order);

Ключевые возможности:

• слушатель может быть queued (implements ShouldQueue) — реакция уйдёт в очередь;
• Event::fake() в тестах: Event::assertDispatched(OrderShipped::class);
• подписчики (Event::subscribe) — один класс на много событий;
• события моделей (created/updated) — частный случай.

События ≠ очереди: событие — факт, очередь — способ асинхронно выполнить реакцию.

Notification — абстракция «уведомить пользователя» поверх каналов (mail, database, broadcast, SMS через Vonage, Slack, кастомные):

php artisan make:notification InvoicePaid

class InvoicePaid extends Notification implements ShouldQueue
{
    public function __construct(public Invoice $invoice) {}

    public function via(object $notifiable): array
    {
        return ['mail', 'database'];          // каналы
    }

    public function toMail(object $notifiable): MailMessage
    {
        return (new MailMessage)
            ->subject('Счёт оплачен')
            ->line("Сумма: {$this->invoice->total}")
            ->action('Открыть', url("/invoices/{$this->invoice->id}"));
    }

    public function toArray(object $notifiable): array
    {
        return ['invoice_id' => $this->invoice->id];   // в таблицу notifications
    }
}

$user->notify(new InvoicePaid($invoice));
Notification::send($users, new InvoicePaid($invoice));   // группе

Модель использует трейт Notifiable. Красивые письма — Markdown mailables / MailMessage с темой. Для сложных писем отдельно есть Mailable (php artisan make:mail). Notification::fake() + assertSentTo() — в тестах.

Horizon — панель управления и оркестратор воркеров для очередей на Redis.

Что даёт:

• дашборд: пропускная способность, время выполнения, failed jobs, ретраи;
• автобалансировка: стратегии simple, auto (подстраивает число процессов под нагрузку);
• конфигурация воркеров кодом (config/horizon.php): сколько процессов на какую очередь, таймауты, tries;
• мониторинг тегов: следить за джобами конкретного пользователя/заказа (Horizon::tag);
• метрики и уведомления при «зависших» очередях.

'environments' => [
    'production' => [
        'supervisor-1' => [
            'connection' => 'redis',
            'queue' => ['high', 'default'],
            'balance' => 'auto',
            'maxProcesses' => 10,
        ],
    ],
],

php artisan horizon    # вместо queue:work; под Supervisor — один процесс horizon

Когда нужен: Redis-очереди + нагрузка/важность мониторинга. Для database-драйвера или мелких проектов — избыточен, хватает queue:work под Supervisor.