Очереди

• Введение
  • Подключения и очереди
  • Особенности драйверов и требования
• Создание заданий
  • Генерация классов заданий
  • Структура класса
  • Уникальные задания
  • Зашифрованные задания
• Промежуточное ПО для заданий
  • Ограничение скорости
  • Предотвращение наложения заданий
  • Ограничение исключений
• Отправка заданий
  • Отложенная отправка
  • Синхронная отправка
  • Задания и транзакции базы данных
  • Цепочки заданий
  • Настройка очереди и подключения
  • Указание максимального количества попыток и таймаутов
  • Обработка ошибок
• Пакетная обработка заданий
  • Определение пакетируемых заданий
  • Отправка пакетов
  • Цепочки и пакеты
  • Добавление заданий в пакеты
  • Просмотр пакетов
  • Отмена пакетов
  • Ошибки в пакетах
  • Очистка пакетов
  • Хранение пакетов в DynamoDB
• Очередь замыканий
• Запуск обработчика очереди
  • Команда queue:work
  • Приоритеты очередей
  • Обработчики очередей и деплой
  • Истечение срока и таймауты заданий
• Настройка Supervisor
• Работа с неудачными заданиями
  • Очистка после неудачных заданий
  • Повторное выполнение неудачных заданий
  • Игнорирование отсутствующих моделей
  • Очистка неудачных заданий
  • Хранение неудачных заданий в DynamoDB
  • Отключение хранения неудачных заданий
  • События неудачных заданий
• Очистка заданий из очередей
• Мониторинг очередей
• Тестирование
  • Фальсификация части заданий
  • Тестирование цепочек заданий
  • Тестирование пакетных заданий
• События заданий

#Введение

При разработке веб-приложения могут возникать задачи, например, разбор и сохранение загруженного CSV-файла, которые занимают слишком много времени для выполнения в рамках обычного веб-запроса. К счастью, Laravel позволяет легко создавать задания, помещаемые в очередь и обрабатываемые в фоновом режиме. Перемещая ресурсоёмкие задачи в очередь, ваше приложение сможет отвечать на веб-запросы с высокой скоростью и обеспечивать лучший пользовательский опыт.

Очереди Laravel предоставляют единый API для работы с различными бэкендами очередей, такими как Amazon SQS, Redis или даже реляционная база данных.

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

#Подключения и очереди

Перед началом работы с очередями Laravel важно понять разницу между «подключениями» и «очередями». В вашем файле конфигурации 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 вам потребуется таблица в базе данных для хранения заданий. Чтобы создать миграцию для этой таблицы, выполните Artisan-команду queue:table. После создания миграции выполните миграцию базы данных командой migrate:

php artisan queue:table

php artisan migrate

Наконец, не забудьте указать приложению использовать драйвер database, обновив переменную QUEUE_CONNECTION в файле .env вашего приложения:

QUEUE_CONNECTION=database

#Redis

Для использования драйвера очереди redis необходимо настроить подключение к базе данных Redis в файле конфигурации config/database.php.

Кластер Redis

Если ваше подключение к очереди Redis использует кластер Redis, имена очередей должны содержать хеш-тег ключа. Это необходимо, чтобы все ключи Redis для данной очереди находились в одном хеш-слоте:

'redis' => [
    'driver' => 'redis',
    'connection' => 'default',
    'queue' => '{default}',
    'retry_after' => 90,
],

Блокировка

При использовании очереди Redis можно использовать опцию конфигурации block_for, чтобы указать, как долго драйвер должен ждать появления задания, прежде чем повторно опросить базу Redis в цикле обработчика.

Настройка этого значения в зависимости от нагрузки на очередь может быть эффективнее, чем постоянный опрос базы Redis на наличие новых заданий. Например, можно установить значение 5, чтобы драйвер блокировался на пять секунд в ожидании задания:

'redis' => [
    'driver' => 'redis',
    'connection' => 'default',
    'queue' => 'default',
    'retry_after' => 90,
    'block_for' => 5,
],

#Другие требования к драйверам

Для следующих драйверов очередей требуются соответствующие зависимости. Их можно установить через менеджер пакетов Composer:

#Создание заданий

#Генерация классов заданий

По умолчанию все задания, помещаемые в очередь, хранятся в каталоге app/Jobs. Если каталога app/Jobs нет, он будет создан при выполнении Artisan-команды make:job:

php artisan make:job ProcessPodcast

Сгенерированный класс будет реализовывать интерфейс Illuminate\Contracts\Queue\ShouldQueue, что указывает Laravel, что задание должно быть помещено в очередь для асинхронного выполнения.

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

Классы заданий очень просты, обычно содержат только метод handle, который вызывается при обработке задания из очереди. Для начала рассмотрим пример класса задания. В этом примере предположим, что мы управляем сервисом публикации подкастов и нам нужно обработать загруженные файлы подкастов перед публикацией:

<?php

namespace App\Jobs;

use App\Models\Podcast;
use App\Services\AudioProcessor;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

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

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

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

В этом примере обратите внимание, что мы можем передать Eloquent-модель напрямую в конструктор задания. Благодаря трейту SerializesModels, который используется в задании, модели Eloquent и их загруженные связи будут корректно сериализованы и десериализованы при обработке задания.

