Черги · Laravel

Під час створення вебзастосунку у вас можуть бути завдання, як-от розбір і збереження завантаженого CSV-файлу, які виконуються надто довго для типового вебзапиту. На щастя, Laravel дозволяє легко створювати завдання (job) у чергах, що можуть оброблятися у фоні. Переносячи тривалі за часом завдання в чергу, ваш застосунок може відповідати на вебзапити блискавично швидко й забезпечувати кращий досвід для ваших клієнтів.

Черги Laravel надають уніфікований API для роботи з чергами поверх різних бекендів, як-от Amazon SQS, Redis або навіть реляційна база даних.

Параметри конфігурації черг Laravel зберігаються у файлі конфігурації config/queue.php вашого застосунку. У цьому файлі ви знайдете конфігурації зʼєднань для кожного з драйверів черг, що входять до складу фреймворку, зокрема драйверів database, Amazon SQS, Redis та Beanstalkd, а також синхронного драйвера, який виконуватиме завдання негайно (для використання під час розробки або тестування). Також включено драйвер черги null, який відкидає завдання з черги.

Примітка. Laravel Horizon — це красива панель керування й система конфігурації для ваших черг на базі Redis. Перегляньте повну документацію Horizon для отримання додаткової інформації.

Зʼєднання проти черг

Перш ніж починати роботу з чергами Laravel, важливо зрозуміти різницю між «зʼєднаннями» (connections) і «чергами» (queues). У вашому файлі конфігурації config/queue.php є масив конфігурації connections. Цей параметр визначає зʼєднання з бекенд-сервісами черг, як-от Amazon SQS, Beanstalk або Redis. Однак будь-яке окреме зʼєднання черги може мати кілька «черг», які можна уявити як різні стеки або купи завдань у черзі.

Зауважте, що кожен приклад конфігурації зʼєднання у файлі конфігурації queue містить атрибут queue. Це черга за замовчуванням, до якої диспетчеризуватимуться завдання, коли вони надсилаються до відповідного зʼєднання. Іншими словами, якщо ви диспетчеризуєте завдання без явного визначення, до якої черги його слід надіслати, завдання буде розміщено в черзі, визначеній в атрибуті queue конфігурації зʼєднання:

use App\Jobs\ProcessPodcast;

// Це завдання надсилається до черги за замовчуванням зʼєднання за замовчуванням...
ProcessPodcast::dispatch();

// Це завдання надсилається до черги "emails" зʼєднання за замовчуванням...
ProcessPodcast::dispatch()->onQueue('emails');

Деяким застосункам може ніколи не знадобитися надсилати завдання до кількох черг — натомість вони віддають перевагу одній простій черзі. Однак надсилання завдань до кількох черг може бути особливо корисним для застосунків, які хочуть пріоритизувати або сегментувати обробку завдань, оскільки обробник черги Laravel дозволяє вказати, які черги він має обробляти за пріоритетом. Наприклад, якщо ви надсилаєте завдання до черги high, ви можете запустити обробник, який надає їм вищий пріоритет обробки:

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

Нотатки щодо драйверів і передумови

База даних

Щоб використовувати драйвер черги database, вам знадобиться таблиця бази даних для зберігання завдань. Зазвичай вона включена до стандартної міграції бази даних Laravel 0001_01_01_000002_create_jobs_table.php; однак, якщо ваш застосунок не містить цієї міграції, ви можете скористатися Artisan-командою make:queue-table, щоб її створити:

php artisan make:queue-table

php artisan migrate

Redis

Щоб використовувати драйвер черги redis, вам слід налаштувати зʼєднання з базою даних Redis у вашому файлі конфігурації config/database.php.

Попередження. Параметри Redis serializer і compression не підтримуються драйвером черги redis.

Redis Cluster

Якщо ваше зʼєднання черги Redis використовує Redis Cluster, імена ваших черг мають містити key hash tag. Це потрібно для того, щоб гарантувати, що всі ключі Redis для певної черги потраплять в один хеш-слот:

'redis' => [
    'driver' => 'redis',
    'connection' => env('REDIS_QUEUE_CONNECTION', 'default'),
    'queue' => env('REDIS_QUEUE', '{default}'),
    'retry_after' => env('REDIS_QUEUE_RETRY_AFTER', 90),
    'block_for' => null,
    'after_commit' => false,
],

Блокування

Використовуючи чергу Redis, ви можете застосувати параметр конфігурації block_for, щоб указати, як довго драйвер має чекати на появу доступного завдання, перш ніж перейти до наступної ітерації циклу обробника й повторно опитати базу даних Redis.

Налаштування цього значення відповідно до навантаження вашої черги може бути ефективнішим, ніж постійне опитування бази даних Redis на предмет нових завдань. Наприклад, ви можете встановити значення 5, щоб указати, що драйвер має блокуватися на пʼять секунд, очікуючи, поки завдання стане доступним:

'redis' => [
    'driver' => 'redis',
    'connection' => env('REDIS_QUEUE_CONNECTION', 'default'),
    'queue' => env('REDIS_QUEUE', 'default'),
    'retry_after' => env('REDIS_QUEUE_RETRY_AFTER', 90),
    'block_for' => 5,
    'after_commit' => false,
],

Попередження. Встановлення block_for у 0 призведе до того, що обробники черги блокуватимуться нескінченно, доки не зʼявиться доступне завдання. Це також завадить обробці сигналів на кшталт SIGTERM, доки не буде оброблено наступне завдання.

Сховище переповнення SQS

Amazon SQS обмежує максимальний розмір корисного навантаження (payload) повідомлення в черзі. Якщо вам потрібно диспетчеризувати завдання з корисним навантаженням, яке може перевищувати цей ліміт, ви можете налаштувати Laravel зберігати завеликі корисні навантаження SQS у сховищі кешу й надсилати через SQS лише вказівник на них. Щоб увімкнути цю можливість, додайте масив overflow до конфігурації вашого зʼєднання черги SQS:

'sqs' => [
    'driver' => 'sqs',
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'prefix' => env('SQS_PREFIX', 'https://sqs.us-east-1.amazonaws.com/your-account-id'),
    'queue' => env('SQS_QUEUE', 'default'),
    'suffix' => env('SQS_SUFFIX'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'after_commit' => false,
    'overflow' => [
        'enabled' => env('SQS_OVERFLOW_ENABLED', false),
        'store' => env('SQS_OVERFLOW_STORE'),
        'always' => false,
        'delete_after_processing' => true,
        'flush_on_clear' => env('SQS_OVERFLOW_FLUSH_ON_CLEAR', false),
    ],
],

Коли сховище переповнення ввімкнено, Laravel зберігатиме корисні навантаження розміром щонайменше 1 МБ у налаштованому сховищі кешу. Якщо параметр always має значення true, кожне корисне навантаження SQS зберігатиметься у сховищі кешу незалежно від його розміру. Оскільки завданням у черзі потрібно буде отримувати свої корисні навантаження зі сховища кешу під час обробки, вам слід обрати сховище, яке здатне зберігати ці корисні навантаження, доки ваші обробники їх не опрацюють. За замовчуванням збережені корисні навантаження видаляються після того, як їхні завдання були успішно оброблені й видалені з SQS.

Якщо параметр flush_on_clear має значення true, налаштоване сховище кешу переповнення буде очищено, коли команда queue:clear очищає чергу SQS. Оскільки очищення сховища кешу може видалити всі елементи з цього сховища, вам слід налаштувати сховище переповнення SQS на використання окремого виділеного сховища кешу, коли ви вмикаєте цей параметр.

Передумови інших драйверів

Для перелічених драйверів черг потрібні такі залежності. Ці залежності можна встановити через менеджер пакетів Composer:

• Amazon SQS: aws/aws-sdk-php ~3.0
• Beanstalkd: pda/pheanstalk ~5.0
• Redis: predis/predis ~3.0 або PHP-розширення phpredis
• MongoDB: mongodb/laravel-mongodb

Створення завдань

Генерація класів завдань

За замовчуванням усі завдання вашого застосунку, придатні для черг, зберігаються в теці app/Jobs. Якщо тека app/Jobs не існує, її буде створено, коли ви запустите Artisan-команду make:job:

php artisan make:job ProcessPodcast

Згенерований клас реалізовуватиме інтерфейс Illuminate\Contracts\Queue\ShouldQueue, вказуючи Laravel, що завдання слід відправити в чергу для асинхронного виконання.

Примітка. Заготовки (stubs) завдань можна кастомізувати за допомогою публікації заготовок.

Структура класу

Класи завдань дуже прості — зазвичай вони містять лише метод handle, який викликається під час обробки завдання чергою. Для початку погляньмо на приклад класу завдання. У цьому прикладі ми уявимо, що керуємо сервісом публікації подкастів і маємо обробити завантажені файли подкастів перед їх публікацією:

<?php

namespace App\Jobs;

use App\Models\Podcast;
use App\Services\AudioProcessor;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public Podcast $podcast,
    ) {}

    /**
     * Execute the job.
     */
    public function handle(AudioProcessor $processor): void
    {
        // Обробка завантаженого подкасту...
    }
}

У цьому прикладі зверніть увагу, що ми змогли передати модель Eloquent безпосередньо в конструктор завдання в черзі. Завдяки трейту Queueable, який використовує це завдання, моделі Eloquent та їхні завантажені звʼязки будуть коректно серіалізовані й десеріалізовані під час обробки завдання.

Якщо ваше завдання в черзі приймає модель Eloquent у своєму конструкторі, у чергу буде серіалізовано лише ідентифікатор моделі. Коли завдання фактично обробляється, система черг автоматично повторно отримає повний екземпляр моделі та її завантажені звʼязки з бази даних. Такий підхід до серіалізації моделей дозволяє надсилати драйверу черги значно менші корисні навантаження завдань.

Впровадження залежностей у метод handle

Метод handle викликається під час обробки завдання чергою. Зверніть увагу, що ми можемо вказувати типи залежностей у методі handle завдання. Контейнер сервісів Laravel автоматично впроваджує ці залежності.

Якщо ви хочете отримати повний контроль над тим, як контейнер впроваджує залежності в метод handle, ви можете скористатися методом контейнера bindMethod. Метод bindMethod приймає колбек, який отримує завдання і контейнер. У межах колбеку ви вільні викликати метод handle як забажаєте. Зазвичай цей метод слід викликати з методу boot вашого сервіс-провайдера App\Providers\AppServiceProvider:

use App\Jobs\ProcessPodcast;
use App\Services\AudioProcessor;
use Illuminate\Contracts\Foundation\Application;

$this->app->bindMethod([ProcessPodcast::class, 'handle'], function (ProcessPodcast $job, Application $app) {
    return $job->handle($app->make(AudioProcessor::class));
});

Попередження. Бінарні дані, як-от сирий вміст зображення, слід пропускати через функцію base64_encode, перш ніж передавати їх у завдання в черзі. Інакше завдання може некоректно серіалізуватися в JSON під час розміщення в черзі.

Звʼязки в чергах

Оскільки всі завантажені звʼязки моделей Eloquent також серіалізуються, коли завдання потрапляє в чергу, серіалізований рядок завдання іноді може стати доволі великим. Крім того, коли завдання десеріалізується і звʼязки моделей повторно отримуються з бази даних, вони отримуватимуться повністю. Будь-які попередні обмеження звʼязку, застосовані до моделі перед її серіалізацією в процесі постановки завдання в чергу, не будуть застосовані під час десеріалізації завдання. Тому, якщо ви хочете працювати з підмножиною певного звʼязку, вам слід повторно накласти обмеження на цей звʼязок у вашому завданні в черзі.

Або, щоб запобігти серіалізації звʼязків, ви можете викликати метод withoutRelations на моделі під час встановлення значення властивості. Цей метод поверне екземпляр моделі без її завантажених звʼязків:

/**
 * Create a new job instance.
 */
public function __construct(
    Podcast $podcast,
) {
    $this->podcast = $podcast->withoutRelations();
}

Якщо вам потрібно видалити лише конкретні звʼязки, зберігши решту, ви можете скористатися методом withoutRelation:

$this->podcast = $podcast->withoutRelation('comments');

Якщо ви використовуєте просування властивостей конструктора PHP і хочете вказати, що звʼязки моделі Eloquent не слід серіалізувати, ви можете скористатися атрибутом WithoutRelations:

use Illuminate\Queue\Attributes\WithoutRelations;

/**
 * Create a new job instance.
 */
public function __construct(
    #[WithoutRelations]
    public Podcast $podcast,
) {}

Для зручності, якщо ви хочете серіалізувати всі моделі без звʼязків, ви можете застосувати атрибут WithoutRelations до всього класу замість того, щоб застосовувати атрибут до кожної моделі:

<?php

namespace App\Jobs;

use App\Models\DistributionPlatform;
use App\Models\Podcast;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\WithoutRelations;

#[WithoutRelations]
class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public Podcast $podcast,
        public DistributionPlatform $platform,
    ) {}
}

Якщо завдання отримує колекцію або масив моделей Eloquent замість однієї моделі, звʼязки моделей у цій колекції не будуть відновлені під час десеріалізації та виконання завдання. Це зроблено для того, щоб запобігти надмірному використанню ресурсів у завданнях, які працюють із великою кількістю моделей.

Унікальні завдання

Попередження. Унікальні завдання потребують драйвера кешу, який підтримує блокування. Наразі атомарні блокування підтримують драйвери кешу memcached, redis, dynamodb, database, file і array.

Попередження. Обмеження унікальності завдань не застосовуються до завдань у пакетах (batches).

Іноді вам може знадобитися гарантувати, що в черзі в будь-який момент часу перебуває лише один екземпляр конкретного завдання. Це можна зробити, реалізувавши інтерфейс ShouldBeUnique у вашому класі завдання. Цей інтерфейс не вимагає визначати жодних додаткових методів у вашому класі:

<?php

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUnique;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    // ...
}

У наведеному вище прикладі завдання UpdateSearchIndex є унікальним. Тож завдання не буде диспетчеризовано, якщо інший екземпляр цього завдання вже перебуває в черзі й не завершив обробку.

