Урок 03

Laravel Jobs: углублённо

Класс Job

Всё, что можно настроить:

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

    public int $tries = 5;
    public int $maxExceptions = 3;
    public int $timeout = 120;
    public bool $failOnTimeout = true;
    public array $backoff = [10, 30, 60];

    public function __construct(
        public Podcast $podcast,
    ) {}

    public function handle(): void { }

    public function middleware(): array { return []; }

    public function retryUntil(): DateTime { }

    public function failed(\Throwable $e): void { }

    public function uniqueId(): string { }

    public function tags(): array { }
}

Конструктор

Только данные:

Скаляры — string, int, bool, array.
Модели — через SerializesModels.
DTO — если implements Serializable или readonly с публичными полями.

⚠️ Нельзя: Closure, PDO, resource.

Модели в конструкторе

SerializesModels сохраняет ID.

При загрузке — модель перезагружается из БД.

Что даёт:

Свежие данные.
Небольшой размер payload.
⚠️ Если модель удалена — ModelNotFoundException.

Решение:

public function __construct(
    public Podcast $podcast,
) {}

public function handle(): void
{
    if (!$this->podcast->exists) {
        return;   // молча
    }

    // ...
}

Или deleteWhenMissingModels = true.

Восстановление связей

Если модель удалена:

public function handle(): void
{
    $podcast = Podcast::withTrashed()->find($this->podcast->id);

    if (!$podcast) return;

    // ...
}

Очереди

Разные приоритеты:

ProcessPodcast::dispatch($podcast)->onQueue('high');
SendEmail::dispatch($user)->onQueue('emails');

Worker с несколькими:

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

Что даёт: сначала high, потом default, потом low.

Задержка

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

Момент: задача не выполнится раньше.

Используется для: напоминания, отложенные email, retry после сбоя.

Connection

Разные брокеры для разных задач:

ProcessPodcast::dispatch($podcast)->onConnection('redis');
SendEmail::dispatch($user)->onConnection('sqs');

Sync — для тестов

// В тестах
Queue::fake();

ProcessPodcast::dispatch($podcast);

Queue::assertPushed(ProcessPodcast::class);

Или явно:

ProcessPodcast::dispatchSync($podcast);

Chaining

Цепочка задач:

ProcessPodcast::withChain([
    new OptimizePodcast,
    new ReleasePodcast,
])->dispatch($podcast);

Что происходит:

ProcessPodcast.
Если успешно → OptimizePodcast.
Если успешно → ReleasePodcast.

Если упадёт — следующая не запустится.

Batches

Группа задач с общим завершением:

use Illuminate\Bus\Batch;

$batch = Bus::batch([
    new ProcessPodcast($podcast1),
    new ProcessPodcast($podcast2),
    new ProcessPodcast($podcast3),
])
    ->then(function (Batch $batch) {
        // все успешно
    })
    ->catch(function (Batch $batch, Throwable $e) {
        // первый fail
    })
    ->finally(function (Batch $batch) {
        // всегда
    })
    ->name('Podcast processing')
    ->dispatch();

$batch->id;
$batch->progress();   // 0-100
$batch->finished();
$batch->cancelled();

Когда: массовая обработка с общим завершением.

Отмена

$batch->cancel();

Что даёт: задачи не запускаются, отменяются.

Progress

$batch->progress();
// 45

Процент завершённых.

Методы внутри Job с batch

public function handle(): void
{
    // Проверить, что batch не отменён
    if ($this->batch()?->cancelled()) {
        return;
    }

    // ...

    // Добавить новую задачу
    $this->batch()->add(new AnotherJob);
}

События Batch

BatchStarted
BatchFinished
BatchCancelled
BatchDispatched

Пример:

Event::listen(BatchFinished::class, function ($event) {
    Log::info('Batch done', ['id' => $event->batch->id]);
});

Unique — предотвращение дублей

class ProcessPodcast implements ShouldQueue, ShouldBeUnique
{
    public $uniqueFor = 3600;   // 1 час

    public function uniqueId(): string
    {
        return (string) $this->podcast->id;
    }
}

Что даёт: если такой же Job уже в очереди — новый не добавится.

Через middleware:

public function middleware(): array
{
    return [(new WithoutOverlapping($this->podcast->id))->dontRelease()];
}

Rate limiting

public function middleware(): array
{
    return [new RateLimited('podcasts')];
}

// В AppServiceProvider
RateLimiter::for('podcasts', fn () => Limit::perMinute(10));

Что даёт: не более 10 задач в минуту.

Middleware Skipping

Пропустить если:

public function middleware(): array
{
    return [
        (new SkipIf(fn () => $this->podcast->trashed())),
        (new SkipUnless(fn () => $this->podcast->published)),
    ];
}

retryUntil

public function retryUntil(): DateTime
{
    return now()->addHours(2);
}

Что даёт: пытаться 2 часа, потом сдаться.

Практика: обработка видео

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

    public int $tries = 3;
    public int $timeout = 600;   // 10 минут
    public array $backoff = [60, 300, 900];

    public function __construct(
        public Video $video,
    ) {}

    public function handle(): void
    {
        $video->update(['status' => 'processing']);

        try {
            $result = FFmpeg::convert($this->video);

            $this->video->update([
                'status' => 'ready',
                'url' => $result->url,
                'duration' => $result->duration,
            ]);

            NotifyUploader::dispatch($this->video->user, $this->video);
        } catch (\Throwable $e) {
            $this->video->update(['status' => 'failed']);
            throw $e;
        }
    }

    public function failed(\Throwable $e): void
    {
        Log::error('Video processing failed', [
            'video_id' => $this->video->id,
            'error' => $e->getMessage(),
        ]);

        Mail::to($this->video->user)->send(
            new VideoFailedNotification($this->video)
        );
    }

    public function middleware(): array
    {
        return [(new WithoutOverlapping($this->video->id))->releaseAfter(60)];
    }

    public function tags(): array
    {
        return ['video', 'video:' . $this->video->id];
    }
}

Dispatch

ProcessVideo::dispatch($video);

// Или с цепочкой
ProcessVideo::withChain([
    new GenerateThumbnail($video),
    new NotifySubscribers($video),
])->dispatch($video);

Частые ошибки

1. Closure в конструкторе:

public function __construct(public \Closure $callback) { }
// Serialization of 'Closure' is not allowed

Решение: имя метода или отдельный класс.

2. Модель удалена до обработки:

ModelNotFoundException

Решение: deleteWhenMissingModels = true.

3. Слишком длинный timeout:

public int $timeout = 3600;   // 1 час
// Worker убьёт раньше

Решение: queue:work --timeout=3700.

4. $tries меньше, чем попыток:

public int $tries = 3;
public array $backoff = [10, 30, 60, 120, 240];   // 5 задержек

5. Unique ID пустой:

public function uniqueId(): string { return ''; }
// все задачи считаются уникальными

Итог

$tries, $timeout, $backoff, $maxExceptions — настройки.
failed — при провале.
middleware — RateLimited, WithoutOverlapping, SkipIf, SkipUnless.
Chaining — цепочки.
Batches — группы с завершением.
Unique — без дублей.
Tags — для Horizon.
retryUntil — до какого времени.

Дальше — драйверы очередей.