Если ваше задание принимает модель Eloquent в конструкторе, в очередь сериализуется только идентификатор модели. При фактической обработке задания система очереди автоматически заново извлечёт полный экземпляр модели и её загруженные связи из базы данных. Такой подход к сериализации моделей позволяет значительно уменьшить размер полезной нагрузки задания, отправляемой драйверу очереди.

#Внедрение зависимостей в метод handle

Метод handle вызывается при обработке задания из очереди. Обратите внимание, что мы можем указывать типы зависимостей в методе handle. Laravel контейнер сервисов автоматически внедряет эти зависимости.

Если вы хотите полностью контролировать, как контейнер внедряет зависимости в метод handle, вы можете использовать метод bindMethod контейнера. Метод bindMethod принимает callback, который получает задание и контейнер. Внутри callback вы можете вызвать метод 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));
});

#Связи в очереди

Поскольку все загруженные связи моделей Eloquent также сериализуются при постановке задания в очередь, сериализованная строка задания может стать довольно большой. Кроме того, при десериализации задания и повторном извлечении связей из базы данных они будут загружены полностью. Любые ограничения связей, применённые до сериализации модели при постановке в очередь, не будут применены при десериализации. Поэтому, если вы хотите работать с подмножеством связи, следует повторно наложить ограничения внутри задания.

Или, чтобы предотвратить сериализацию связей, можно вызвать метод withoutRelations у модели при установке значения свойства. Этот метод вернёт экземпляр модели без загруженных связей:

/**
 * Создать новый экземпляр задания.
 */
public function __construct(Podcast $podcast)
{
    $this->podcast = $podcast->withoutRelations();
}

Если вы используете продвижение свойств в конструкторе PHP и хотите указать, что связи модели Eloquent не должны сериализоваться, можно использовать атрибут WithoutRelations:

use Illuminate\Queue\Attributes\WithoutRelations;

/**
 * Создать новый экземпляр задания.
 */
public function __construct(
    #[WithoutRelations]
    public Podcast $podcast
) {
}

Если задание получает коллекцию или массив моделей Eloquent вместо одной модели, связи моделей в этой коллекции не будут восстановлены при десериализации и выполнении задания. Это сделано для предотвращения чрезмерного использования ресурсов при работе с большим количеством моделей.

#Уникальные задания

Иногда нужно гарантировать, что в очереди одновременно находится только один экземпляр конкретного задания. Для этого можно реализовать интерфейс ShouldBeUnique в классе задания. Этот интерфейс не требует определения дополнительных методов:

<?php

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

class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    ...
}

В приведённом примере задание UpdateSearchIndex является уникальным. Поэтому оно не будет отправлено в очередь, если другой экземпляр этого задания уже находится в очереди и ещё не завершил обработку.

В некоторых случаях вы можете захотеть определить конкретный «ключ», который делает задание уникальным, или указать таймаут, по истечении которого задание перестаёт быть уникальным. Для этого можно определить свойства или методы uniqueId и uniqueFor в классе задания:

<?php

use App\Models\Product;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Contracts\Queue\ShouldBeUnique;

class UpdateSearchIndex implements ShouldQueue, ShouldBeUnique
{
    /**
     * Экземпляр продукта.
     *
     * @var \App\Product
     */
    public $product;

    /**
     * Количество секунд, после которых уникальная блокировка задания будет снята.
     *
     * @var int
     */
    public $uniqueFor = 3600;

    /**
     * Получить уникальный ID для задания.
     */
    public function uniqueId(): string
    {
        return $this->product->id;
    }
}

В приведённом примере задание UpdateSearchIndex уникально по ID продукта. Таким образом, любые новые отправки задания с тем же ID продукта будут игнорироваться, пока существующее задание не завершит обработку. Кроме того, если существующее задание не будет обработано в течение часа, уникальная блокировка будет снята, и другое задание с тем же уникальным ключом может быть отправлено в очередь.

#Сохранение уникальности заданий до начала обработки

По умолчанию уникальные задания «разблокируются» после завершения обработки или после неудачных попыток повторения. Однако бывают ситуации, когда нужно разблокировать задание сразу перед обработкой. Для этого задание должно реализовать контракт ShouldBeUniqueUntilProcessing вместо ShouldBeUnique:

<?php

use App\Models\Product;
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
{
    ...

    /**
     * Получить драйвер кеша для уникальной блокировки задания.
     */
    public function uniqueVia(): Repository
    {
        return Cache::driver('redis');
    }
}

#Зашифрованные задания

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

<?php

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

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

#Промежуточное ПО для заданий

Промежуточное ПО для заданий позволяет обернуть выполнение очередных заданий в пользовательскую логику, уменьшая дублирование кода в самих заданиях. Например, рассмотрим следующий метод handle, который использует возможности ограничения скорости Redis в Laravel, чтобы разрешить обработку только одного задания каждые пять секунд:

use Illuminate\Support\Facades\Redis;

/**
 * Выполнить задание.
 */