У певних випадках ви можете захотіти визначити конкретний «ключ», який робить завдання унікальним, або вказати таймаут, після якого завдання перестає бути унікальним. Щоб цього досягти, ви можете скористатися атрибутом UniqueFor і визначити метод uniqueId у вашому класі завдання:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Queue\Attributes\UniqueFor;

#[UniqueFor(3600)]
class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    /**
     * The product instance.
     *
     * @var \App\Models\Product
     */
    public $product;

    /**
     * Get the unique ID for the job.
     */
    public function uniqueId(): string
    {
        return $this->product->id;
    }
}

У наведеному вище прикладі завдання UpdateSearchIndex є унікальним за ID продукту. Тож будь-які нові диспетчеризації цього завдання з тим самим ID продукту ігноруватимуться, доки наявне завдання не завершить обробку. Крім того, якщо наявне завдання не буде оброблено протягом однієї години, блокування унікальності буде звільнено й інше завдання з тим самим унікальним ключем можна буде диспетчеризувати в чергу.

Попередження. Якщо ваш застосунок диспетчеризує завдання з кількох вебсерверів або контейнерів, вам слід переконатися, що всі ваші сервери спілкуються з одним і тим самим центральним сервером кешу, щоб Laravel міг точно визначити, чи є завдання унікальним.

Збереження унікальності завдань до початку обробки

За замовчуванням унікальні завдання «розблоковуються» після того, як завдання завершує обробку або вичерпує всі свої спроби повтору. Однак можуть бути ситуації, коли ви хочете, щоб ваше завдання розблоковувалося безпосередньо перед його обробкою. Щоб цього досягти, ваше завдання має реалізовувати контракт ShouldBeUniqueUntilProcessing замість контракту ShouldBeUnique:

<?php

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUniqueUntilProcessing
{
    // ...
}

Блокування унікальних завдань

Під капотом, коли завдання ShouldBeUnique диспетчеризується, Laravel намагається отримати блокування з ключем uniqueId. Якщо блокування вже утримується, завдання не диспетчеризується. Це блокування звільняється, коли завдання завершує обробку або вичерпує всі свої спроби повтору. За замовчуванням Laravel використовуватиме драйвер кешу за замовчуванням для отримання цього блокування. Однак, якщо ви бажаєте використовувати інший драйвер для отримання блокування, ви можете визначити метод uniqueVia, який повертає драйвер кешу, що має використовуватися:

use Illuminate\Contracts\Cache\Repository;
use Illuminate\Support\Facades\Cache;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    // ...

    /**
     * Get the cache driver for the unique job lock.
     */
    public function uniqueVia(): Repository
    {
        return Cache::driver('redis');
    }
}

Примітка. Якщо вам потрібно лише обмежити паралельну обробку завдання, скористайтеся натомість job middleware WithoutOverlapping.

Завдання з дебаунсом

Іноді вам може знадобитися гарантувати, що коли одне й те саме завдання диспетчеризується багато разів у короткому проміжку часу, фактично виконається лише остання диспетчеризація. Це можна зробити, додавши атрибут DebounceFor до вашого завдання:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\DebounceFor;

#[DebounceFor(30)]
class UpdateSearchIndex implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct(public int $productId)
    {
    }

    /**
     * Get the debounce ID for the job.
     */
    public function debounceId(): string
    {
        return (string) $this->productId;
    }
}

У наведеному вище прикладі багаторазова диспетчеризація UpdateSearchIndex для того самого продукту протягом 30 секунд застосує дебаунс до завдання так, що виконається лише остання диспетчеризація.

Якщо ви хочете обмежити, наскільки довго може відкладатися часто передиспетчеризоване завдання, ви можете передати аргумент maxWait до атрибута DebounceFor:

#[DebounceFor(30, maxWait: 120)]
class UpdateSearchIndex implements ShouldQueue
{
    use Queueable;

    // ...
}

Ви можете налаштувати сховище кешу, яке використовується для відстеження дебаунсу, визначивши метод debounceVia у вашому завданні:

use Illuminate\Contracts\Cache\Repository;
use Illuminate\Support\Facades\Cache;

public function debounceVia(): Repository
{
    return Cache::driver('redis');
}

Якщо завдання з дебаунсом витісняється новішою диспетчеризацією, Laravel відправить подію Illuminate\Queue\Events\JobDebounced і видалить витіснене завдання з черги.

Попередження. Завдання з дебаунсом і унікальні завдання є взаємовиключними. Завдання, що використовує атрибут DebounceFor, не повинно реалізовувати ShouldBeUnique.

Попередження. Якщо ваш застосунок диспетчеризує завдання з дебаунсом із кількох вебсерверів або контейнерів, вам слід переконатися, що всі ваші сервери спілкуються з одним і тим самим центральним сервером кешу.

Зашифровані завдання

Laravel дозволяє забезпечити приватність і цілісність даних завдання за допомогою шифрування. Для початку просто додайте інтерфейс ShouldBeEncrypted до класу завдання. Щойно цей інтерфейс додано до класу, Laravel автоматично шифруватиме ваше завдання перед відправленням його в чергу:

<?php

use Illuminate\Contracts\Queue\ShouldBeEncrypted;
use Illuminate\Contracts\Queue\ShouldQueue;

class UpdateSearchIndex implements ShouldQueue, ShouldBeEncrypted
{
    // ...
}

Job Middleware

Job middleware дозволяють обгортати власну логіку навколо виконання завдань у черзі, зменшуючи кількість шаблонного коду в самих завданнях. Наприклад, розгляньте наступний метод handle, який використовує можливості обмеження частоти (rate limiting) Redis у Laravel, щоб дозволити обробку лише одного завдання кожні пʼять секунд:

use Illuminate\Support\Facades\Redis;

/**
 * Execute the job.
 */
public function handle(): void
{
    Redis::throttle('key')->block(0)->allow(1)->every(5)->then(function () {
        info('Lock obtained...');

        // Обробка завдання...
    }, function () {
        // Не вдалося отримати блокування...

        return $this->release(5);
    });
}

Хоча цей код і робочий, реалізація методу handle стає «шумною», оскільки захаращена логікою обмеження частоти Redis. Крім того, цю логіку обмеження частоти доведеться дублювати для будь-яких інших завдань, для яких ми хочемо обмежити частоту. Замість обмеження частоти в методі handle ми могли б визначити job middleware, який обробляє обмеження частоти:

<?php

namespace App\Jobs\Middleware;

use Closure;
use Illuminate\Support\Facades\Redis;

class RateLimited
{
    /**
     * Process the queued job.
     *
     * @param  \Closure(object): void  $next
     */
    public function handle(object $job, Closure $next): void
    {
        Redis::throttle('key')
            ->block(0)->allow(1)->every(5)
            ->then(function () use ($job, $next) {
                // Блокування отримано...

                $next($job);
            }, function () use ($job) {
                // Не вдалося отримати блокування...

                $job->release(5);
            });
    }
}

Як бачите, подібно до middleware маршрутів, job middleware отримують завдання, що обробляється, і колбек, який слід викликати, щоб продовжити обробку завдання.

Ви можете згенерувати новий клас job middleware за допомогою Artisan-команди make:job-middleware. Після створення job middleware їх можна прикріпити до завдання, повернувши їх із методу middleware завдання. Цей метод не існує в завданнях, згенерованих Artisan-командою make:job, тому вам потрібно буде вручну додати його до вашого класу завдання:

use App\Jobs\Middleware\RateLimited;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited];
}

Примітка. Job middleware також можна призначати слухачам подій у чергах, mailable-класам і сповіщенням.

Обмеження частоти

Хоча ми щойно продемонстрували, як написати власний job middleware для обмеження частоти, Laravel насправді містить middleware обмеження частоти, який ви можете використати для обмеження частоти завдань. Подібно до обмежувачів частоти маршрутів, обмежувачі частоти завдань визначаються за допомогою методу for фасаду RateLimiter.

Наприклад, ви можете захотіти дозволити користувачам створювати резервні копії своїх даних раз на годину, не накладаючи такого обмеження на преміум-клієнтів. Щоб цього досягти, ви можете визначити RateLimiter у методі boot вашого AppServiceProvider:

use Illuminate\Cache\RateLimiting\Limit;
use Illuminate\Support\Facades\RateLimiter;

/**
 * Bootstrap any application services.
 */
public function boot(): void
{
    RateLimiter::for('backups', function (object $job) {
        return $job->user->vipCustomer()
            ? Limit::none()
            : Limit::perHour(1)->by($job->user->id);
    });
}

У наведеному вище прикладі ми визначили погодинне обмеження частоти; однак ви можете легко визначити обмеження частоти на основі хвилин за допомогою методу perMinute. Крім того, ви можете передати будь-яке значення, яке забажаєте, до методу by обмеження частоти; однак це значення найчастіше використовується для сегментації обмежень частоти за клієнтом:

return Limit::perMinute(50)->by($job->user->id);

Щойно ви визначили своє обмеження частоти, ви можете прикріпити обмежувач частоти до вашого завдання за допомогою middleware Illuminate\Queue\Middleware\RateLimited. Щоразу, коли завдання перевищує обмеження частоти, цей middleware повертатиме завдання назад у чергу з відповідною затримкою на основі тривалості обмеження частоти:

use Illuminate\Queue\Middleware\RateLimited;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited('backups')];
}

Повернення завдання з обмеженою частотою назад у чергу все одно збільшуватиме загальну кількість спроб (attempts) завдання. Ви можете відповідно налаштувати атрибути Tries і MaxExceptions у вашому класі завдання. Або ви можете скористатися методом retryUntil, щоб визначити проміжок часу, після якого завдання більше не слід намагатися виконати.

За допомогою методу releaseAfter ви також можете вказати кількість секунд, які мають минути, перш ніж повернене завдання буде спробувано знову:

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->releaseAfter(60)];
}

Якщо ви не хочете, щоб завдання повторювалося, коли до нього застосовано обмеження частоти, ви можете скористатися методом dontRelease:

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->dontRelease()];
}

Обмеження частоти з Redis

Якщо ви використовуєте Redis, ви можете скористатися middleware Illuminate\Queue\Middleware\RateLimitedWithRedis, який точно налаштований під Redis і ефективніший за базовий middleware обмеження частоти:

use Illuminate\Queue\Middleware\RateLimitedWithRedis;

public function middleware(): array
{
    return [new RateLimitedWithRedis('backups')];
}

Метод connection можна використовувати, щоб указати, яке зʼєднання Redis має використовувати middleware:

return [(new RateLimitedWithRedis('backups'))->connection('limiter')];

Запобігання накладанню завдань

Laravel містить middleware Illuminate\Queue\Middleware\WithoutOverlapping, який дозволяє запобігти накладанню завдань на основі довільного ключа. Це може бути корисним, коли завдання в черзі змінює ресурс, який має змінюватися лише одним завданням за раз.

Наприклад, уявімо, що у вас є завдання в черзі, яке оновлює кредитний рейтинг користувача, і ви хочете запобігти накладанню завдань оновлення кредитного рейтингу для того самого ID користувача. Щоб цього досягти, ви можете повернути middleware WithoutOverlapping із методу middleware вашого завдання:

use Illuminate\Queue\Middleware\WithoutOverlapping;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new WithoutOverlapping($this->user->id)];
}

Повернення завдання, що накладається, назад у чергу все одно збільшуватиме загальну кількість спроб завдання. Ви можете відповідно налаштувати атрибути Tries і MaxExceptions у вашому класі завдання. Наприклад, залишення Tries рівним 1, як за замовчуванням, завадить будь-якому завданню, що накладається, бути повтореним пізніше.

Будь-які завдання того самого типу, що накладаються, будуть повернені назад у чергу. Ви також можете вказати кількість секунд, які мають минути, перш ніж повернене завдання буде спробовано знову:

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->releaseAfter(60)];
}

Якщо ви хочете негайно видаляти будь-які завдання, що накладаються, щоб вони не повторювалися, ви можете скористатися методом dontRelease:

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->dontRelease()];
}

Middleware WithoutOverlapping працює на основі можливості атомарних блокувань Laravel. Іноді ваше завдання може несподівано впасти або завершитися за таймаутом так, що блокування не буде звільнено. Тому ви можете явно визначити час закінчення терміну дії блокування за допомогою методу expireAfter. Наприклад, наведений нижче приклад вкаже Laravel звільнити блокування WithoutOverlapping через три хвилини після початку обробки завдання:

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->expireAfter(180)];
}

Попередження. Middleware WithoutOverlapping потребує драйвера кешу, який підтримує блокування. Наразі атомарні блокування підтримують драйвери кешу memcached, redis, dynamodb, database, file і array.

Спільні ключі блокування між класами завдань

За замовчуванням middleware WithoutOverlapping запобігатиме накладанню лише завдань того самого класу. Тож, хоча два різні класи завдань можуть використовувати той самий ключ блокування, їм не буде заборонено накладатися. Однак ви можете вказати Laravel застосовувати ключ між класами завдань за допомогою методу shared:

use Illuminate\Queue\Middleware\WithoutOverlapping;

class ProviderIsDown
{
    // ...

    public function middleware(): array
    {
        return [
            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
        ];
    }
}

class ProviderIsUp
{
    // ...

    public function middleware(): array
    {
        return [
            (new WithoutOverlapping("status:{$this->provider}"))->shared(),
        ];
    }
}

Тротлінг винятків

Laravel містить middleware Illuminate\Queue\Middleware\ThrottlesExceptions, який дозволяє застосовувати тротлінг до винятків. Щойно завдання викине задану кількість винятків, усі подальші спроби виконати завдання відкладаються, доки не мине вказаний часовий інтервал. Цей middleware особливо корисний для завдань, які взаємодіють зі сторонніми сервісами, що є нестабільними.

