Как сделать какое-то действие в Laravel по окончанию набора очередей?

Организация действий после завершения всех задач в очереди Laravel

В Laravel для выполнения действий после завершения обработки всех задач в очереди можно использовать несколько подходов. Основные методы зависят от конкретной цели: выполнить действие после завершения одной конкретной очереди (job chain), после завершения всех задач в определенной очереди или после завершения всех задач в системе.

Основные подходы и их реализация

1. Использование Job Chains (Цепочки задач)

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

// Пример создания цепочки
Bus::chain([
    new ProcessFirstTask(),
    new ProcessSecondTask(),
    new FinalActionTask(), // Задача, выполняемая после всех
])->dispatch();

В методе handle() класса FinalActionTask вы можете разместить логику, которая должна выполниться после предыдущих задач.

Преимущества:

• Прямая и явная связь задач
• Гарантированное выполнение последовательности (если не происходит сбой)

Ограничения:

• Применимо только для заранее определенных групп задач
• Не подходит для динамически добавляемых задач

2. Использование событий и слушателей Laravel

Laravel генерирует события для различных этапов обработки задач в очереди. Вы можете создать слушателя (Listener) для события JobProcessed или WorkerStopping, но они срабатывают для каждой задачи или при остановке worker, что не равно "по окончанию всех задач".

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

// Создание собственного события
class AllQueueTasksCompleted
{
    public $queueName;

    public function __construct($queueName)
    {
        $this->queueName = $queueName;
    }
}

// Слушатель события
class HandleQueueCompletion
{
    public function handle(AllQueueTasksCompleted $event)
    {
        // Ваше действие здесь
        Log::info("Queue {$event->queueName} processing completed");
    }
}

3. Мониторинг состояния очереди через Redis или Database

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

// Проверка пустой очереди через Redis (для драйвера redis)
use Illuminate\Support\Facades\Redis;

public function checkQueueEmpty()
{
    $queueName = 'default';
    $length = Redis::connection('default')->llen('queues:' . $queueName);
    
    if ($length === 0) {
        // Выполнить ваше действие
        $this->performFinalAction();
    }
}

Этот метод можно запускать через планировщик задач (Scheduler) или через специальную команду Artisan.

4. Использование хуков в Worker процессах

В более сложных сценариях можно модифицировать Worker процесс Laravel. Worker имеет методы stop() и pause(), но они не предоставляют прямого хука на "полное завершение".

Один из подходов - создать кастомную команду Worker, которая выполняет ваше действие перед остановкой.

# Создание кастомной команды
php artisan make:command CustomQueueWorker

В команде можно переопределить логику обработки:

class CustomQueueWorker extends WorkerCommand
{
    protected function runWorker($connection, $queue)
    {
        // Стандартная обработка
        parent::runWorker($connection, $queue);
        
        // После завершения работы worker
        $this->onWorkerStopping();
    }
    
    protected function onWorkerStopping()
    {
        // Проверить, что очередь пуста и выполнить действие
        if ($this->queueIsEmpty()) {
            event(new AllQueueTasksCompleted($this->queue));
        }
    }
}

Рекомендации по выбору метода

• Для цепочек задач - используйте Job Chains
• Для отслеживания завершения конкретной очереди - комбинация событий + мониторинг состояния
• Для системных действий после всех очередей - кастомный Worker или планировщик

Практический пример: действие после импорта данных

Предположим, у вас есть очередь data-import, которая обрабатывает множество задач импорта. После завершения всех задач нужно отправлять отчет.

Реализация через событие:

// В последней задаче импорта или в мониторинге
if (Queue::size('data-import') == 0) {
    event(new ImportQueueCompleted());
    
    // Или напрямую
    Mail::send(new ImportReportMail());
}

Реализация через планировщик:

// В App\Console\Kernel
protected function schedule(Schedule $schedule)
{
    $schedule->call(function () {
        if (Queue::size('data-import') == 0) {
            // Выполнить действие
        }
    })->everyFiveMinutes();
}

Ключевые моменты для успешной реализации

1. Определите точный критерий "окончания" - пустая очередь? завершение цепочки? остановка worker?
2. Учитывайте надежность - некоторые задачи могут fail, что повлияет на ваше условие
3. Тестируйте в production-like environment - поведение очереди может отличаться
4. Добавьте logging для отслеживания работы механизма

Выбор конкретного метода зависит от бизнес-логики, масштаба системы и требований к надежности. В большинстве случаев комбинация Job Chains для управляемых процессов и Event-based мониторинга для динамических очередей дает оптимальный результат.