public function handle(): void
{
    Redis::throttle('key')->block(0)->allow(1)->every(5)->then(function () {
        info('Блокировка получена...');

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

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

Хотя этот код корректен, реализация метода handle становится громоздкой из-за логики ограничения скорости Redis. Кроме того, эту логику придётся дублировать для всех заданий, которые нужно ограничивать.

Вместо ограничения частоты запросов в методе handle можно определить middleware для заданий, которое будет выполнять это ограничение. В Laravel нет стандартного места для размещения middleware для заданий, поэтому их можно поместить в любом месте приложения. В этом примере мы положим middleware в директорию app/Jobs/Middleware:

<?php

namespace App\Jobs\Middleware;

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

class RateLimited
{
    /**
     * Обработать задание из очереди.
     *
     * @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, промежуточное ПО для заданий получает обрабатываемое задание и callback, который нужно вызвать для продолжения обработки.

После создания промежуточного ПО для заданий его можно прикрепить к заданию, вернув из метода middleware задания. Этот метод отсутствует в заданиях, созданных командой make:job, поэтому его нужно добавить вручную:

use App\Jobs\Middleware\RateLimited;

/**
 * Получить промежуточное ПО, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited];
}

#Ограничение скорости

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

Например, вы можете разрешить пользователям создавать резервные копии данных не чаще одного раза в час, при этом не ограничивая премиум-клиентов. Для этого можно определить RateLimiter в методе boot вашего AppServiceProvider:

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

/**
 * Загрузить сервисы приложения.
 */
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;

/**
 * Получить промежуточное ПО, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new RateLimited('backups')];
}

Возврат задания с ограничением скорости обратно в очередь увеличит общее число attempts у задания. Возможно, вам потребуется настроить свойства tries и maxExceptions в классе задания. Либо вы можете использовать метод retryUntil, чтобы задать время, до которого задание будет пытаться выполниться.

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

/**
 * Получить промежуточное ПО, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new RateLimited('backups'))->dontRelease()];
}

#Предотвращение перекрытия заданий

В Laravel есть middleware Illuminate\Queue\Middleware\WithoutOverlapping, который позволяет предотвратить перекрытие заданий на основе произвольного ключа. Это полезно, когда в очереди есть задание, изменяющее ресурс, который должен изменяться только одним заданием одновременно.

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

use Illuminate\Queue\Middleware\WithoutOverlapping;

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new WithoutOverlapping($this->user->id)];
}

Любые перекрывающиеся задания того же типа будут возвращены обратно в очередь. Вы также можете указать количество секунд, которое должно пройти перед повторной попыткой выполнения возвращённого задания:

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->releaseAfter(60)];
}

Если вы хотите сразу удалить перекрывающиеся задания, чтобы они не выполнялись повторно, используйте метод dontRelease:

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->dontRelease()];
}

Middleware WithoutOverlapping работает на основе атомарных блокировок Laravel. Иногда задание может неожиданно завершиться с ошибкой или тайм-аутом так, что блокировка не будет снята. Поэтому вы можете явно задать время истечения блокировки с помощью метода expireAfter. Например, в примере ниже Laravel снимет блокировку WithoutOverlapping через три минуты после начала обработки задания:

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new WithoutOverlapping($this->order->id))->expireAfter(180)];
}

#Использование общих ключей блокировок для разных классов заданий

По умолчанию 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;

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [new ThrottlesExceptions(10, 5)];
}

/**
 * Определить время, до которого задание может повторяться.
 */
public function retryUntil(): DateTime
{
    return now()->addMinutes(5);
}

Первый аргумент конструктора middleware — количество исключений, которые задание может выбросить до ограничения, а второй — количество минут, которое должно пройти перед повторной попыткой после ограничения. В приведённом примере, если задание выбросит 10 исключений за 5 минут, мы подождём 5 минут перед следующей попыткой.

Если задание выбрасывает исключение, но порог исключений ещё не достигнут, задание обычно повторяется сразу. Однако вы можете указать количество минут задержки, вызвав метод backoff при добавлении middleware к заданию:

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 5))->backoff(5)];
}

Внутри этот middleware использует систему кеша Laravel для реализации ограничения скорости, а имя класса задания используется как ключ кеша. Вы можете переопределить этот ключ, вызвав метод by при добавлении middleware к заданию. Это полезно, если у вас несколько заданий, взаимодействующих с одним и тем же сторонним сервисом, и вы хотите, чтобы они разделяли общее ограничение:

use Illuminate\Queue\Middleware\ThrottlesExceptions;

/**
 * Получить middleware, через которое должно пройти задание.
 *
 * @return array<int, object>
 */
public function middleware(): array
{
    return [(new ThrottlesExceptions(10, 10))->by('key')];
}

#Отправка заданий в очередь

После того как вы создали класс задания, вы можете отправить его в очередь с помощью метода dispatch самого задания. Аргументы, переданные в dispatch, будут переданы в конструктор задания:

<?php

namespace App\Http\Controllers;

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