Наприклад, уявімо завдання в черзі, яке взаємодіє зі стороннім API, що починає викидати винятки. Щоб застосувати тротлінг до винятків, ви можете повернути middleware ThrottlesExceptions із методу middleware вашого завдання. Зазвичай цей middleware слід поєднувати із завданням, що реалізує спроби на основі часу:

use DateTime;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new ThrottlesExceptions(10, 5 * 60)];
}

/**
 * Determine the time at which the job should timeout.
 */
public function retryUntil(): DateTime
{
    return now()->plus(minutes: 30);
}

Перший аргумент конструктора, який приймає middleware, — це кількість винятків, які завдання може викинути, перш ніж до нього буде застосовано тротлінг, а другий аргумент конструктора — кількість секунд, які мають минути, перш ніж завдання буде спробовано знову після застосування тротлінгу. У наведеному вище прикладі коду, якщо завдання викине 10 послідовних винятків, ми зачекаємо 5 хвилин, перш ніж спробувати виконати завдання знову, з обмеженням у 30-хвилинний ліміт часу.

Коли завдання викидає виняток, але порогу винятків ще не досягнуто, завдання зазвичай буде повторено негайно. Однак ви можете вказати кількість хвилин, на які таке завдання має бути відкладене, викликавши метод backoff під час прикріплення middleware до завдання:

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 5 * 60))->backoff(5)];
}

Метод backoff також приймає замикання, яке отримує викинутий виняток, дозволяючи визначати затримку динамічно:

use App\Exceptions\RateLimitedException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;
use Throwable;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 5 * 60))->backoff(
        fn (Throwable $throwable) => $throwable instanceof RateLimitedException
            ? $throwable->retryAfterMinutes()
            : 5
    )];
}

Усередині цей middleware використовує систему кешу Laravel для реалізації обмеження частоти, а імʼя класу завдання використовується як «ключ» кешу. Ви можете перевизначити цей ключ, викликавши метод by під час прикріплення middleware до вашого завдання. Це може бути корисним, якщо у вас є кілька завдань, які взаємодіють із тим самим стороннім сервісом, і ви хочете, щоб вони спільно використовували загальний «кошик» тротлінгу, гарантуючи дотримання єдиного спільного ліміту:

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->by('key')];
}

За замовчуванням цей middleware застосовуватиме тротлінг до кожного винятку. Ви можете змінити цю поведінку, викликавши метод when під час прикріплення middleware до вашого завдання. Тоді тротлінг до винятку буде застосовано лише в разі, якщо замикання, передане в метод when, повертає true:

use Illuminate\Http\Client\HttpClientException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->when(
        fn (Throwable $throwable) => $throwable instanceof HttpClientException
    )];
}

На відміну від методу when, який повертає завдання назад у чергу або викидає виняток, метод deleteWhen дозволяє повністю видалити завдання, коли виникає певний виняток:

use App\Exceptions\CustomerDeletedException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(2, 10 * 60))->deleteWhen(CustomerDeletedException::class)];
}

Якщо ви хочете, щоб винятки, до яких застосовано тротлінг, повідомлялися обробнику винятків вашого застосунку, ви можете зробити це, викликавши метод report під час прикріплення middleware до вашого завдання. За бажанням ви можете передати замикання в метод report, і виняток буде повідомлено лише в разі, якщо задане замикання повертає true:

use Illuminate\Http\Client\HttpClientException;
use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Get the middleware the job should pass through.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10 * 60))->report(
        fn (Throwable $throwable) => $throwable instanceof HttpClientException
    )];
}

Тротлінг винятків із Redis

Якщо ви використовуєте Redis, ви можете скористатися middleware Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis, який точно налаштований під Redis і ефективніший за базовий middleware тротлінгу винятків:

use Illuminate\Queue\Middleware\ThrottlesExceptionsWithRedis;

public function middleware(): array
{
    return [new ThrottlesExceptionsWithRedis(10, 10 * 60)];
}

Метод connection можна використовувати, щоб указати, яке зʼєднання Redis має використовувати middleware:

return [(new ThrottlesExceptionsWithRedis(10, 10 * 60))->connection('limiter')];

Повернення завдань у чергу

Middleware Release дозволяє повернути завдання назад у чергу без його виконання. Метод Release::when поверне завдання в чергу, якщо задана умова обчислюється як true, а метод Release::unless поверне завдання в чергу, якщо умова обчислюється як false:

use Illuminate\Queue\Middleware\Release;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Release::when($condition, releaseAfter: 60),
    ];
}

Повернення завдання назад у чергу все одно збільшуватиме загальну кількість спроб завдання. Ви можете відповідно налаштувати атрибути Tries і MaxExceptions у вашому класі завдання.

Ви також можете передати Closure у методи when і unless для складнішого обчислення умов:

use Illuminate\Queue\Middleware\Release;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Release::when(function (): bool {
            return ! $this->order->isPaid();
        }, releaseAfter: 60),
    ];
}

Пропуск завдань

Middleware Skip дозволяє вказати, що завдання слід пропустити / видалити без потреби змінювати логіку самого завдання. Метод Skip::when видалить завдання, якщо задана умова обчислюється як true, а метод Skip::unless видалить завдання, якщо умова обчислюється як false:

use Illuminate\Queue\Middleware\Skip;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Skip::when($condition),
    ];
}

Ви також можете передати Closure у методи when і unless для складнішого обчислення умов:

use Illuminate\Queue\Middleware\Skip;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [
        Skip::when(function (): bool {
            return $this->shouldSkip();
        }),
    ];
}

Диспетчеризація завдань

Щойно ви написали свій клас завдання, ви можете диспетчеризувати його за допомогою методу dispatch на самому завданні. Аргументи, передані в метод dispatch, буде передано в конструктор завдання:

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // ...

        ProcessPodcast::dispatch($podcast);

        return redirect('/podcasts');
    }
}

Якщо ви хочете диспетчеризувати завдання за умовою, ви можете скористатися методами dispatchIf і dispatchUnless:

ProcessPodcast::dispatchIf($accountActive, $podcast);

ProcessPodcast::dispatchUnless($accountSuspended, $podcast);

У нових застосунках Laravel зʼєднання database визначено як чергу за замовчуванням. Ви можете вказати інше зʼєднання черги за замовчуванням, змінивши змінну середовища QUEUE_CONNECTION у файлі .env вашого застосунку.

Відкладена диспетчеризація

Якщо ви хочете вказати, що завдання не має бути одразу доступним для обробки обробником черги, ви можете скористатися методом delay під час диспетчеризації завдання. Наприклад, укажімо, що завдання не має бути доступним для обробки протягом 10 хвилин після його диспетчеризації:

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // ...

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

        return redirect('/podcasts');
    }
}

У деяких випадках завдання можуть мати налаштовану затримку за замовчуванням. Якщо вам потрібно обійти цю затримку й диспетчеризувати завдання для негайної обробки, ви можете скористатися методом withoutDelay:

ProcessPodcast::dispatch($podcast)->withoutDelay();

Попередження. Сервіс черг Amazon SQS має максимальний час затримки 15 хвилин.

Синхронна диспетчеризація

Якщо потрібно виконати завдання негайно (синхронно), скористайтеся методом dispatchSync. Тоді завдання не потрапляє в чергу і виконується одразу в поточному процесі:

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Створюємо подкаст...

        ProcessPodcast::dispatchSync($podcast);

        return redirect('/podcasts');
    }
}

Відкладена диспетчеризація

Відкладена синхронна диспетчеризація дозволяє поставити завдання на обробку в поточному процесі, але вже після того, як HTTP-відповідь надіслано користувачу. Так «чергові» завдання виконуються синхронно й не сповільнюють роботу користувача з застосунком. Щоб відкласти виконання синхронного завдання, надішліть його на зʼєднання deferred:

RecordDelivery::dispatch($order)->onConnection('deferred');

Зʼєднання deferred також слугує стандартною failover-чергою.

Схожим чином зʼєднання background обробляє завдання після надсилання HTTP-відповіді користувачу, але завдання виконується в окремо породженому PHP-процесі, тож PHP-FPM / воркер застосунку лишається вільним для обробки наступного вхідного HTTP-запиту:

RecordDelivery::dispatch($order)->onConnection('background');

Масова диспетчеризація

Коли треба відправити багато незалежних завдань одразу, а відстеження пакетів чи колбеки не потрібні, скористайтеся методом bulk фасаду Bus. Laravel згрупує завдання за налаштованим зʼєднанням черги та назвою черги і запушить кожну групу у відповідну чергу однією операцією:

use App\Jobs\ProcessUser;
use Illuminate\Support\Facades\Bus;

Bus::bulk(
    $users->map(fn ($user) => new ProcessUser($user))
);

Підготовка завдань перед диспетчеризацією

Якщо завдання має підготувати або перевірити свій стан перед тим, як потрапити в чергу, воно може реалізувати інтерфейс Illuminate\Contracts\Queue\PreparesForDispatch. Laravel викличе метод prepareForDispatch перед диспетчеризацією завдання. Якщо цей метод поверне false, завдання не буде відправлено:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\PreparesForDispatch;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Support\Facades\Cache;

class SyncPodcasts implements PreparesForDispatch, ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public array $podcastIds,
    ) {}

    /**
     * Prepare the job before dispatching.
     */
    public function prepareForDispatch(): bool
    {
        return collect($this->podcastIds)
            ->reject(fn (int $id) => Cache::has("podcast-syncing:{$id}"))
            ->isNotEmpty();
    }
}

Завдання і транзакції бази даних

Відправляти завдання всередині транзакцій бази даних цілком припустимо, але тут треба подбати про те, щоб завдання справді змогло виконатися успішно. Коли завдання відправляється в межах транзакції, обробник черги може взяти його в роботу ще до того, як батьківська транзакція закомітилась. У такому разі всі зміни моделей чи записів бази, зроблені під час транзакції, можуть ще не бути відображені в базі. Так само моделі й записи, створені в транзакції, можуть у базі ще не існувати.

На щастя, Laravel пропонує кілька способів обійти цю проблему. По-перше, у масиві конфігурації зʼєднання черги можна виставити опцію after_commit:

'redis' => [
    'driver' => 'redis',
    // ...
    'after_commit' => true,
],

Коли опція after_commit має значення true, ви можете відправляти завдання всередині транзакцій бази даних; Laravel зачекає, поки відкриті батьківські транзакції закомітяться, і лише тоді реально відправить завдання. Ясна річ, якщо жодної відкритої транзакції немає, завдання відправиться одразу.

Якщо транзакцію відкотили через виняток, що стався під час неї, завдання, відправлені в межах цієї транзакції, будуть відкинуті.

[!NOTE] Значення true для опції конфігурації after_commit також призведе до того, що всі слухачі подій у черзі, mailable-класи, сповіщення та broadcast-події будуть відправлені після коміту всіх відкритих транзакцій бази даних.

Задання поведінки диспетчеризації після коміту на місці

Якщо ви не виставляєте опцію конфігурації зʼєднання черги after_commit у true, ви все одно можете вказати, що конкретне завдання має бути відправлене після коміту всіх відкритих транзакцій бази даних. Для цього додайте до операції диспетчеризації метод afterCommit:

use App\Jobs\ProcessPodcast;

ProcessPodcast::dispatch($podcast)->afterCommit();

І навпаки, якщо опція конфігурації after_commit виставлена в true, ви можете вказати, що конкретне завдання треба відправити негайно, не чекаючи коміту відкритих транзакцій бази даних:

ProcessPodcast::dispatch($podcast)->beforeCommit();

Ланцюжки завдань

Ланцюжок завдань дозволяє задати список завдань у черзі, які мають виконатися послідовно після успішного виконання головного завдання. Якщо одне завдання в послідовності зазнає невдачі, решта не запуститься. Щоб виконати ланцюжок завдань у черзі, скористайтеся методом chain фасаду Bus. Командна шина Laravel - це компонент нижчого рівня, на якому побудована диспетчеризація завдань у чергу:

use App\Jobs\OptimizePodcast;
use App\Jobs\ProcessPodcast;
use App\Jobs\ReleasePodcast;
use Illuminate\Support\Facades\Bus;

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->dispatch();

Окрім екземплярів класів завдань, у ланцюжок можна ставити й замикання:

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    function () {
        Podcast::update(/* ... */);
    },
])->dispatch();

[!WARNING] Видалення завдання методом $this->delete() всередині самого завдання не завадить обробці завдань із ланцюжка. Ланцюжок припиняє виконання, лише якщо завдання в ньому зазнало невдачі.

Зʼєднання та черга ланцюжка

Щоб задати зʼєднання й чергу для завдань у ланцюжку, скористайтеся методами onConnection та onQueue. Вони визначають зʼєднання черги й назву черги, які використовуватимуться, якщо завдання в черзі явно не призначене іншому зʼєднанню / черзі:

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->onConnection('redis')->onQueue('podcasts')->dispatch();

Додавання завдань до ланцюжка

Інколи потрібно додати завдання на початок або в кінець наявного ланцюжка з іншого завдання цього ж ланцюжка. Для цього є методи prependToChain та appendToChain:

/**
 * Execute the job.
 */
public function handle(): void
{
    // ...

    // Додаємо на початок поточного ланцюжка, завдання виконається одразу після поточного...
    $this->prependToChain(new TranscribePodcast);

    // Додаємо в кінець поточного ланцюжка, завдання виконається наприкінці...
    $this->appendToChain(new TranscribePodcast);
}

Невдачі в ланцюжку

Будуючи ланцюжок завдань, ви можете скористатися методом catch, щоб задати замикання, яке викличеться, якщо завдання в ланцюжку зазнає невдачі. Переданий колбек отримає екземпляр Throwable, що спричинив невдачу:

use Illuminate\Support\Facades\Bus;
use Throwable;

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->catch(function (Throwable $e) {
    // Завдання в ланцюжку зазнало невдачі...
})->dispatch();

[!WARNING] Оскільки колбеки ланцюжка серіалізуються й виконуються пізніше чергою Laravel, у них не можна використовувати змінну $this.

Налаштування черги та зʼєднання