class PodcastController extends Controller
{
    /**
     * Сохранить новый подкаст.
     */
    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 драйвер очереди по умолчанию — sync. Этот драйвер выполняет задания синхронно в рамках текущего запроса, что удобно при локальной разработке. Если вы хотите начать использовать очереди для фоновой обработки, укажите другой драйвер в конфигурационном файле config/queue.php.

#Отложенная отправка

Если вы хотите, чтобы задание не было доступно для обработки сразу после отправки, используйте метод delay. Например, укажем, что задание должно стать доступным для обработки через 10 минут после отправки:

<?php

namespace App\Http\Controllers;

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

class PodcastController extends Controller
{
    /**
     * Сохранить новый подкаст.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // ...

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

        return redirect('/podcasts');
    }
}

#Отправка после ответа браузеру

Метод dispatchAfterResponse откладывает отправку задания до тех пор, пока HTTP-ответ не будет отправлен браузеру, если ваш веб-сервер использует FastCGI. Это позволяет пользователю начать работу с приложением, даже если задание в очереди ещё выполняется. Обычно этот метод используют для заданий, которые выполняются около секунды, например, отправка письма. Поскольку такие задания обрабатываются в рамках текущего HTTP-запроса, для их выполнения не требуется запущенный воркер очереди:

use App\Jobs\SendNotification;

SendNotification::dispatchAfterResponse();

Вы также можете вызвать dispatch с замыканием и цепочкой добавить метод afterResponse к хелперу dispatch, чтобы выполнить замыкание после отправки HTTP-ответа в браузер:

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

dispatch(function () {
    Mail::to('taylor@example.com')->send(new WelcomeMessage);
})->afterResponse();

#Синхронная отправка

Если вы хотите выполнить задание немедленно (синхронно), используйте метод dispatchSync. При этом задание не будет помещено в очередь, а выполнится сразу в текущем процессе:

<?php

namespace App\Http\Controllers;

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

class PodcastController extends Controller
{
    /**
     * Сохранить новый подкаст.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Создать подкаст...

        ProcessPodcast::dispatchSync($podcast);

        return redirect('/podcasts');
    }
}

#Задания и транзакции базы данных

Хотя отправлять задания внутри транзакций базы данных допустимо, следует убедиться, что задание сможет успешно выполниться. При отправке задания внутри транзакции возможно, что воркер обработает задание до того, как родительская транзакция будет зафиксирована. В этом случае изменения моделей или записей базы данных, сделанные в транзакции, могут ещё не появиться в базе. Кроме того, модели или записи, созданные в транзакции, могут отсутствовать в базе.

К счастью, Laravel предлагает несколько способов решения этой проблемы. Во-первых, вы можете установить опцию after_commit в конфигурации подключения очереди:

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

Если опция after_commit установлена в true, вы можете отправлять задания внутри транзакций, но Laravel будет ждать, пока все открытые родительские транзакции не будут зафиксированы, прежде чем фактически отправить задание. Если же открытых транзакций нет, задание отправляется сразу.

Если транзакция откатывается из-за исключения, возникшего во время транзакции, задания, отправленные в её рамках, будут отброшены.

#Указание поведения отправки после фиксации транзакции inline

Если вы не установите опцию конфигурации подключения очереди 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();

#Указание подключения и очереди для цепочки

Если вы хотите указать подключение и очередь для цепочки заданий, используйте методы onConnection и onQueue. Они задают подключение и имя очереди, которые будут использоваться, если для заданий явно не указано другое подключение или очередь:

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

#Обработка ошибок в цепочке

При использовании цепочек вы можете вызвать метод catch, чтобы указать замыкание, которое будет вызвано при ошибке в одном из заданий цепочки. В колбэк будет передан объект Throwable, вызвавший ошибку:

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

Bus::chain([
    new ProcessPodcast,
    new OptimizePodcast,
    new ReleasePodcast,
])->catch(function (Throwable $e) {
    // В цепочке произошло исключение...
})->dispatch();

#Настройка очереди для подключения

#Отправка в конкретную очередь

Размещая задания в разных очередях, вы можете «категоризировать» задания и даже приоритизировать количество воркеров для разных очередей. Учтите, что это не переключает подключение к очереди, а лишь указывает конкретную очередь внутри одного подключения. Чтобы указать очередь, используйте метод onQueue при отправке задания:

<?php

namespace App\Http\Controllers;

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

class PodcastController extends Controller
{
    /**
     * Сохранить новый подкаст.
     */
    public function store(Request $request): RedirectResponse
    {
        $podcast = Podcast::create(/* ... */);

        // Создать подкаст...

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

        return redirect('/podcasts');
    }
}

В качестве альтернативы вы можете указать очередь задания, вызвав метод onQueue в конструкторе задания:

<?php

namespace App\Jobs;

 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;

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

    /**
     * Создать новый экземпляр задания.
     */
    public function __construct()
    {
        $this->onQueue('processing');
    }
}

#Отправка в конкретное подключение

Если ваше приложение работает с несколькими подключениями очередей, вы можете указать, в какое подключение отправлять задание, используя метод onConnection:

<?php

namespace App\Http\Controllers;

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

class PodcastController extends Controller
{
    /**
     * Сохранить новый подкаст.
     */
    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\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;

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

    /**
     * Создать новый экземпляр задания.
     */
    public function __construct()
    {
        $this->onConnection('sqs');
    }
}

#Указание максимального количества попыток и времени ожидания

#Максимальное количество попыток

Если одно из ваших заданий в очереди вызывает ошибку, скорее всего, вы не хотите, чтобы оно пыталось выполняться бесконечно. Поэтому Laravel предоставляет разные способы указать, сколько раз или как долго задание может пытаться выполниться.

Один из способов указать максимальное количество попыток — использовать переключатель --tries в командной строке Artisan. Это будет применяться ко всем заданиям, обрабатываемым воркером, если только само задание не задаёт своё максимальное количество попыток:

php artisan queue:work --tries=3

Если задание превысит максимальное количество попыток, оно будет считаться «неудачным». Подробнее о работе с неудачными заданиями смотрите в разделе обработка неудачных заданий. Если в команду queue:work передать --tries=0, задание будет пытаться выполняться бесконечно.

Вы можете более точно указать максимальное количество попыток, задав это значение в самом классе задания. В этом случае оно будет иметь приоритет над значением --tries из командной строки:

<?php

namespace App\Jobs;

class ProcessPodcast implements ShouldQueue
{
    /**
     * Количество попыток выполнения задания.
     *
     * @var int
     */
    public $tries = 5;
}

Если вам нужно динамически управлять максимальным количеством попыток для конкретного задания, вы можете определить метод tries:

/**
 * Определить количество попыток выполнения задания.
 */
public function tries(): int
{
    return 5;
}

#Попытки с ограничением по времени

Вместо указания количества попыток вы можете определить время, до которого задание может пытаться выполняться. Это позволяет выполнять задание любое количество раз в пределах заданного временного интервала. Чтобы указать время, до которого задание может повторяться, добавьте метод retryUntil в класс задания. Метод должен возвращать объект DateTime:

use DateTime;

/**
 * Определить время, до которого задание может повторяться.
 */
public function retryUntil(): DateTime
{
    return now()->addMinutes(10);
}

#Максимальное количество исключений

Иногда нужно указать, что задание может пытаться выполняться много раз, но должно провалиться, если количество необработанных исключений превысит заданное значение (в отличие от вызова метода release). Для этого определите свойство maxExceptions в классе задания:

<?php

namespace App\Jobs;

use Illuminate\Support\Facades\Redis;

class ProcessPodcast implements ShouldQueue
{
    /**
     * Количество попыток выполнения задания.
     *
     * @var int
     */
    public $tries = 25;