Диспетчеризація в конкретну чергу

Розкладаючи завдання по різних чергах, ви можете «категоризувати» їх і навіть керувати пріоритетами, призначаючи різну кількість обробників на різні черги. Зверніть увагу: завдання при цьому не потрапляють у різні «зʼєднання» черги, визначені у файлі конфігурації, а лише в конкретні черги в межах одного зʼєднання. Щоб указати чергу, використайте метод onQueue під час диспетчеризації завдання:

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Створюємо подкаст...

        ProcessPodcast::dispatch($podcast)->onQueue('processing');

        return redirect('/podcasts');
    }
}

Як варіант, чергу завдання можна задати викликом методу onQueue у конструкторі завдання:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct()
    {
        $this->onQueue('processing');
    }
}

Диспетчеризація в конкретне зʼєднання

Якщо ваш застосунок працює з кількома зʼєднаннями черг, ви можете вказати, у яке зʼєднання запушити завдання, методом onConnection:

<?php

namespace App\Http\Controllers;

use App\Jobs\ProcessPodcast;
use App\Models\Podcast;
use Illuminate\Http\RedirectResponse;
use Illuminate\Http\Request;

class PodcastController extends Controller
{
    /**
     * Store a new podcast.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Створюємо подкаст...

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

        return redirect('/podcasts');
    }
}

Методи onConnection та onQueue можна ланцюжком поєднати, щоб задати і зʼєднання, і чергу для завдання:

ProcessPodcast::dispatch($podcast)
    ->onConnection('sqs')
    ->onQueue('processing');

Як варіант, зʼєднання завдання можна задати викликом методу onConnection у конструкторі завдання:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct()
    {
        $this->onConnection('sqs');
    }
}

Маршрутизація черг

Методом route фасаду Queue можна визначити стандартне зʼєднання й чергу для конкретних класів завдань. Це стає в пригоді, коли ви хочете гарантувати, що певні завдання завжди йдуть у визначені черги, і при цьому не вказувати зʼєднання чи чергу в самому завданні.

Крім конкретних класів завдань, у метод route можна передати інтерфейс, трейт або батьківський клас. Тоді будь-яке завдання, яке реалізує цей інтерфейс, використовує цей трейт або успадковується від цього батьківського класу, автоматично отримає налаштоване зʼєднання й чергу.

Зазвичай метод route викликають із методу boot сервіс-провайдера:

use App\Concerns\RequiresVideo;
use App\Jobs\ProcessPodcast;
use App\Jobs\ProcessVideo;
use Illuminate\Support\Facades\Queue;

/**
 * Bootstrap any application services.
 */
public function boot(): void
{
    Queue::route(ProcessPodcast::class, connection: 'redis', queue: 'podcasts');
    Queue::route(RequiresVideo::class, queue: 'video');
}

Якщо вказано зʼєднання без черги, завдання піде в чергу за замовчуванням:

Queue::route(ProcessPodcast::class, connection: 'redis');

Кілька класів завдань можна маршрутизувати за раз, передавши в метод route масив:

Queue::route([
    ProcessPodcast::class => ['redis', 'podcasts'], // Зʼєднання і черга
    ProcessVideo::class => 'videos', // Тільки черга (використовує стандартне зʼєднання)
]);

[!NOTE] Маршрутизацію черг усе одно можна перевизначити на рівні окремого завдання.

Метод forward перенаправляє завдання з однієї черги в іншу чергу та / або зʼєднання. Це корисно, коли треба змінити інфраструктуру черг, не чіпаючи окремі завдання чи місця їх диспетчеризації:

Queue::forward('reports', 'reports.fifo', 'sqs');
Queue::forward('payments', connection: 'sqs');
Queue::forward('updates', 'notifications');

Кілька черг можна перенаправити за раз, передавши масив:

Queue::forward([
    'reports' => 'reports.fifo',
    'emails' => 'emails.fifo',
], connection: 'sqs');

Явно налаштоване на завданні зʼєднання має пріоритет над перенаправленим зʼєднанням.

Задання максимальної кількості спроб / значень таймауту

Максимальна кількість спроб

Спроби виконання завдання - базове поняття системи черг Laravel, на якому тримається чимало просунутих можливостей. Спершу вони можуть здаватися заплутаними, тож перед зміною стандартної конфігурації розберіться, як вони працюють.

Коли завдання відправляють, воно потрапляє в чергу. Далі обробник бере його й намагається виконати. Це і є спроба виконання завдання.

Проте спроба не обовʼязково означає, що виконався метод handle завдання. Спробу можна «витратити» кількома способами:

• Під час виконання завдання стався необроблений виняток.
• Завдання вручну повернули в чергу через $this->release().
• Middleware на кшталт WithoutOverlapping чи RateLimited не змогло отримати блокування й повернуло завдання в чергу.
• Завдання перевищило таймаут.
• Метод handle завдання відпрацював і завершився без винятків.

Навряд чи ви хочете, щоб завдання намагалося виконатися нескінченно. Тому Laravel дає різні способи вказати, скільки разів або як довго завдання можна намагатися виконати.

[!NOTE] За замовчуванням Laravel робить лише одну спробу виконати завдання. Якщо ваше завдання використовує middleware на зразок WithoutOverlapping чи RateLimited або ви вручну повертаєте завдання в чергу, вам, найімовірніше, доведеться збільшити кількість дозволених спроб через опцію tries.

Один зі способів задати максимальну кількість спроб виконання завдання - ключ --tries у командному рядку Artisan. Він застосується до всіх завдань, які обробляє цей обробник, якщо саме завдання не вказує власну кількість спроб:

php artisan queue:work --tries=3

Якщо завдання вичерпало максимальну кількість спроб, воно вважається «невдалим» (failed). Докладніше про роботу з невдалими завданнями - у документації про невдалі завдання. Якщо команді queue:work передати --tries=0, завдання повторюватиметься нескінченно.

Можна діяти точніше й визначити максимальну кількість спроб на самому класі завдання за допомогою атрибута Tries. Кількість спроб, задана на завданні, має пріоритет над значенням --tries із командного рядка:

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Tries;

#[Tries(5)]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

Якщо для конкретного завдання потрібне динамічне керування максимальною кількістю спроб, визначте на завданні метод tries:

/**
 * Determine number of times the job may be attempted.
 */
public function tries(): int
{
    return 5;
}

Спроби на основі часу

Замість кількості спроб до провалу можна задати момент часу, після якого завдання більше не намагатимуться виконати. Так завдання можна повторювати будь-яку кількість разів у межах заданого проміжку. Щоб визначити цей момент, додайте до класу завдання метод retryUntil. Він має повертати екземпляр DateTime:

use DateTime;

/**
 * Determine the time at which the job should timeout.
 */
public function retryUntil(): DateTime
{
    return now()->plus(minutes: 10);
}

Якщо визначено і retryUntil, і tries, Laravel надає пріоритет методу retryUntil.

[!NOTE] Атрибут Tries або метод retryUntil можна також визначати на слухачах подій у черзі та сповіщеннях у черзі.

Максимальна кількість винятків

Іноді потрібно, щоб завдання можна було повторювати багато разів, але воно провалювалося після певної кількості необроблених винятків (на відміну від повернень у чергу через метод release). Для цього використайте на класі завдання атрибути Tries та MaxExceptions:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\MaxExceptions;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Support\Facades\Redis;

#[Tries(25)]
#[MaxExceptions(3)]
class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        Redis::throttle('key')->allow(10)->every(60)->then(function () {
            // Блокування отримано, обробляємо подкаст...
        }, function () {
            // Не вдалося отримати блокування...
            return $this->release(10);
        });
    }
}

У цьому прикладі завдання повертається в чергу на десять секунд, якщо застосунок не може отримати блокування Redis, і повторюватиметься до 25 разів. Проте завдання провалиться, якщо воно кине три необроблені винятки.

Припинення повторів за винятком

Іноді виняток означає, що завдання в черзі має провалитися негайно, а не повертатися в чергу для ще однієї спроби. Типи винятків, які припиняють повтори, налаштовуються методом винятків dontRetry у файлі bootstrap/app.php вашого застосунку:

use App\Exceptions\InvalidPodcastSourceException;
use Illuminate\Foundation\Configuration\Exceptions;

->withExceptions(function (Exceptions $exceptions): void {
    $exceptions->dontRetry([
        InvalidPodcastSourceException::class,
    ]);
})

Якщо потрібно точніше керувати тим, коли повтори мають припинятися, передайте замикання в метод dontRetryWhen. Коли замикання повертає true, завдання позначається як невдале й не повторюється:

use App\Exceptions\PodcastProcessingException;
use Illuminate\Foundation\Configuration\Exceptions;

->withExceptions(function (Exceptions $exceptions): void {
    $exceptions->dontRetryWhen(function (PodcastProcessingException $e) {
        return $e->reason() === 'Subscription expired';
    });
})

Таймаут

Зазвичай ви приблизно знаєте, скільки має тривати виконання ваших завдань у черзі. Тому Laravel дозволяє задати значення «таймауту». За замовчуванням воно становить 60 секунд. Якщо завдання обробляється довше за вказану кількість секунд, обробник, який його виконує, завершиться з помилкою. Як правило, обробник автоматично перезапустить менеджер процесів, налаштований на вашому сервері.

Максимальну кількість секунд виконання завдань можна задати ключем --timeout у командному рядку Artisan:

php artisan queue:work --timeout=30

Якщо завдання вичерпає максимальну кількість спроб через постійні таймаути, воно буде позначене як невдале.

Максимальну кількість секунд, яку дозволено виконуватися завданню, можна також визначити атрибутом Timeout на класі завдання. Таймаут, заданий на завданні, має пріоритет над будь-яким таймаутом із командного рядка:

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Timeout;

#[Timeout(120)]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

Іноді процеси, що блокують ввід-вивід, як-от сокети чи вихідні HTTP-зʼєднання, можуть не зважати на заданий вами таймаут. Тому, працюючи з ними, завжди намагайтеся задати таймаут і засобами їхніх API. Наприклад, використовуючи Guzzle, завжди вказуйте значення таймауту зʼєднання й запиту.

[!WARNING] Щоб задавати таймаути завдань, має бути встановлене PHP-розширення PCNTL. До того ж значення «timeout» завдання завжди має бути меншим за його значення "retry after". Інакше завдання можуть спробувати виконати повторно ще до того, як воно насправді завершилося або перевищило таймаут. Опція --timeout не діє, коли команду queue:work викликано з опцією --once.

Провал завдання за таймаутом

Якщо ви хочете, щоб завдання за таймаутом позначалося як невдале, скористайтеся атрибутом FailOnTimeout на класі завдання:

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\FailOnTimeout;

#[FailOnTimeout]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

[!NOTE] За замовчуванням, коли завдання перевищує таймаут, воно витрачає одну спробу й повертається в чергу (якщо повтори дозволені). Проте якщо ви налаштували завдання провалюватися за таймаутом, воно не повторюватиметься, незалежно від значення tries.

SQS FIFO та справедливі черги

Laravel підтримує черги Amazon SQS FIFO (First-In-First-Out) та справедливі (fair) черги. FIFO-черги дозволяють обробляти завдання точно в тому порядку, у якому їх надіслали, і водночас гарантують обробку рівно один раз завдяки дедуплікації повідомлень.

FIFO-черги потребують ідентифікатора групи повідомлень, щоб визначити, які завдання можна обробляти паралельно. Завдання з однаковим ідентифікатором групи обробляються послідовно, а повідомлення з різними ідентифікаторами груп - паралельно.

Laravel надає плавний метод onGroup для вказання ідентифікатора групи повідомлень під час диспетчеризації завдань:

ProcessOrder::dispatch($order)
    ->onGroup("customer-{$order->customer_id}");

Черги SQS FIFO підтримують дедуплікацію повідомлень, щоб гарантувати обробку рівно один раз. Реалізуйте в класі завдання метод deduplicationId, щоб задати власний ідентифікатор дедуплікації:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessSubscriptionRenewal implements ShouldQueue
{
    use Queueable;

    // ...

    /**
     * Get the job's deduplication ID.
     */
    public function deduplicationId(): string
    {
        return "renewal-{$this->subscription->id}";
    }
}

Справедливі черги

Якщо ви користуєтесь стандартною (standard) чергою SQS, задання групи повідомлень вмикає справедливу обробку черги. Інакше кажучи, щойно ви призначите групи, SQS використовуватиме їх для рівномірної доставки між тенантами / навантаженнями. Додаткових налаштувань у Laravel не потрібно.

Замість виклику onGroup під час диспетчеризації ви можете визначити метод messageGroup безпосередньо на завданні:

<?php

namespace App\Jobs;

use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ProcessOrder implements ShouldQueue
{
    use Queueable;

    // ...

    /**
     * Get the job's message group.
     */
    public function messageGroup(): string
    {
        return "customer-{$this->order->customer_id}";
    }
}

FIFO: слухачі, пошта та сповіщення

Використовуючи FIFO-черги, ви маєте визначити групи повідомлень також для слухачів, пошти та сповіщень. Як альтернатива, ставте ці обʼєкти в чергу, яка не є FIFO.

Щоб задати групу повідомлень для слухача подій у черзі, визначте на слухачі метод messageGroup. За бажанням можна також визначити метод deduplicationId:

<?php

namespace App\Listeners;

class SendShipmentNotification
{
    // ...

    /**
     * Get the job's message group.
     */
    public function messageGroup(): string
    {
        return 'shipments';
    }

    /**
     * Get the job's deduplication ID.
     */
    public function deduplicationId(): string
    {
        return "shipment-notification-{$this->shipment->id}";
    }
}

Надсилаючи поштове повідомлення, яке потрапить у FIFO-чергу, викличте метод onGroup і, за потреби, метод withDeduplicator:

use App\Mail\InvoicePaid;
use Illuminate\Support\Facades\Mail;

$invoicePaid = (new InvoicePaid($invoice))
    ->onGroup('invoices')
    ->withDeduplicator(fn () => 'invoices-'.$invoice->id);