    /**
     * Максимальное количество необработанных исключений до провала задания.
     *
     * @var int
     */
    public $maxExceptions = 3;

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

В этом примере задание будет возвращено в очередь на 10 секунд, если не удастся получить блокировку Redis, и будет повторяться до 25 раз. Однако задание провалится, если будет выброшено 3 необработанных исключения.

#Тайм-аут

Часто вы примерно знаете, сколько времени должны занимать ваши задания. Поэтому Laravel позволяет указать значение «тайм-аут». По умолчанию тайм-аут равен 60 секундам. Если задание выполняется дольше указанного времени, воркер завершится с ошибкой. Обычно воркер автоматически перезапускается менеджером процессов, настроенным на сервере.

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

php artisan queue:work --timeout=30

Если задание превысит максимальное количество попыток из-за постоянных тайм-аутов, оно будет помечено как неудачное.

Вы также можете указать максимальное время выполнения задания в самом классе. В этом случае значение будет иметь приоритет над значением из командной строки:

<?php

namespace App\Jobs;

class ProcessPodcast implements ShouldQueue
{
    /**
     * Максимальное время выполнения задания в секундах.
     *
     * @var int
     */
    public $timeout = 120;
}

Иногда процессы с блокировкой ввода-вывода, такие как сокеты или исходящие HTTP-соединения, могут не учитывать заданный вами таймаут. Поэтому при использовании этих функций всегда следует пытаться указать таймаут через их API. Например, при использовании Guzzle всегда указывайте значения таймаута соединения и запроса.

#Отметка задания как неудачного при таймауте

Если вы хотите, чтобы задание помечалось как неудачное при превышении таймаута, вы можете определить свойство $failOnTimeout в классе задания:

/**
 * Указывает, должно ли задание помечаться как неудачное при таймауте.
 *
 * @var bool
 */
public $failOnTimeout = true;

#Обработка ошибок

Если во время обработки задания возникает исключение, задание автоматически возвращается в очередь для повторной попытки. Задание будет возвращаться до тех пор, пока не будет достигнуто максимальное количество попыток, разрешённых вашим приложением. Максимальное число попыток задаётся параметром --tries команды Artisan queue:work. Кроме того, максимальное количество попыток можно определить непосредственно в классе задания. Подробнее о работе с очередями смотрите в разделе запуск обработчика очереди.

#Ручной возврат задания в очередь

Иногда может потребоваться вручную вернуть задание в очередь, чтобы попытаться выполнить его позже. Это можно сделать, вызвав метод release:

/**
 * Выполнить задание.
 */
public function handle(): void
{
    // ...

    $this->release();
}

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

$this->release(10);

$this->release(now()->addSeconds(10));

#Ручное помечание задания как неудачного

Иногда необходимо вручную пометить задание как "неудачное". Для этого вызовите метод fail:

/**
 * Выполнить задание.
 */
public function handle(): void
{
    // ...

    $this->fail();
}

Если вы хотите пометить задание как неудачное из-за пойманного исключения, вы можете передать исключение в метод fail. Для удобства можно также передать строку с сообщением об ошибке, которая будет преобразована в исключение:

$this->fail($exception);

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

#Пакетная обработка заданий

Функция пакетной обработки заданий в Laravel позволяет легко выполнять группу заданий и затем выполнять определённое действие после завершения всей группы. Перед началом работы следует создать миграцию базы данных для таблицы, которая будет содержать метаинформацию о пакетах заданий, например, процент выполнения. Миграция может быть сгенерирована с помощью команды Artisan queue:batches-table:

php artisan queue:batches-table

php artisan migrate

#Определение пакетных заданий

Чтобы определить пакетное задание, создайте обычное очередь-ориентированное задание, но добавьте в класс задания трейд Illuminate\Bus\Batchable. Этот трейд предоставляет метод batch, который позволяет получить текущий пакет, в рамках которого выполняется задание:

<?php

namespace App\Jobs;

use Illuminate\Bus\Batchable;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;

class ImportCsv implements ShouldQueue
{
    use Batchable, Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    /**
     * Выполнить задание.
     */
    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;

ID пакета, доступный через свойство $batch->id, можно использовать для запроса информации о пакете через Laravel command bus после его отправки.

#Именование пакетов

Некоторые инструменты, такие как 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;

/**
 * Выполнить задание.
 */
public function handle(): void
{
    if ($this->batch()->cancelled()) {
        return;
    }

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

#Просмотр пакетов

Экземпляр 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 с информацией о пакете, включая прогресс выполнения. Это удобно для отображения статуса пакета в интерфейсе приложения.

Чтобы получить пакет по его ID, используйте метод 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:

/**
 * Выполнить задание.
 */
public function handle(): void
{
    if ($this->user->exceedsImportLimit()) {
        return $this->batch()->cancel();
    }

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

Как видно из предыдущих примеров, пакетные задания обычно проверяют, отменён ли их пакет, перед продолжением выполнения. Для удобства можно назначить задание middleware SkipIfBatchCancelled. Как следует из названия, этот middleware не позволит Laravel обрабатывать задание, если его пакет отменён:

use Illuminate\Queue\Middleware\SkipIfBatchCancelled;

/**
 * Получить middleware, через которые должно пройти задание.
 */
public function middleware(): array
{
    return [new SkipIfBatchCancelled];
}

#Ошибки в пакетах

При сбое задания в пакете вызывается обратный вызов catch (если он назначен). Этот обратный вызов вызывается только для первого неудачного задания в пакете.

#Разрешение ошибок

Когда задание в пакете неудачно, Laravel автоматически помечает пакет как "отменённый". При желании это поведение можно отключить, чтобы сбой задания не приводил к отмене пакета. Для этого вызовите метод allowFailures при отправке пакета:

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

#Повторная попытка неудачных заданий пакета

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

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

#Очистка пакетов

Без очистки таблица job_batches может быстро заполняться записями. Чтобы избежать этого, следует запланировать выполнение команды Artisan queue:prune-batches ежедневно:

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

По умолчанию будут очищены все завершённые пакеты старше 24 часов. Вы можете использовать опцию hours при вызове команды, чтобы задать период хранения данных пакетов. Например, следующая команда удалит все пакеты, завершённые более 48 часов назад:

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

Иногда в таблице jobs_batches могут накапливаться записи о пакетах, которые так и не были успешно завершены, например, если задание в пакете неудачно и не было успешно повторено. Вы можете указать команде queue:prune-batches очищать такие незавершённые записи с помощью опции unfinished:

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

Аналогично, в таблице jobs_batches могут накапливаться записи о пакетах, которые были отменены. Вы можете указать команде queue:prune-batches удалять эти записи об отменённых пакетах с помощью опции cancelled:

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

#Хранение пакетов в DynamoDB

Laravel также поддерживает хранение метаинформации о пакетах в DynamoDB вместо реляционной базы данных. Однако таблицу DynamoDB для хранения записей пакетов нужно создать вручную.

Обычно таблица должна называться job_batches, но имя таблицы следует задать в соответствии со значением конфигурации queue.batching.table в файле конфигурации queue вашего приложения.

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

Таблица job_batches должна иметь строковой первичный ключ раздела application и строковой первичный ключ сортировки id. Часть ключа application содержит имя вашего приложения, заданное в конфигурации name в файле app. Поскольку имя приложения является частью ключа таблицы DynamoDB, вы можете использовать одну таблицу для хранения пакетов нескольких приложений Laravel.

Кроме того, вы можете определить атрибут ttl для таблицы, если хотите использовать автоматическую очистку пакетов.

#Конфигурация DynamoDB

Далее установите AWS SDK, чтобы ваше приложение Laravel могло взаимодействовать с Amazon DynamoDB:

composer require aws/aws-sdk-php

Затем установите значение конфигурации queue.batching.driver в dynamodb. Также задайте параметры key, secret и region в массиве конфигурации batching. Эти параметры используются для аутентификации в AWS. При использовании драйвера dynamodb параметр queue.batching.database не нужен:

'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',
],

#Очистка пакетов в DynamoDB

При использовании DynamoDB для хранения информации о пакетах стандартные команды очистки пакетов для реляционных баз данных не работают. Вместо этого можно использовать встроенную функцию TTL DynamoDB для автоматического удаления записей о старых пакетах.

Если вы определили атрибут ttl в таблице DynamoDB, можно задать параметры конфигурации, чтобы указать Laravel, как очищать записи пакетов. Параметр queue.batching.ttl_attribute задаёт имя атрибута с TTL, а queue.batching.ttl — количество секунд, после которых запись о пакете может быть удалена из таблицы относительно времени последнего обновления записи:

'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 дней...
],

#Очередь замыканий

Вместо отправки класса задания в очередь, вы можете отправить замыкание. Это удобно для быстрых и простых задач, которые нужно выполнить вне текущего цикла запроса. При отправке замыканий в очередь содержимое кода замыкания криптографически подписывается, чтобы предотвратить его изменение в пути:

$podcast = App\Podcast::find(1);

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

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

use Throwable;

dispatch(function () use ($podcast) {
    $podcast->publish();
})->catch(function (Throwable $e) {
    // Это задание не удалось...
});

#Запуск обработчика очереди

#Команда queue:work

Laravel включает команду Artisan, которая запускает обработчик очереди и обрабатывает новые задания по мере их поступления. Вы можете запустить обработчик с помощью команды Artisan queue:work. Обратите внимание, что после запуска команда queue:work будет работать до тех пор, пока её не остановят вручную или не закроют терминал:

php artisan queue:work

При вызове команды queue:work можно добавить флаг -v, чтобы в выводе отображались ID обработанных заданий:

php artisan queue:work -v

Помните, что обработчики очереди — это долгоживущие процессы, которые хранят загруженное состояние приложения в памяти. Поэтому они не увидят изменения в коде после запуска. Во время деплоя обязательно перезапускайте обработчики очереди. Также учтите, что любое статическое состояние, созданное или изменённое приложением, не будет автоматически сбрасываться между заданиями.

В качестве альтернативы, вы можете запустить команду queue:listen. При использовании команды 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 workers — это долгоживущие процессы, они не увидят изменения в вашем коде без перезапуска. Поэтому самый простой способ развернуть приложение с использованием queue workers — перезапустить воркеры в процессе деплоя. Вы можете аккуратно перезапустить все воркеры, выполнив команду queue:restart:

php artisan queue:restart

Эта команда даст всем queue workers команду аккуратно завершить работу после обработки текущей задачи, чтобы не потерять ни одной задачи. Поскольку queue workers завершат работу при выполнении команды queue:restart, рекомендуется использовать менеджер процессов, например Supervisor, для автоматического перезапуска воркеров.

#Истечение срока и таймауты задач

#Истечение срока задачи

В вашем конфигурационном файле config/queue.php для каждого соединения очереди определяется опция retry_after. Эта опция указывает, сколько секунд соединение очереди должно ждать перед повторной попыткой обработки задачи, которая уже обрабатывается. Например, если значение retry_after установлено в 90, задача будет возвращена обратно в очередь, если она обрабатывается 90 секунд без удаления или освобождения. Обычно следует устанавливать retry_after равным максимальному разумному времени обработки задачи.

#Таймауты воркера

Команда Artisan queue:work предоставляет опцию --timeout. По умолчанию значение --timeout равно 60 секундам. Если задача обрабатывается дольше, чем указано в таймауте, воркер, обрабатывающий задачу, завершится с ошибкой. Обычно воркер автоматически перезапускается менеджером процессов, настроенным на вашем сервере:

php artisan queue:work --timeout=60

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

#Конфигурация 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

#Настройка 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 sqs --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 в конфигурации, чтобы указать нужное соединение очереди и параметры воркера.

#Запуск Supervisor

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

sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start "laravel-worker:*"

Для получения дополнительной информации о Supervisor обратитесь к документации Supervisor.

#Работа с неудачными задачами

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

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

php artisan queue:failed-table

php artisan migrate

При запуске процесса queue worker вы можете указать максимальное количество попыток выполнения задачи с помощью опции --tries команды queue:work. Если значение --tries не указано, задача будет попытана выполнить один раз или столько раз, сколько указано в свойстве $tries класса задачи:

php artisan queue:work redis --tries=3

С помощью опции --backoff вы можете указать, сколько секунд Laravel должен ждать перед повторной попыткой задачи, которая вызвала исключение. По умолчанию задача сразу возвращается в очередь для повторной попытки:

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

Если вы хотите настроить время ожидания перед повторной попыткой задачи индивидуально для каждой задачи, вы можете определить свойство backoff в классе задачи:

/**
 * Количество секунд ожидания перед повторной попыткой задачи.
 *
 * @var int
 */
public $backoff = 3;

Если вам нужна более сложная логика для определения времени ожидания, вы можете определить метод backoff в классе задачи:

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

Вы можете легко настроить "экспоненциальные" интервалы ожидания, возвращая массив значений из метода backoff. В этом примере задержка повторной попытки будет 1 секунда для первой, 5 секунд для второй, 10 секунд для третьей и 10 секунд для всех последующих, если попытки ещё остались:

/**
* Вычислить количество секунд ожидания перед повторной попыткой задачи.
*
* @return array<int, int>
*/
public function backoff(): array
{
    return [1, 5, 10];
}

#Очистка после неудачных задач

Если конкретное задание терпит неудачу, вы можете отправить уведомление пользователям или отменить действия, которые задание выполнило частично. Для этого вы можете определить метод failed в вашем классе задания. Экземпляр Throwable, который вызвал сбой задания, будет передан в метод failed:

<?php

namespace App\Jobs;

use App\Models\Podcast;
use App\Services\AudioProcessor;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Throwable;

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

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

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

    /**
     * Обработать ошибку задачи.
     */
    public function failed(?Throwable $exception): void
    {
        // Отправить уведомление пользователю об ошибке и т.д...
    }
}

#Повторные попытки неудачных задач

Чтобы просмотреть все неудачные задачи, добавленные в таблицу 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:

php artisan queue:retry all

Если вы хотите удалить неудачную задачу, используйте команду queue:forget:

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

Чтобы удалить все неудачные задачи из таблицы failed_jobs, используйте команду queue:flush:

php artisan queue:flush

#Игнорирование отсутствующих моделей

При внедрении модели Eloquent в задачу модель автоматически сериализуется перед помещением в очередь и повторно извлекается из базы при обработке. Однако если модель была удалена, пока задача ожидала обработки, задача может завершиться с исключением ModelNotFoundException.

Для удобства можно автоматически удалять задания, у которых отсутствуют модели, установив свойство deleteWhenMissingModels вашего задания в true. Когда это свойство установлено в true, Laravel тихо отбросит задание, не выбрасывая исключение:

/**
 * Удалять задачу, если её модели больше не существуют.
 *
 * @var bool
 */
public $deleteWhenMissingModels = true;

#Очистка неудачных задач

Вы можете очистить записи в таблице 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
        });
    }
}

#Очистка задач из очередей