Mail::to($request->user())->send($invoicePaid);

Надсилаючи сповіщення, яке потрапить у FIFO-чергу, викличте під час надсилання метод onGroup і, за потреби, метод withDeduplicator:

use App\Notifications\InvoicePaid;

$invoicePaid = (new InvoicePaid($invoice))
    ->onGroup('invoices')
    ->withDeduplicator(fn () => 'invoices-'.$invoice->id);

$user->notify($invoicePaid);

Failover черги

Драйвер черги failover забезпечує автоматичне перемикання при відправленні завдань у чергу. Якщо основне зʼєднання черги з конфігурації failover з будь-якої причини недоступне, Laravel автоматично спробує запушити завдання в наступне зʼєднання зі списку. Це особливо корисно для високої доступності в продакшн-середовищах, де надійність черг критична.

Щоб налаштувати failover-зʼєднання черги, вкажіть драйвер failover і масив назв зʼєднань, які треба пробувати по черзі. За замовчуванням Laravel уже містить приклад конфігурації failover у файлі config/queue.php вашого застосунку:

'failover' => [
    'driver' => 'failover',
    'connections' => [
        'redis',
        'database',
        'sync',
    ],
],

Налаштувавши зʼєднання з драйвером failover, зробіть його стандартним зʼєднанням черги у файлі .env вашого застосунку, щоб скористатися функціональністю failover:

QUEUE_CONNECTION=failover

Далі запустіть щонайменше один обробник для кожного зʼєднання зі списку failover:

php artisan queue:work redis
php artisan queue:work database

[!NOTE] Для зʼєднань із драйверами черги sync, background чи deferred запускати обробник не потрібно, бо ці драйвери обробляють завдання в поточному PHP-процесі.

Коли операція зі зʼєднанням черги провалюється й спрацьовує failover, Laravel відправляє подію Illuminate\Queue\Events\QueueFailedOver, тож ви можете повідомити про це чи записати в лог, що зʼєднання черги відмовило.

[!NOTE] Якщо ви користуєтесь Laravel Horizon, памʼятайте: Horizon керує лише чергами Redis. Якщо ваш список failover містить database, поряд із Horizon слід запускати звичайний процес php artisan queue:work database.

Обробка помилок

Якщо під час обробки завдання кинуто виняток, завдання автоматично повертається в чергу для наступної спроби. Воно повертатиметься доти, доки не вичерпає максимальну кількість спроб, дозволену вашим застосунком. Максимальна кількість спроб визначається ключем --tries команди Artisan queue:work. Як варіант, її можна визначити на самому класі завдання. Докладніше про запуск обробника черги нижче.

Ручне повернення завдання в чергу

Іноді потрібно вручну повернути завдання в чергу, щоб спробувати виконати його пізніше. Це робиться викликом методу release:

/**
 * Execute the job.
 */
public function handle(): void
{
    // ...

    $this->release();
}

За замовчуванням метод release повертає завдання в чергу для негайної обробки. Проте ви можете вказати черзі не робити завдання доступним для обробки, доки не мине задана кількість секунд, передавши в метод release ціле число або екземпляр дати:

$this->release(10);

$this->release(now()->plus(seconds: 10));

Ручне позначення завдання невдалим

Іноді потрібно вручну позначити завдання як «невдале». Для цього викличте метод fail:

/**
 * Execute the job.
 */
public function handle(): void
{
    // ...

    $this->fail();
}

Якщо ви хочете позначити завдання невдалим через виняток, який перехопили, передайте цей виняток у метод fail. Або, для зручності, передайте рядок із повідомленням про помилку, який буде перетворено на виняток за вас:

$this->fail($exception);

$this->fail('Something went wrong.');

[!NOTE] Докладніше про невдалі завдання - у документації про обробку невдач завдань.

Провал завдань на конкретних винятках

Job middleware FailOnException дозволяє припинити повтори, коли кинуто певні винятки. Так можна повторювати завдання на тимчасових винятках, як-от помилки зовнішнього API, але остаточно провалювати його на стійких винятках, наприклад коли в користувача відкликали права:

<?php

namespace App\Jobs;

use App\Models\User;
use Illuminate\Auth\Access\AuthorizationException;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Attributes\Tries;
use Illuminate\Queue\Middleware\FailOnException;
use Illuminate\Support\Facades\Http;

#[Tries(3)]
class SyncChatHistory implements ShouldQueue
{
    use Queueable;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public User $user,
    ) {}

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        $this->user->authorize('sync-chat-history');

        $response = Http::throw()->get(
            "https://chat.laravel.test/?user={$this->user->uuid}"
        );

        // ...
    }

    /**
     * Get the middleware the job should pass through.
     */
    public function middleware(): array
    {
        return [
            new FailOnException([AuthorizationException::class])
        ];
    }
}

Пакетна обробка завдань

Пакетна обробка завдань у Laravel дозволяє легко виконати групу завдань паралельно, а після завершення всього пакета виконати якусь дію.

Спершу створіть міграцію бази даних для таблиці, у якій зберігатиметься метаінформація про пакети завдань, зокрема відсоток їх виконання. Цю міграцію можна згенерувати командою Artisan make:queue-batches-table:

php artisan make:queue-batches-table

php artisan migrate

Визначення завдань для пакетів

Щоб зробити завдання пакетним, створіть завдання для черги як зазвичай, але додайте до класу трейт Illuminate\Bus\Batchable. Він дає доступ до методу batch, яким можна отримати поточний пакет, у межах якого виконується завдання:

<?php

namespace App\Jobs;

use Illuminate\Bus\Batchable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ImportCsv implements ShouldQueue
{
    use Batchable, Queueable;

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        if ($this->batch()->cancelled()) {
            // Визначаємо, чи пакет було скасовано...

            return;
        }

        // Імпортуємо частину CSV-файлу...
    }
}

Диспетчеризація пакетів

Щоб відправити пакет завдань, скористайтеся методом batch фасаду Bus. Пакети найкорисніші в поєднанні з колбеками завершення, тож для них є методи then, catch та finally. Кожен із цих колбеків отримує екземпляр Illuminate\Bus\Batch.

Коли працює кілька обробників черги, завдання пакета оброблятимуться паралельно. Тому порядок їх завершення може не збігатися з порядком, у якому їх додали до пакета. Про те, як виконати серію завдань послідовно, читайте в документації про ланцюжки й пакети завдань.

У цьому прикладі уявімо, що ставимо в чергу пакет завдань, кожне з яких обробляє певну кількість рядків CSV-файлу:

use App\Jobs\ImportCsv;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;
use Throwable;

$batch = Bus::batch([
    new ImportCsv(1, 100),
    new ImportCsv(101, 200),
    new ImportCsv(201, 300),
    new ImportCsv(301, 400),
    new ImportCsv(401, 500),
])->before(function (Batch $batch) {
    // Пакет створено, але жодного завдання ще не додано...
})->progress(function (Batch $batch) {
    // Одне завдання успішно завершилося...
})->then(function (Batch $batch) {
    // Усі завдання успішно завершилися...
})->catch(function (Batch $batch, Throwable $e) {
    // Виявлено невдале завдання в пакеті...
})->finally(function (Batch $batch) {
    // Пакет завершив виконання...
})->dispatch();

return $batch->id;

Ідентифікатор пакета, доступний через властивість $batch->id, можна використати, щоб запитати командну шину Laravel про інформацію щодо пакета після його відправлення.

[!WARNING] Оскільки колбеки пакета серіалізуються й виконуються пізніше чергою Laravel, у них не можна використовувати змінну $this. До того ж, оскільки завдання пакета обгортаються в транзакції бази даних, усередині них не варто виконувати SQL-інструкції, що спричиняють неявний коміт.

Назви пакетів

Деякі інструменти, як-от Laravel Horizon та Laravel Telescope, показують зручнішу для читання зневаджувальну інформацію, якщо пакети мають назви. Щоб призначити пакету довільну назву, викличте метод name під час його визначення:

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // Усі завдання успішно завершилися...
})->name('Import CSV')->dispatch();

Зʼєднання та черга пакета

Щоб задати зʼєднання й чергу для завдань пакета, скористайтеся методами onConnection та onQueue. Усі завдання пакета мають виконуватися в межах того самого зʼєднання й черги:

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // Усі завдання успішно завершилися...
})->onConnection('redis')->onQueue('imports')->dispatch();

Ланцюжки та пакети

Усередині пакета можна визначити набір завдань у ланцюжку, помістивши їх у масив. Наприклад, так можна виконати два ланцюжки паралельно й запустити колбек, коли обидва завершать обробку:

use App\Jobs\ReleasePodcast;
use App\Jobs\SendPodcastReleaseNotification;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;

Bus::batch([
    [
        new ReleasePodcast(1),
        new SendPodcastReleaseNotification(1),
    ],
    [
        new ReleasePodcast(2),
        new SendPodcastReleaseNotification(2),
    ],
])->then(function (Batch $batch) {
    // Усі завдання успішно завершилися...
})->dispatch();

І навпаки, пакети завдань можна виконувати всередині ланцюжка, визначивши їх у ньому. Наприклад, спершу можна запустити пакет завдань, який публікує кілька подкастів, а потім пакет завдань, який розсилає сповіщення про публікацію:

use App\Jobs\FlushPodcastCache;
use App\Jobs\ReleasePodcast;
use App\Jobs\SendPodcastReleaseNotification;
use Illuminate\Support\Facades\Bus;

Bus::chain([
    new FlushPodcastCache,
    Bus::batch([
        new ReleasePodcast(1),
        new ReleasePodcast(2),
    ]),
    Bus::batch([
        new SendPodcastReleaseNotification(1),
        new SendPodcastReleaseNotification(2),
    ]),
])->dispatch();

Додавання завдань до пакетів

Іноді корисно додати додаткові завдання до пакета зсередини завдання цього ж пакета. Такий підхід зручний, коли потрібно зібрати пакет із тисяч завдань, диспетчеризація яких зайняла б забагато часу під час веб-запиту. Замість цього можна відправити початковий пакет «завантажувальних» завдань, які наповнять пакет рештою:

$batch = Bus::batch([
    new LoadImportBatch,
    new LoadImportBatch,
    new LoadImportBatch,
])->then(function (Batch $batch) {
    // Усі завдання успішно завершилися...
})->name('Import Contacts')->dispatch();

У цьому прикладі завдання LoadImportBatch наповнює пакет додатковими завданнями. Для цього використовується метод add на екземплярі пакета, доступному через метод batch завдання:

use App\Jobs\ImportContacts;
use Illuminate\Support\Collection;

/**
 * Execute the job.
 */
public function handle(): void
{
    if ($this->batch()->cancelled()) {
        return;
    }

    $this->batch()->add(Collection::times(1000, function () {
        return new ImportContacts;
    }));
}

[!WARNING] Додавати завдання до пакета можна лише зсередини завдання, що належить до цього ж пакета.

Огляд пакетів

Екземпляр Illuminate\Bus\Batch, який передається в колбеки завершення пакета, має чимало властивостей і методів для взаємодії з конкретним пакетом завдань та його огляду:

// UUID пакета...
$batch->id;

// Назва пакета (якщо є)...
$batch->name;

// Кількість завдань, призначених пакету...
$batch->totalJobs;

// Кількість завдань, які черга ще не обробила...
$batch->pendingJobs;

// Кількість невдалих завдань...
$batch->failedJobs;

// Кількість оброблених на цей момент завдань...
$batch->processedJobs();

// Відсоток виконання пакета (0-100)...
$batch->progress();

// Чи завершив пакет виконання...
$batch->finished();

// Скасувати виконання пакета...
$batch->cancel();

// Чи було пакет скасовано...
$batch->cancelled();

Повернення пакетів із маршрутів

Усі екземпляри Illuminate\Bus\Batch серіалізуються в JSON, тож їх можна повертати прямо з маршрутів застосунку й отримувати JSON-навантаження з інформацією про пакет, зокрема про прогрес виконання. Це зручно, коли треба показати прогрес пакета в інтерфейсі застосунку.

Щоб отримати пакет за його ідентифікатором, скористайтеся методом findBatch фасаду Bus:

use Illuminate\Support\Facades\Bus;
use Illuminate\Support\Facades\Route;

Route::get('/batch/{batchId}', function (string $batchId) {
    return Bus::findBatch($batchId);
});

Скасування пакетів

Іноді потрібно скасувати виконання певного пакета. Це робиться викликом методу cancel на екземплярі Illuminate\Bus\Batch:

/**
 * Execute the job.
 */
public function handle(): void
{
    if ($this->user->exceedsImportLimit()) {
        $this->batch()->cancel();

        return;
    }

    if ($this->batch()->cancelled()) {
        return;
    }
}

Як ви могли помітити в попередніх прикладах, завдання пакета зазвичай мають перевіряти, чи не скасовано відповідний пакет, перш ніж продовжувати виконання. Проте для зручності завданню можна призначити middleware SkipIfBatchCancelled. Як видно з назви, це middleware вкаже Laravel не обробляти завдання, якщо його пакет скасовано:

use Illuminate\Queue\Middleware\SkipIfBatchCancelled;

/**
 * Get the middleware the job should pass through.
 */
public function middleware(): array
{
    return [new SkipIfBatchCancelled];
}

Невдачі в пакетах

Коли завдання пакета зазнає невдачі, викликається колбек catch (якщо його призначено). Цей колбек викликається лише для першого завдання пакета, яке провалилося.

Дозвіл на невдачі

Коли завдання в пакеті провалюється, Laravel автоматично позначає пакет як «скасований». За бажанням цю поведінку можна вимкнути, щоб невдача завдання не скасовувала пакет автоматично. Для цього під час диспетчеризації пакета викличте метод allowFailures:

$batch = Bus::batch([
    // ...
])->then(function (Batch $batch) {
    // Усі завдання успішно завершилися...
})->allowFailures()->dispatch();

У метод allowFailures можна передати замикання, яке виконуватиметься на кожній невдачі завдання:

$batch = Bus::batch([
    // ...
])->allowFailures(function (Batch $batch, $exception) {
    // Обробляємо невдачі окремих завдань...
})->dispatch();

Повторний запуск невдалих завдань пакета

Для зручності Laravel має команду Artisan queue:retry-batch, яка дозволяє легко повторити всі невдалі завдання конкретного пакета. Команда приймає UUID пакета, чиї невдалі завдання треба повторити:

php artisan queue:retry-batch 32dbc76c-4f82-4749-b610-a639fe0099b5

Очищення пакетів

Без очищення таблиця job_batches дуже швидко накопичує записи. Щоб цьому запобігти, заплануйте щоденний запуск команди Artisan queue:prune-batches:

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches')->daily();

За замовчуванням видаляються всі завершені пакети, старші за 24 години. Опція hours при виклику команди задає, як довго зберігати дані пакетів. Наприклад, така команда видалить усі пакети, що завершилися понад 48 годин тому:

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches --hours=48')->daily();

Іноді в таблиці job_batches накопичуються записи про пакети, які так і не завершилися успішно, наприклад ті, де завдання провалилося й не було успішно повторене. Такі записи незавершених пакетів команда queue:prune-batches може видаляти за опцією unfinished:

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches --hours=48 --unfinished=72')->daily();

Так само в таблиці job_batches можуть накопичуватися записи про скасовані пакети. Їх команда queue:prune-batches видаляє за опцією cancelled:

use Illuminate\Support\Facades\Schedule;

Schedule::command('queue:prune-batches --hours=48 --cancelled=72')->daily();

Зберігання пакетів у DynamoDB

Laravel також підтримує зберігання метаінформації про пакети (batch) у DynamoDB замість реляційної бази даних. Проте таблицю DynamoDB для зберігання всіх записів про пакети доведеться створити вручну.

Зазвичай ця таблиця має називатися job_batches, але назву слід узгодити зі значенням конфігурації queue.batching.table у файлі конфігурації queue вашого застосунку.

Конфігурація таблиці пакетів у DynamoDB

Таблиця job_batches повинна мати рядковий первинний ключ розділу (partition key) з назвою application і рядковий первинний ключ сортування (sort key) з назвою id. Частина ключа application міститиме назву вашого застосунку, визначену значенням конфігурації name у файлі конфігурації app. Оскільки назва застосунку є частиною ключа таблиці DynamoDB, ту саму таблицю можна використовувати для зберігання пакетів завдань кількох застосунків Laravel.

Крім того, ви можете визначити атрибут ttl для своєї таблиці, якщо хочете скористатися автоматичним очищенням пакетів.

Конфігурація DynamoDB

Далі встановіть AWS SDK, щоб ваш застосунок Laravel міг взаємодіяти з Amazon DynamoDB:

composer require aws/aws-sdk-php

Потім задайте опції конфігурації queue.batching.driver значення dynamodb. Крім того, у масиві конфігурації batching слід визначити опції key, secret і region. Вони використовуються для автентифікації в AWS. При використанні драйвера dynamodb опція конфігурації queue.batching.database не потрібна:

'batching' => [
    'driver' => env('QUEUE_BATCHING_DRIVER', 'dynamodb'),
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'table' => 'job_batches',
],

Очищення пакетів у DynamoDB

Коли інформація про пакети завдань зберігається в DynamoDB, звичні команди очищення, розраховані на реляційну базу даних, не працюють. Замість них можна скористатися власною функціональністю TTL у DynamoDB, щоб автоматично видаляти записи старих пакетів.

Якщо ви визначили в таблиці DynamoDB атрибут ttl, ви можете задати параметри конфігурації, які підкажуть Laravel, як очищати записи пакетів. Значення конфігурації queue.batching.ttl_attribute визначає назву атрибута, що зберігає TTL, а значення queue.batching.ttl - кількість секунд, після якої запис пакета можна видалити з таблиці DynamoDB, рахуючи від останнього оновлення запису:

'batching' => [
    'driver' => env('QUEUE_FAILED_DRIVER', 'dynamodb'),
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'table' => 'job_batches',
    'ttl_attribute' => 'ttl',
    'ttl' => 60 * 60 * 24 * 7, // 7 днів...
],

Черга із замикань

Замість того щоб відправляти в чергу клас завдання, ви можете відправити замикання. Це зручно для швидких простих задач, які треба виконати поза поточним циклом запиту. Під час відправлення замикань у чергу вміст коду замикання криптографічно підписується, тож змінити його дорогою неможливо:

use App\Models\Podcast;

$podcast = Podcast::find(1);

dispatch(function () use ($podcast) {
    $podcast->publish();
});

Щоб дати замиканню в черзі назву, яку зможуть використати дашборди звітності черг і яку показує команда queue:work, скористайтеся методом name:

dispatch(function () {
    // ...
})->name('Publish Podcast');

За допомогою методу catch ви можете передати замикання, яке виконається, якщо замикання в черзі так і не завершиться успішно після вичерпання всіх налаштованих спроб повтору:

use Throwable;

dispatch(function () use ($podcast) {
    $podcast->publish();
})->catch(function (Throwable $e) {
    // Це завдання зазнало невдачі...
});

[!WARNING] Оскільки колбеки catch серіалізуються й виконуються пізніше чергою Laravel, усередині них не слід використовувати змінну $this.

Запуск обробника черги

Команда queue:work

Laravel містить Artisan-команду, яка запускає обробник черги й обробляє нові завдання в міру їх надходження в чергу. Запустити обробник можна командою queue:work. Зверніть увагу: після запуску queue:work працюватиме доти, доки ви не зупините її вручну або не закриєте термінал:

php artisan queue:work

[!NOTE] Щоб процес queue:work постійно працював у фоні, використовуйте монітор процесів на кшталт Supervisor, який стежитиме, щоб обробник черги не зупинявся.

Якщо ви хочете, щоб у виводі команди були ID оброблених завдань, назви з'єднань і назви черг, додайте прапорець -v:

php artisan queue:work -v

Пам'ятайте, що обробники черги - це довгоживучі процеси, які тримають завантажений стан застосунку в пам'яті. Тому вони не помічають змін у вашому коді після запуску. Отже, під час деплою обов'язково перезапускайте обробники черги. Крім того, пам'ятайте, що будь-який статичний стан, створений або змінений вашим застосунком, не скидається автоматично між завданнями.

Як альтернативу можна запустити команду queue:listen. З нею не потрібно вручну перезапускати обробник, щоб підхопити оновлений код або скинути стан застосунку; але ця команда значно менш ефективна за queue:work:

php artisan queue:listen

Запуск кількох обробників черги

Щоб призначити на чергу кілька обробників і виконувати завдання паралельно, просто запустіть кілька процесів queue:work. Локально це можна зробити через кілька вкладок термінала, а в продакшені - через налаштування вашого менеджера процесів. При використанні Supervisor для цього є параметр numprocs.

Вказання з'єднання і черги

Ви також можете вказати, яке з'єднання черги має використовувати обробник. Назва з'єднання, передана команді work, повинна відповідати одному зі з'єднань, визначених у файлі конфігурації config/queue.php:

php artisan queue:work redis

За замовчуванням команда queue:work обробляє лише завдання з черги за замовчуванням для конкретного з'єднання. Проте обробник можна налаштувати ще точніше, обробляючи тільки певні черги конкретного з'єднання. Наприклад, якщо всі ваші листи обробляються в черзі emails на з'єднанні redis, запустіть обробник, який працюватиме лише з цією чергою:

php artisan queue:work redis --queue=emails

Обробка визначеної кількості завдань

Опція --once вказує обробнику взяти з черги лише одне завдання:

php artisan queue:work --once

Опція --max-jobs вказує обробнику обробити задану кількість завдань і завершитися. Це буває корисно в поєднанні з Supervisor: обробники автоматично перезапускаються після обробки певної кількості завдань, звільняючи накопичену пам'ять:

php artisan queue:work --max-jobs=1000

Обробка всіх завдань у черзі й вихід

Опція --stop-when-empty вказує обробнику обробити всі завдання й коректно завершити роботу. Це зручно, коли ви обробляєте черги Laravel усередині Docker-контейнера й хочете зупинити контейнер після спорожнення черги:

php artisan queue:work --stop-when-empty

Обробка завдань протягом заданої кількості секунд

Опція --max-time вказує обробнику обробляти завдання задану кількість секунд, а потім завершитися. Це може бути корисно в поєднанні з Supervisor: обробники автоматично перезапускаються після певного часу роботи, звільняючи накопичену пам'ять:

# Обробляти завдання одну годину, потім завершитися...
php artisan queue:work --max-time=3600

Тривалість сну обробника

Поки в черзі є завдання, обробник виконує їх без пауз між завданнями. Проте опція sleep визначає, скільки секунд обробник «спатиме», якщо доступних завдань немає. Зрозуміло, під час сну обробник не обробляє нових завдань:

php artisan queue:work --sleep=3

Режим обслуговування і черги

Поки застосунок перебуває в режимі обслуговування, завдання з черги не обробляються. Обробка відновиться у звичайному режимі, щойно застосунок вийде з режиму обслуговування.

Щоб змусити обробники черги виконувати завдання навіть при увімкненому режимі обслуговування, скористайтеся опцією --force:

php artisan queue:work --force

Зауваження щодо ресурсів

Демон-обробники черги не «перезавантажують» фреймворк перед обробкою кожного завдання. Тому вивільняйте всі важкі ресурси після завершення кожного завдання. Наприклад, якщо ви робите маніпуляції із зображеннями за допомогою бібліотеки GD, звільняйте пам'ять через imagedestroy, коли обробку зображення завершено.

Пріоритети черг

Іноді потрібно визначити пріоритет обробки черг. Наприклад, у файлі конфігурації config/queue.php ви можете задати чергу queue за замовчуванням для з'єднання redis як low. Проте час від часу вам може знадобитися відправити завдання в чергу високого пріоритету high:

dispatch((new Job)->onQueue('high'));

Щоб запустити обробник, який гарантує обробку всіх завдань черги high перед переходом до завдань черги low, передайте команді work розділений комами список назв черг:

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

Обробники черги і деплой

Оскільки обробники черги - довгоживучі процеси, вони не помітять змін у коді без перезапуску. Тому найпростіший спосіб задеплоїти застосунок з обробниками черги - перезапускати їх під час деплою. Коректно перезапустити всі обробники можна командою queue:restart:

php artisan queue:restart

Ця команда вказує всім обробникам черги коректно завершитися після того, як вони закінчать поточне завдання, тож жодне з наявних завдань не буде втрачено. Оскільки після виконання queue:restart обробники черги завершуються, вам слід використовувати менеджер процесів на кшталт Supervisor, який автоматично їх перезапустить.

[!NOTE] Черга використовує кеш для зберігання сигналів перезапуску, тож перед використанням цієї можливості переконайтеся, що драйвер кешу у вашому застосунку налаштований коректно.

Реакція на сигнали обробника

Коли обробник черги під час виконання завдання отримує сигнал завершення, наприклад SIGQUIT, SIGTERM або SIGINT, він доробляє поточне завдання й лише потім завершується. Однак вашому завданню може знадобитися відреагувати на сигнал до того, як процес зупинить сервер чи оркестратор контейнерів. Скажімо, тривалий імпорт може припинити витягувати нові записи й зберегти поточний прогрес.

Щоб реагувати на сигнали обробника зсередини завдання, реалізуйте інтерфейс Illuminate\Contracts\Queue\Interruptible і визначте в завданні метод interrupted. Номер сигналу, який отримав обробник, буде передано в метод interrupted:

<?php

namespace App\Jobs;

use App\Models\Import;
use Illuminate\Contracts\Queue\Interruptible;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;

class ImportProducts implements ShouldQueue, Interruptible
{
    use Queueable;

    protected bool $shouldStop = false;

    /**
     * Створити новий екземпляр завдання.
     */
    public function __construct(
        public Import $import,
    ) {}

    /**
     * Виконати завдання.
     */
    public function handle(): void
    {
        foreach ($this->import->pendingRows() as $row) {
            if ($this->shouldStop) {
                break;
            }

            // Імпортувати рядок продукту...
        }

        $this->import->saveProgress();
    }

    /**
     * Обробити сигнал, отриманий обробником черги.
     */
    public function interrupted(int $signal): void
    {
        $this->shouldStop = true;
    }
}

Метод interrupted викликається лише тоді, коли обробник отримує сигнал процесу під час виконання завдання. Він не замінює таймаути чи метод failed завдання.

Терміни придатності й таймаути завдань

Термін придатності завдання

У файлі конфігурації config/queue.php кожне з'єднання черги визначає опцію retry_after. Вона задає, скільки секунд з'єднання черги має чекати, перш ніж повторити завдання, яке зараз обробляється. Наприклад, якщо retry_after дорівнює 90, завдання повернеться в чергу, якщо воно обробляється 90 секунд і за цей час не було звільнене або видалене. Зазвичай значення retry_after варто ставити рівним максимальній кількості секунд, за яку ваші завдання розумно встигають виконатися.

[!WARNING] Єдине з'єднання черги, яке не містить значення retry_after, - Amazon SQS. SQS повторює завдання на основі Default Visibility Timeout, який керується в консолі AWS.

Таймаути обробника

Artisan-команда queue:work має опцію --timeout. За замовчуванням її значення - 60 секунд. Якщо завдання обробляється довше за вказану кількість секунд, обробник, який його виконує, завершиться з помилкою. Зазвичай обробник автоматично перезапускає менеджер процесів, налаштований на вашому сервері:

php artisan queue:work --timeout=60

Опція конфігурації retry_after і CLI-опція --timeout різні, але працюють разом, щоб завдання не губилися й оброблялися успішно лише один раз.

[!WARNING] Значення --timeout завжди має бути щонайменше на кілька секунд меншим за значення конфігурації retry_after. Це гарантує, що обробник, який виконує зависле завдання, буде завершений раніше, ніж завдання повторять. Якщо опція --timeout більша за значення retry_after, ваші завдання можуть виконатися двічі.