Если вы хотите удалить все задачи из очереди по умолчанию для соединения по умолчанию, используйте команду Artisan queue:clear:

php artisan queue:clear

Вы также можете указать аргумент connection и опцию queue для удаления задач из конкретного соединения и очереди:

php artisan queue:clear redis --queue=emails

#Мониторинг ваших очередей

Если в вашу очередь поступит резкий поток задач, она может перегрузиться, что приведёт к долгому времени ожидания выполнения задач. При необходимости Laravel может уведомлять вас, когда количество задач в очереди превысит заданный порог.

Для начала запланируйте выполнение команды queue:monitor каждую минуту. Команда принимает имена очередей для мониторинга и желаемый порог количества задач:

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

Простое планирование этой команды недостаточно для отправки уведомления о перегрузке очереди. Когда команда обнаружит очередь с количеством задач выше порога, будет сгенерировано событие Illuminate\Queue\Events\QueueBusy. Вы можете слушать это событие в EventServiceProvider вашего приложения, чтобы отправлять уведомления вам или вашей команде разработчиков:

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->connection,
                    $event->queue,
                    $event->size
                ));
    });
}

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

При тестировании кода, который отправляет задачи в очередь, вы можете захотеть, чтобы Laravel не выполнял сами задачи, так как код задач можно тестировать отдельно. Для тестирования самой задачи вы можете создать её экземпляр и вызвать метод handle напрямую.

Вы можете использовать метод fake фасада Queue, чтобы предотвратить фактическую постановку отложенных заданий в очередь. После вызова метода fake фасада Queue вы затем можете проверить, что приложение попыталось поставить задания в очередь:

<?php

namespace Tests\Feature;

use App\Jobs\AnotherJob;
use App\Jobs\FinalJob;
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, 2);

        // Проверить, что задача не была поставлена...
        Queue::assertNotPushed(AnotherJob::class);

        // Проверить, что в очередь была поставлена Closure...
        Queue::assertClosurePushed();

        // Проверить общее количество поставленных задач...
        Queue::assertCount(3);
    }
}

Вы можете передать замыкание в методы assertPushed или assertNotPushed, чтобы проверить, что была поставлена задача, удовлетворяющая определённому условию. Если хотя бы одна задача удовлетворяет условию, проверка будет успешной:

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

#Фейковать только часть задач

Если нужно фейковать только определённые задачи, позволяя остальным выполняться, передайте имена классов задач в метод fake:

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

    // Выполнить отправку заказа...

    // Проверить, что задача была поставлена дважды...
    Queue::assertPushed(ShipOrder::class, 2);
}

Вы можете фейковать все задачи, кроме указанных, используя метод except:

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

#Тестирование цепочек задач

Чтобы протестировать цепочки заданий, вам потребуется использовать возможности подмены фасада Bus. Метод фасада Bus assertChained можно использовать, чтобы проверить, была ли отправлена цепочка заданий. Метод 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);

#Тестирование цепочек с батчами

Если ваша цепочка задач содержит батч задач, вы можете проверить соответствие батча ожиданиям, вставив определение 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;
});

Вы можете использовать метод assertBatchCount для проверки количества отправленных батчей:

Bus::assertBatchCount(3);

Метод assertNothingBatched проверяет, что батчи не отправлялись:

Bus::assertNothingBatched();

#Тестирование взаимодействия задачи с батчем

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

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

$job->handle();

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

#События задач

Используя методы before и after фасада Queue facade, вы можете указать обратные вызовы, которые будут выполнены до или после обработки задачи из очереди. Эти обратные вызовы полезны для дополнительного логирования или сбора статистики для панели мониторинга. Обычно эти методы вызываются из метода boot service provider. Например, можно использовать 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 facade, вы можете указать обратные вызовы, которые выполняются перед попыткой воркера получить задачу из очереди. Например, можно зарегистрировать замыкание для отката всех транзакций, оставшихся открытыми после предыдущей неудачной задачи:

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

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