Призупинення й відновлення обробників черги

Іноді потрібно тимчасово заборонити обробнику черги брати нові завдання, не зупиняючи його повністю. Наприклад, ви можете захотіти призупинити обробку завдань на час системного обслуговування. Laravel надає Artisan-команди queue:pause і queue:continue для призупинення й відновлення обробників черги.

Щоб призупинити конкретну чергу, вкажіть назву з'єднання черги й назву черги:

php artisan queue:pause database:default

У цьому прикладі database - назва з'єднання черги, а default - назва черги. Після призупинення черги обробники, які виконують завдання з неї, доведуть до кінця поточне завдання, але не братимуть нових, доки чергу не відновлять.

Щоб призупинити обробку завдань для всіх черг на всіх з'єднаннях, скористайтеся опцією --all:

php artisan queue:pause --all

Щоб відновити обробку завдань у призупиненій черзі, використовуйте команду queue:continue:

php artisan queue:continue database:default

Щоб відновити обробку завдань для всіх черг на всіх з'єднаннях, використайте опцію --all з командою queue:resume:

php artisan queue:resume --all

Після відновлення черги обробники одразу почнуть виконувати нові завдання з неї. Відновлення всіх черг не відновлює черги, які були призупинені окремо. Зауважте, що призупинення черги не зупиняє сам процес обробника - воно лише не дає обробнику брати нові завдання із зазначеної черги.

Сигнали перезапуску й призупинення обробника

За замовчуванням обробники черги опитують драйвер кешу щодо сигналів перезапуску й призупинення на кожній ітерації завдання. Це опитування необхідне для реакції на команди queue:restart і queue:pause, але воно дає невеликі накладні витрати на продуктивність.

Якщо вам треба оптимізувати продуктивність і ці можливості переривання не потрібні, ви можете вимкнути опитування глобально, викликавши метод withoutInterruptionPolling на фасаді Queue. Зазвичай це роблять у методі boot вашого AppServiceProvider:

use Illuminate\Support\Facades\Queue;

/**
 * Завантажити будь-які сервіси застосунку.
 */
public function boot(): void
{
    Queue::withoutInterruptionPolling();
}

Також можна вимкнути опитування перезапуску або призупинення окремо, задавши статичні властивості $restartable чи $pausable класу Illuminate\Queue\Worker:

use Illuminate\Queue\Worker;

/**
 * Завантажити будь-які сервіси застосунку.
 */
public function boot(): void
{
    Worker::$restartable = false;
    Worker::$pausable = false;
}

[!WARNING] Коли опитування переривань вимкнене, обробники не реагуватимуть на команди queue:restart чи queue:pause (залежно від того, які саме можливості вимкнено).

Налаштування Supervisor

У продакшені потрібен спосіб тримати процеси queue:work живими. Процес queue:work може зупинитися з різних причин: перевищено таймаут обробника, виконано команду queue:restart тощо.

Через це вам потрібен монітор процесів, який виявить завершення процесів queue:work і автоматично перезапустить їх. Крім того, монітори процесів дозволяють вказати, скільки процесів queue:work ви хочете запускати одночасно. Supervisor - монітор процесів, поширений у Linux-середовищах, і далі в документації ми розберемо його налаштування.

Встановлення Supervisor

Supervisor - це монітор процесів для операційної системи Linux, який автоматично перезапустить ваші процеси queue:work, якщо вони впадуть. Щоб встановити Supervisor на Ubuntu, скористайтеся такою командою:

sudo apt-get install supervisor

[!NOTE] Якщо самостійне налаштування й адміністрування Supervisor здається надто складним, розгляньте Laravel Cloud, який надає повністю керовану платформу для запуску обробників черг Laravel.

Налаштування Supervisor

Файли конфігурації Supervisor зазвичай лежать у теці /etc/supervisor/conf.d. У ній ви можете створити будь-яку кількість конфігураційних файлів, які вказують supervisor, як стежити за вашими процесами. Наприклад, створімо файл laravel-worker.conf, який запускає процеси queue:work і моніторить їх:

[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /home/forge/app.com/artisan queue:work --sleep=3 --tries=3 --max-time=3600
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=forge
numprocs=8
redirect_stderr=true
stdout_logfile=/home/forge/app.com/worker.log
stopwaitsecs=3600

У цьому прикладі директива numprocs вказує Supervisor запустити вісім процесів queue:work і стежити за всіма, автоматично перезапускаючи ті, що впали. Директиву command у конфігурації змініть відповідно до потрібного з'єднання черги й опцій обробника.

[!WARNING] Переконайтеся, що значення stopwaitsecs більше за кількість секунд, яку споживає ваше найдовше завдання. Інакше Supervisor може вбити завдання до завершення обробки.

Запуск Supervisor

Після створення конфігураційного файла оновіть конфігурацію Supervisor і запустіть процеси такими командами:

sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start "laravel-worker:*"

Докладніше про Supervisor читайте в документації Supervisor.

Робота з невдалими завданнями

Іноді ваші завдання в черзі зазнають невдачі. Не хвилюйтеся, не все завжди йде за планом! Laravel має зручний спосіб вказати максимальну кількість спроб виконання завдання. Після того, як асинхронне завдання вичерпає цю кількість спроб, воно потрапить у таблицю failed_jobs. Синхронно відправлені завдання, які зазнали невдачі, у цю таблицю не записуються, а їхні винятки одразу обробляє застосунок.

Міграція для створення таблиці failed_jobs зазвичай уже присутня в нових застосунках Laravel. Проте якщо у вашому застосунку її немає, створіть міграцію командою make:queue-failed-table:

php artisan make:queue-failed-table

php artisan migrate

Запускаючи процес обробника черги, ви можете вказати максимальну кількість спроб виконання завдання через ключ --tries команди queue:work. Якщо значення для --tries не задане, завдання буде спробуване один раз або стільки разів, скільки визначено атрибутом Tries класу завдання:

php artisan queue:work redis --tries=3

За допомогою опції --backoff ви можете вказати, скільки секунд Laravel має чекати перед повторною спробою завдання, яке отримало виняток. За замовчуванням завдання одразу повертається в чергу, щоб спробувати ще раз:

php artisan queue:work redis --tries=3 --backoff=3

Якщо ви хочете налаштувати час очікування перед повтором для кожного завдання окремо, скористайтеся атрибутом Backoff у класі завдання:

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Backoff;

#[Backoff(3)]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

Якщо для визначення часу відкату потрібна складніша логіка, визначте в класі завдання метод backoff:

/**
 * Обчислити кількість секунд очікування перед повторною спробою завдання.
 */
public function backoff(): int
{
    return 3;
}

«Експоненційні» відкати легко налаштувати, задавши масив значень. У цьому прикладі затримка перед повтором становитиме 1 секунду для першої спроби, 5 секунд для другої, 10 секунд для третьої і 10 секунд для кожної наступної, якщо спроби ще лишилися:

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\Backoff;

#[Backoff([1, 5, 10])]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

Прибирання після невдалих завдань

Коли конкретне завдання зазнає невдачі, ви можете надіслати сповіщення користувачам або відкотити дії, які завдання виконало частково. Для цього визначте в класі завдання метод failed. У нього буде передано екземпляр Throwable, що спричинив невдачу:

<?php

namespace App\Jobs;

use App\Models\Podcast;
use App\Services\AudioProcessor;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Throwable;

class ProcessPodcast implements ShouldQueue
{
    use Queueable;

    /**
     * Створити новий екземпляр завдання.
     */
    public function __construct(
        public Podcast $podcast,
    ) {}

    /**
     * Виконати завдання.
     */
    public function handle(AudioProcessor $processor): void
    {
        // Обробити завантажений подкаст...
    }

    /**
     * Обробити невдачу завдання.
     */
    public function failed(?Throwable $exception): void
    {
        // Надіслати користувачу сповіщення про невдачу тощо...
    }
}

[!WARNING] Перед викликом методу failed створюється новий екземпляр завдання, тому будь-які зміни властивостей класу, що сталися в методі handle, будуть втрачені.

Невдалим вважається не тільки те завдання, у якому стався необроблений виняток. Завдання також може вважатися невдалим, коли вичерпано всі дозволені спроби. Спроби витрачаються кількома способами:

• Завдання перевищило таймаут.
• Під час виконання завдання стався необроблений виняток.
• Завдання повернули в чергу вручну або через middleware.

Якщо остання спроба зазнала невдачі через виняток, кинутий під час виконання завдання, цей виняток буде передано в метод failed. Проте якщо завдання зазнало невдачі через досягнення максимальної кількості дозволених спроб, $exception буде екземпляром Illuminate\Queue\MaxAttemptsExceededException. Так само, якщо завдання зазнало невдачі через перевищення налаштованого таймауту, $exception буде екземпляром Illuminate\Queue\TimeoutExceededException.

Повторення невдалих завдань

Щоб переглянути всі невдалі завдання, записані в таблицю failed_jobs, скористайтеся Artisan-командою queue:failed:

php artisan queue:failed

Команда queue:failed виведе ID завдання, з'єднання, чергу, час невдачі та іншу інформацію про завдання. ID завдання можна використати для його повторного запуску. Наприклад, щоб повторити невдале завдання з ID ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece, виконайте таку команду:

php artisan queue:retry ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece

За потреби команді можна передати кілька ID:

php artisan queue:retry ce7bb17c-cdd8-41f0-a8ec-7b4fef4e5ece 91401d2c-0784-4f43-824c-34f94a33c24d

Також можна повторити всі невдалі завдання конкретної черги:

php artisan queue:retry --queue=name

Щоб повторити всі невдалі завдання, виконайте команду queue:retry, передавши all замість ID:

php artisan queue:retry all

Якщо ви хочете видалити невдале завдання, скористайтеся командою queue:forget:

php artisan queue:forget 91401d2c-0784-4f43-824c-34f94a33c24d

[!NOTE] При використанні Horizon для видалення невдалого завдання слід використовувати команду horizon:forget, а не queue:forget.

Щоб видалити всі невдалі завдання з таблиці failed_jobs, скористайтеся командою queue:flush:

php artisan queue:flush

Команда queue:flush видаляє з черги всі записи про невдалі завдання, незалежно від того, наскільки давно вони сталися. Опція --hours дозволяє видалити лише ті завдання, які зазнали невдачі задану кількість годин тому або раніше:

php artisan queue:flush --hours=48

Ігнорування відсутніх моделей

Коли ви впроваджуєте модель Eloquent у завдання, модель автоматично серіалізується перед потраплянням у чергу й повторно вибирається з бази даних під час обробки завдання. Проте якщо модель була видалена, поки завдання чекало на обробника, завдання може зазнати невдачі з ModelNotFoundException.

Для зручності ви можете автоматично видаляти завдання з відсутніми моделями за допомогою атрибута DeleteWhenMissingModels у класі завдання. Коли цей атрибут присутній, Laravel тихо відкине завдання, не кидаючи винятку:

<?php

namespace App\Jobs;

use Illuminate\Queue\Attributes\DeleteWhenMissingModels;

#[DeleteWhenMissingModels]
class ProcessPodcast implements ShouldQueue
{
    // ...
}

Очищення невдалих завдань

Ви можете очистити записи в таблиці failed_jobs вашого застосунку, викликавши Artisan-команду queue:prune-failed:

php artisan queue:prune-failed

За замовчуванням буде видалено всі записи про невдалі завдання, старші за 24 години. Якщо передати команді опцію --hours, збережуться лише ті записи, які були додані за останні N годин. Наприклад, така команда видалить усі записи про невдалі завдання, додані понад 48 годин тому:

php artisan queue:prune-failed --hours=48

Зберігання невдалих завдань у DynamoDB

Laravel також підтримує зберігання записів про невдалі завдання в DynamoDB замість таблиці реляційної бази даних. Проте таблицю DynamoDB для зберігання всіх таких записів доведеться створити вручну. Зазвичай вона має називатися failed_jobs, але назву слід узгодити зі значенням конфігурації queue.failed.table у файлі конфігурації queue вашого застосунку.

Таблиця failed_jobs повинна мати рядковий первинний ключ розділу з назвою application і рядковий первинний ключ сортування з назвою uuid. Частина ключа application міститиме назву вашого застосунку, визначену значенням конфігурації name у файлі конфігурації app. Оскільки назва застосунку є частиною ключа таблиці DynamoDB, ту саму таблицю можна використовувати для зберігання невдалих завдань кількох застосунків Laravel.

Крім того, встановіть AWS SDK, щоб ваш застосунок Laravel міг взаємодіяти з Amazon DynamoDB:

composer require aws/aws-sdk-php

Далі задайте опції конфігурації queue.failed.driver значення dynamodb. Крім того, у масиві конфігурації невдалих завдань визначте опції key, secret і region. Вони використовуються для автентифікації в AWS. При використанні драйвера dynamodb опція конфігурації queue.failed.database не потрібна:

'failed' => [
    'driver' => env('QUEUE_FAILED_DRIVER', 'dynamodb'),
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'table' => 'failed_jobs',
],

Вимкнення зберігання невдалих завдань

Ви можете вказати Laravel відкидати невдалі завдання без збереження, задавши опції конфігурації queue.failed.driver значення null. Зазвичай це роблять через змінну оточення QUEUE_FAILED_DRIVER:

QUEUE_FAILED_DRIVER=null

Події невдалих завдань

Якщо ви хочете зареєструвати слухач події, який викликається при невдачі завдання, скористайтеся методом failing фасаду Queue. Наприклад, ми можемо приєднати замикання до цієї події в методі boot класу AppServiceProvider, що входить у поставку Laravel:

<?php

namespace App\Providers;

use Illuminate\Support\Facades\Queue;
use Illuminate\Support\ServiceProvider;
use Illuminate\Queue\Events\JobFailed;

class AppServiceProvider extends ServiceProvider
{
    /**
     * Зареєструвати будь-які сервіси застосунку.
     */
    public function register(): void
    {
        // ...
    }

    /**
     * Завантажити будь-які сервіси застосунку.
     */
    public function boot(): void
    {
        Queue::failing(function (JobFailed $event) {
            // $event->connectionName
            // $event->job
            // $event->exception
        });
    }
}

Очищення завдань із черг

[!NOTE] При використанні Horizon для очищення завдань із черги слід використовувати команду horizon:clear, а не queue:clear.

Якщо ви хочете видалити всі завдання з черги за замовчуванням для з'єднання за замовчуванням, скористайтеся Artisan-командою queue:clear:

php artisan queue:clear

Ви також можете передати аргумент connection і опцію queue, щоб видалити завдання з конкретного з'єднання й черги:

php artisan queue:clear redis --queue=emails

[!WARNING] Очищення завдань із черг доступне лише для драйверів черг SQS, Redis і database. Крім того, видалення повідомлень у SQS триває до 60 секунд, тож завдання, надіслані в чергу SQS протягом 60 секунд після очищення, теж можуть бути видалені.

Моніторинг черг

Якщо ваша черга раптово отримує наплив завдань, вона може захлинутися, і час очікування завершення завдань стане дуже довгим. За бажання Laravel може сповіщати вас, коли кількість завдань у черзі перевищує заданий поріг.

Для початку заплануйте команду queue:monitor на щохвилинний запуск. Команда приймає назви черг, які ви хочете моніторити, а також бажаний поріг кількості завдань:

php artisan queue:monitor redis:default,redis:deployments --max=100

Самого лише планування цієї команди недостатньо, щоб отримати сповіщення про перевантаження черги. Коли команда натрапляє на чергу, кількість завдань у якій перевищує ваш поріг, диспатчиться подія Illuminate\Queue\Events\QueueBusy. Ви можете слухати цю подію в AppServiceProvider свого застосунку, щоб надіслати сповіщення собі або команді розробників:

use App\Notifications\QueueHasLongWaitTime;
use Illuminate\Queue\Events\QueueBusy;
use Illuminate\Support\Facades\Event;
use Illuminate\Support\Facades\Notification;

/**
 * Завантажити будь-які сервіси застосунку.
 */
public function boot(): void
{
    Event::listen(function (QueueBusy $event) {
        Notification::route('mail', 'dev@example.com')
            ->notify(new QueueHasLongWaitTime(
                $event->connectionName,
                $event->queue,
                $event->size
            ));
    });
}

Тестування

Тестуючи код, що відправляє завдання, ви можете захотіти, щоб Laravel не виконував саме завдання, адже код завдання можна протестувати окремо від коду, який його відправляє. Звісно, щоб протестувати саме завдання, ви можете створити його екземпляр і викликати метод handle напряму в тесті.

Метод fake фасаду Queue не дає завданням реально потрапляти в чергу. Після виклику fake ви можете перевіряти, що застосунок намагався відправити завдання в чергу:

<?php

use App\Jobs\AnotherJob;
use App\Jobs\ShipOrder;
use Illuminate\Support\Facades\Queue;

test('orders can be shipped', function () {
    Queue::fake();

    // Виконати відправлення замовлення...

    // Перевірити, що жодного завдання не було відправлено...
    Queue::assertNothingPushed();

    // Перевірити, що завдання відправлено у вказану чергу...
    Queue::assertPushedOn('queue-name', ShipOrder::class);

    // Перевірити, що завдання було відправлено
    Queue::assertPushed(ShipOrder::class);

    // Перевірити, що завдання відправлено рівно один раз...
    Queue::assertPushedOnce(ShipOrder::class);

    // Перевірити, що завдання відправлено двічі...
    Queue::assertPushedTimes(ShipOrder::class, 2);

    // Перевірити, що завдання не відправлялося...
    Queue::assertNotPushed(AnotherJob::class);

    // Перевірити, що замикання було відправлено в чергу...
    Queue::assertClosurePushed();

    // Перевірити, що замикання не відправлялося...
    Queue::assertClosureNotPushed();

    // Перевірити загальну кількість відправлених завдань...
    Queue::assertCount(3);
});

<?php

namespace Tests\Feature;

use App\Jobs\AnotherJob;
use App\Jobs\ShipOrder;
use Illuminate\Support\Facades\Queue;
use Tests\TestCase;

class ExampleTest extends TestCase
{
    public function test_orders_can_be_shipped(): void
    {
        Queue::fake();

        // Виконати відправлення замовлення...

        // Перевірити, що жодного завдання не було відправлено...
        Queue::assertNothingPushed();

        // Перевірити, що завдання відправлено у вказану чергу...
        Queue::assertPushedOn('queue-name', ShipOrder::class);

        // Перевірити, що завдання було відправлено
        Queue::assertPushed(ShipOrder::class);

        // Перевірити, що завдання відправлено рівно один раз...
        Queue::assertPushedOnce(ShipOrder::class);

        // Перевірити, що завдання відправлено двічі...
        Queue::assertPushedTimes(ShipOrder::class, 2);

        // Перевірити, що завдання не відправлялося...
        Queue::assertNotPushed(AnotherJob::class);

        // Перевірити, що замикання було відправлено в чергу...
        Queue::assertClosurePushed();

        // Перевірити, що замикання не відправлялося...
        Queue::assertClosureNotPushed();

        // Перевірити загальну кількість відправлених завдань...
        Queue::assertCount(3);
    }
}

Ви можете передати замикання в методи assertPushed, assertNotPushed, assertClosurePushed або assertClosureNotPushed, щоб перевірити, що було відправлено завдання, яке проходить заданий «тест на істинність». Якщо хоча б одне відправлене завдання проходить цей тест, перевірка буде успішною:

use Illuminate\Queue\CallQueuedClosure;

Queue::assertPushed(function (ShipOrder $job) use ($order) {
    return $job->order->id === $order->id;
});

Queue::assertClosurePushed(function (CallQueuedClosure $job) {
    return $job->name === 'validate-order';
});

Підміна частини завдань

Якщо вам треба підмінити лише певні завдання, а решту виконувати як завжди, передайте методу fake назви класів завдань, які слід підмінити:

test('orders can be shipped', function () {
    Queue::fake([
        ShipOrder::class,
    ]);

    // Виконати відправлення замовлення...

    // Перевірити, що завдання відправлено двічі...
    Queue::assertPushedTimes(ShipOrder::class, 2);
});

public function test_orders_can_be_shipped(): void
{
    Queue::fake([
        ShipOrder::class,
    ]);

    // Виконати відправлення замовлення...

    // Перевірити, що завдання відправлено двічі...
    Queue::assertPushedTimes(ShipOrder::class, 2);
}

Підмінити всі завдання, окрім заданого набору, можна методом except:

Queue::fake()->except([
    ShipOrder::class,
]);

Тестування ланцюжків завдань

Для тестування ланцюжків завдань вам знадобляться можливості підміни фасаду Bus. Метод assertChained фасаду Bus дозволяє перевірити, що ланцюжок завдань було відправлено. Першим аргументом метод assertChained приймає масив завдань ланцюжка:

use App\Jobs\RecordShipment;
use App\Jobs\ShipOrder;
use App\Jobs\UpdateInventory;
use Illuminate\Support\Facades\Bus;

Bus::fake();

// ...

Bus::assertChained([
    ShipOrder::class,
    RecordShipment::class,
    UpdateInventory::class
]);

Як видно з прикладу вище, масив завдань ланцюжка може бути масивом назв класів завдань. Проте ви можете передати й масив реальних екземплярів завдань. У такому разі Laravel перевірить, що екземпляри завдань належать до того самого класу і мають ті самі значення властивостей, що й завдання ланцюжка, відправленого вашим застосунком:

Bus::assertChained([
    new ShipOrder,
    new RecordShipment,
    new UpdateInventory,
]);

Метод assertDispatchedWithoutChain дозволяє перевірити, що завдання було відправлено без ланцюжка завдань:

Bus::assertDispatchedWithoutChain(ShipOrder::class);

Тестування змін ланцюжка

Якщо завдання в ланцюжку додає завдання на початок або в кінець наявного ланцюжка, ви можете скористатися методом завдання assertHasChain, щоб перевірити, що завдання має очікуваний ланцюжок решти завдань:

$job = new ProcessPodcast;

$job->handle();

$job->assertHasChain([
    new TranscribePodcast,
    new OptimizePodcast,
    new ReleasePodcast,
]);

Метод assertDoesntHaveChain дозволяє перевірити, що решта ланцюжка завдання порожня:

$job->assertDoesntHaveChain();

Тестування пакетів у ланцюжку

Якщо ваш ланцюжок завдань містить пакет завдань, ви можете перевірити, що пакет у ланцюжку відповідає вашим очікуванням, вставивши визначення Bus::chainedBatch усередину перевірки ланцюжка:

use App\Jobs\ShipOrder;
use App\Jobs\UpdateInventory;
use Illuminate\Bus\PendingBatch;
use Illuminate\Support\Facades\Bus;

Bus::assertChained([
    new ShipOrder,
    Bus::chainedBatch(function (PendingBatch $batch) {
        return $batch->jobs->count() === 3;
    }),
    new UpdateInventory,
]);

Тестування пакетів завдань

Метод assertBatched фасаду Bus дозволяє перевірити, що пакет завдань було відправлено. Замикання, передане методу assertBatched, отримує екземпляр Illuminate\Bus\PendingBatch, за яким можна дослідити завдання всередині пакета:

use Illuminate\Bus\PendingBatch;
use Illuminate\Support\Facades\Bus;

Bus::fake();

// ...

Bus::assertBatched(function (PendingBatch $batch) {
    return $batch->name == 'Import CSV' &&
           $batch->jobs->count() === 10;
});

Метод hasJobs на очікуваному пакеті дозволяє переконатися, що пакет містить потрібні завдання. Метод приймає масив екземплярів завдань, назв класів або замикань:

Bus::assertBatched(function (PendingBatch $batch) {
    return $batch->hasJobs([
        new ProcessCsvRow(row: 1),
        new ProcessCsvRow(row: 2),
        new ProcessCsvRow(row: 3),
    ]);
});

Коли використовуються замикання, замикання отримує екземпляр завдання. Очікуваний тип завдання виводиться з підказки типу замикання:

Bus::assertBatched(function (PendingBatch $batch) {
    return $batch->hasJobs([
        fn (ProcessCsvRow $job) => $job->row === 1,
        fn (ProcessCsvRow $job) => $job->row === 2,
        fn (ProcessCsvRow $job) => $job->row === 3,
    ]);
});

Метод assertBatchCount дозволяє перевірити, що було відправлено задану кількість пакетів:

Bus::assertBatchCount(3);

Метод assertNothingBatched дозволяє перевірити, що жодного пакета не було відправлено:

Bus::assertNothingBatched();

Тестування взаємодії завдання й пакета

Крім того, іноді треба протестувати взаємодію окремого завдання з його пакетом. Наприклад, вам може знадобитися перевірити, чи завдання скасувало подальшу обробку свого пакета. Для цього призначте завданню фейковий пакет методом withFakeBatch. Метод withFakeBatch повертає кортеж з екземпляра завдання й фейкового пакета:

[$job, $batch] = (new ShipOrder)->withFakeBatch();

$job->handle();

$this->assertTrue($batch->cancelled());
$this->assertEmpty($batch->added);

Тестування взаємодії завдання й черги

Іноді потрібно перевірити, що завдання в черзі повертає себе назад у чергу. Або що завдання видалило саме себе. Такі взаємодії з чергою можна протестувати, створивши екземпляр завдання й викликавши метод withFakeQueueInteractions.

Після підміни взаємодій завдання з чергою ви можете викликати метод handle на завданні. Після виклику завдання доступні різні методи перевірок, щоб верифікувати взаємодії завдання з чергою:

use App\Exceptions\CorruptedAudioException;
use App\Jobs\ProcessPodcast;

$job = (new ProcessPodcast)->withFakeQueueInteractions();

$job->handle();

$job->assertReleased(delay: 30);
$job->assertDeleted();
$job->assertNotDeleted();
$job->assertFailed();
$job->assertFailedWith(CorruptedAudioException::class);
$job->assertNotFailed();

Події завдань

Методи before і after фасаду Queue дозволяють задати колбеки, які виконаються до або після обробки завдання з черги. У таких колбеках зручно вести додаткове логування чи збільшувати лічильники для дашборда. Зазвичай ці методи викликають у методі boot сервіс-провайдера. Наприклад, скористаємось AppServiceProvider, який входить до складу Laravel:

<?php

namespace App\Providers;

use Illuminate\Support\Facades\Queue;
use Illuminate\Support\ServiceProvider;
use Illuminate\Queue\Events\JobProcessed;
use Illuminate\Queue\Events\JobProcessing;

class AppServiceProvider extends ServiceProvider
{
    /**
     * Реєстрація сервісів застосунку.
     */
    public function register(): void
    {
        // ...
    }

    /**
     * Ініціалізація сервісів застосунку.
     */
    public function boot(): void
    {
        Queue::before(function (JobProcessing $event) {
            // $event->connectionName
            // $event->job
            // $event->job->payload()
        });

        Queue::after(function (JobProcessed $event) {
            // $event->connectionName
            // $event->job
            // $event->job->payload()
        });
    }
}

Метод looping фасаду Queue задає колбеки, які виконуються перед тим, як обробник черги спробує взяти завдання з черги. Наприклад, можна зареєструвати замикання, яке відкотить транзакції, залишені відкритими попереднім невдалим завданням:

use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Queue;

Queue::looping(function () {
    while (DB::transactionLevel() > 0) {
        DB::rollBack();
    }
});

Laravel також надсилає подію Illuminate\Queue\Events\WorkerIdle, коли обробник черги не може отримати завдання з черги:

use Illuminate\Queue\Events\WorkerIdle;
use Illuminate\Support\Facades\Event;

Event::listen(function (WorkerIdle $event) {
    // $event->connectionName
    // $event->queue
    // $event->workerOptions
});