Как приостанавливать и контролировать очереди в Laravel пошагово

Информатек Диджитал » Ресурсы » Как приостанавливать и контролировать очереди в Laravel пошагово

Работа с очередями в Laravel практически обязательна в любом более-менее серьезном проекте: отправка электронных писем, обработка CSV-файлов, генерация PDF-файлов, изменение размера изображений, вызов внешних API… Если все это выполняется в основном HTTP-запросе, веб-сайт замедляется, появляются ошибки 500, и пользователь в итоге отчаянно смотрит на пустой экран.

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

Основные понятия: соединения, очереди и драйверы

В Laravel, прежде чем что-либо приостанавливать, вам нужно понять разницу между соединения очередей и сами очереди. En config/queue.php у вас есть массив connections где вы определяете каждый бэкэнд: база данных, redis, beanstalkd, sqs, синхронизация, null, отказоустойчивость, rabbitmq (если вы его добавите) и т.д.

Каждое соединение может иметь несколько логические очереди (такие имена, как default, emails, high, low, importsАтрибут queue Каждое соединение указывает, в какую очередь по умолчанию будут помещаться задания, если не указано иное. onQueue()Таким образом вы можете делать следующее:

• Быстрая очередь для электронных писем или коротких транзакций.
• Еще один низкий приоритет для длительных задач, таких как обработка видео или больших отчетов.
• Очереди, сегментированные по типу задачи: imports, notificaciones, reportes, и т.д.

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

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

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

Драйверы очередей и предварительные условия

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

База данных водителей

Водитель база данных Он сохраняет задания в таблице SQL (обычно jobsЕго легко настроить, и он идеально подходит для небольших проектов или сред разработки, где вы не хотите возиться с Redis или SQS.

Основные шаги:

• Сгенерировать миграцию: php artisan queue:table (или в более новых версиях он идет как create_jobs_table по умолчанию).
• Запуск миграций: php artisan migrate.
• Конфигурировать QUEUE_CONNECTION=база данных в .env и подтвердите в config/queue.php что 'default' => env('QUEUE_CONNECTION','database').

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

Драйвер Redis

Драйвер Redis — это стандартный рабочий инструмент в продакшене. Он быстрый, поддерживает расширенные функции (блокировка, ограничение скорости, атомарные блокировки, Horizon и т. д.) и очень хорошо масштабируется.

Важные моменты:

• Вам необходимо настроить соединение Redis в config/database.php.
• опции serializer y compression от Редиса Они не совместимы с водителем очереди.
• Если вы используете Redis ClusterНазвания очередей должны иметь хэштег, например {default}так, чтобы все ключи в этой очереди попадали в один и тот же слот.
• Выбор block_for Это позволяет работнику блокироваться в ожидании заданий на несколько секунд, вместо того чтобы непрерывно опрашивать: экономит процессор и сокращает количество запросов к серверу Redis.

Например:

'redis' => ,

Если вы поставите блок_для = 0Работник остается заблокированным на неопределенный срок, пока не появится доступная работа. Это следует иметь в виду, если вы хотите, чтобы он быстро реагировал на такие сигналы, как SIGTERM или приостановить команды.

Другие драйверы: SQS, Beanstalkd, null, отказоустойчивость…

Помимо баз данных и Redis, Laravel поддерживает и другие системы :

• Амазонка SQSУправляемые очереди на AWS с поддержкой FIFOГруппы сообщений, дедупликация и расширенные настройки видимости. Требуется SDK. aws/aws-sdk-php.
• бобовый стебельОчень быстро и просто, используя пакет pda/pheanstalk.
• синхронизировать: выполняет задачи inmediatamente в том же процессе. Полезно при разработке или тестировании, но в производстве это практически равносильно полному отказу от очередей.
• нуль: Отменяет задания сразу после их «отправки». Используется для полного отключения системы очередей.
• отказоустойчивыйЭто позволяет настроить несколько подключений, и если основное соединение выходит из строя, Laravel пытается передать задачу следующему в списке. Идеально подходит для сред, где высокая доступность имеет решающее значение.

На уровне приостановки и возобновления работы обработчиков логика практически одинакова независимо от драйвера, поскольку управление осуществляется командами Artisan и классом Worker , а не бэкэндом очереди.

Создание заданий и сериализация моделей

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

Чтобы создать его:

php artisan make:job ProcessPodcast

Сгенерированный класс инвентарь ShouldQueue и обычно использует черты Dispatchable, InteractsWithQueue, Queueable y SerializesModelsСердце в методе handle()что и делает работник, когда получает задание.

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

Типичный пример (на псевдокоде, чтобы не повторять дословно):

• Строитель получает Подкаст $podcast.
• Черта SerializesModels Он отвечает за сохранение идентификатора.
• En handle(AudioProcessor $processor) Вы делаете тяжелую работу (например, обрабатываете аудио).

Если вы не хотите, чтобы связи в модели были сериализованы (чтобы полезная нагрузка не была большой), вы можете использовать $model->withoutRelations() или в PHP 8 атрибут # на объекте недвижимости, продвигаемом застройщиком.

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

Еще одно преимущество в том, что вы можете внедрять службы непосредственно в метод обработки (как если бы это был контроллер), поскольку Laravel также использует контейнер сервисов в Jobs. Например, вы можете запросить AudioProcessorHTTP-клиенты, репозитории и т. д., а фреймворк заботится об их разрешении.

Если вам нужен сверхточный контроль над тем, что называется handleвы можете использовать Container::bindMethod() у поставщика услуг и самостоятельно определить логику внедрения, но в большинстве случаев нет необходимости так сильно усложнять ситуацию.

Промежуточное программное обеспечение для заданий

Так же, как маршруты имеют промежуточное программное обеспечение, задания также могут иметь конкретное промежуточное программное обеспечение которые выполняются до и после задания. Это значительно сокращает количество повторяющегося кода в методах. handle() и это особенно полезно для:

• Ограничение скорости с Redis.
• избежать перекрывается работ, затрагивающих один и тот же ресурс.
• Регулярные исключения для нестабильных API.
• Пропуск заданий при определенных условиях (Skip промежуточное программное обеспечение).

Промежуточное программное обеспечение обычно размещается в app/Jobs/Middleware и возвращаются из метода middleware() Из книги Иова. Концептуальный пример:

public function middleware(): array { return ;

Специальные типы заданий: уникальные, зашифрованные и заблокированные

Laravel позволяет вам отмечать определенные задания как уникальный чтобы гарантировать, что два одинаковых экземпляра никогда не будут поставлены в очередь одновременно. Это достигается реализацией интерфейса ShouldBeUnique o ShouldBeUniqueUntilProcessingи поддерживает определение ключа uniqueId() и время uniqueFor.

Это идеально подходит для таких вещей, как:

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

В основе Laravel лежит атомный замок кэша (Redis, Memcached, база данных и т.д.). Вы также можете выбрать драйвер кэша для этой блокировки с помощью uniqueVia().

Кроме того, есть интерфейс ShouldBeEncrypted, с помощью которого вы указываете, что работа выполнена полностью Введите свой код перед тем, как встать в очередьОчень полезно при передаче конфиденциальных данных, которые не нужно передавать в виде обычного текста через Redis, SQS или базу данных.

Отправка, задержка и выполнение заданий в различных режимах

Как только ваша работа определена, вы запускаете ее с помощью Job::dispatch() из контроллеров, прослушивателей событий, команд Artisan и даже замыканий. Параметры, которые вы передаёте dispatch Они идут прямо к создателю Иова.

Интересные варианты:

• dispatchIf() y dispatchUnless() для размещения груза в одной линии.
• delay() так что задание не может быть обработано до определенной даты/времени.
• dispatchAfterResponse() o afterResponse() en dispatch() Замыкания, которые откладывают выполнение до тех пор, пока не будет отправлен HTTP-ответ, но в пределах того же PHP-процесса.
• dispatchSync() для синхронного выполнения задания (без прохождения очереди), но с повторным использованием того же класса.
• Специальные соединения, такие как deferred o background для выполнения после ответа, в том же процессе или в новом PHP-процессе.

Наконец, вы можете указать соединение и очередь с помощью цепочечных методов. onConnection() y onQueue(), либо при отправке, либо в конструкторе заданий.

Тонкое управление: повторные попытки, отсрочка, тайм-ауты и обработка ошибок

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

Возможные варианты:

• --tries en queue:work установить максимальное количество попыток для всех заданий этого работника (если задание не определяет свое собственное) $tries o tries()).
• $tries или метод tries() в классе Job для управления отдельными заданиями.
• retryUntil() сказать: «Пытайтесь снова столько раз, сколько захотите, до этого момента, после этого — неудача». Если это совпадает с tries, команда retryUntil.
• $maxExceptions для отключения, когда проблема вызвана повторяющимися необработанными исключениями.
• $timeout o --timeout для указания того, сколько секунд может выполняться задание, прежде чем рабочий процесс завершит дочерний процесс.
• $failOnTimeout отметить задание как проваленное в случае истечения времени ожидания.
• $backoff метод backoff() чтобы определить, как долго ждать между повторными попытками, даже при экспоненциальных значениях.

Если задание выдает неперехваченное исключение, Laravel... выпускает обратно в хвост Пока заданное количество попыток не будет исчерпано. Вы можете управлять этим вручную с помощью release($segundos) o fail($exception) когда вы хотите намеренно отметить задание как проваленное.

Неудачные задания, таблица failed_jobs и обрезка

Когда Иов исчерпывает свои попытки, его начинают считать проваленная работа и Laravel вставляет его в таблицу failed_jobs (или в DynamoDB, если настроено соответствующим образом). В этой таблице хранятся UUID, соединение, очередь, полезная нагрузка задания и выданное исключение.

Полезные команды:

• php artisan queue:failed перечислить проваленные.
• php artisan queue:retry <id|all> чтобы попробовать их снова.
• php artisan queue:forget <id> чтобы удалить один.
• php artisan queue:flush удалить все (с опцией удаления часов в более новых версиях).
• php artisan queue:prune-failed --hours=48 для автоматического удаления старых записей.

В разделе «Задания» вы можете определить метод failed(Throwable $exception) Для дополнительной очистки: уведомите пользователя, отмените частичные действия, запишите специализированные журналы и т. д. Обратите внимание, что создается новый экземпляр задания для вызова failed()Поэтому не полагайтесь на изменения свойств, внесенные в handle().

Пакетирование заданий: пакеты заданий и их объединение со строками

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

Основной поток:

• создать таблицу job_batches c php artisan queue:batches-table && php artisan migrate.
• Отметьте задания, которые будут частью партий с характеристикой Batchable.
• Использовать Bus::batch() для отправки набора заданий с обратными вызовами then, catch y finally.
• Доступ к пакету из каждого задания с помощью $this->batch() и проверьте, отменено ли оно.
• Возможность добавления дополнительных заданий в текущий пакет с помощью $this->batch()->add().

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

Выполнение задач, приоритеты и развертывание

Вся магия очередей происходит потому, что у вас есть один или несколько работников выполнение команды queue:work (o queue:listen(хотя последний вариант менее эффективен). Эти процессы — долго работающие демоны, прослушивающие свои очереди.

Ключевые параметры queue:work:

• Связь: php artisan queue:work redis, queue:work database, и т.д.
• Приводы: --queue=high,default,low расставить приоритеты.
• -один раз: обрабатывает одно задание и завершает работу.
• –макс-рабочих мест: обрабатывает X заданий и завершает работу (для периодического перезапуска рабочих процессов и освобождения памяти).
• –макс-время: обрабатывается в течение N секунд, а затем завершается.
• -спать: секунды ожидания при отсутствии доступных заданий.
• –остановка-когда-пусто: очищает очередь и корректно закрывается (идеально подходит для сред Docker или разовых заданий).
• –тайм-аут: секунды тайм-аута на задание, как мы уже упоминали ранее.
• -force: для обработки заданий, даже если приложение находится в режиме обслуживания.

Поскольку эти процессы являются длительными, в производственной среде почти всегда используется монитор процессов , такой как Supervisor (в Linux), для запуска X-воркеров, их перезапуска в случае сбоя и управления логами. Типичный блок конфигурации Supervisor для Laravel определяет:

• команду с php artisan queue:work ... конкретный.
• Numprocs для указания количества рабочих процессов, которые следует запустить параллельно.
• Пути журналов, пользователь, автоматический перезапуск и т. д.

Приостановка и возобновление очередей в Laravel

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

Пауза: очередь: пауза и сигналы прерывания

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

• php artisan queue:pause: сообщает системе о необходимости прекратить прием новых заданий, но позволить тем, которые уже выполняются, завершиться.
• php artisan queue:continue: возобновляет обработку новых заданий с того места, на котором она была прервана.

Эти команды работают, отправляя сигналы рабочим процессам через внутренний механизм кэширования Laravel. Именно поэтому важно правильно настроить драйвер кэширования (memcached, redis, database, file и т. д.), даже если ваша фактическая очередь кэширования — это что-то другое.

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

Опрос сбоев и их влияния на производительность

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

Queue::withoutInterruptionPolling();

Или даже настроить статические свойства Illuminate\Queue\Worker чтобы отключить определенные касания ($restartable, $pausableОднако, если вы это сделаете и затем попытаетесь использовать queue:pause o queue:restartрабочие Они не узнают.Это дизайнерское решение, которое следует тщательно обдумать.

Приостановка вручную с помощью Supervisor или Docker

Помимо собственных команд Laravel , в производственных средах процессы часто «приостанавливаются» косвенно:

• Остановка программы в Supervisor (supervisorctl stop laravel-worker:*), что убивает или останавливает рабочих.
• Используя –остановка-когда-пусто в развертываниях, чтобы дать работникам возможность завершить то, что находится в очереди, а затем завершить работу.
• В Docker, позволяя контейнеру обрабатывать --stop-when-empty а затем он выключается.

Разница с queue:pause Полностью остановив процессы, вы прекратите обработку заданий до тех пор, пока рабочие процессы не возобновятся. внутренняя пауза Вы можете сохранять процессы в памяти, ожидая команду continueЭто полезно для быстрых откатов развертываний или коротких периодов обслуживания.

Транзакции базы данных и after_commit

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

Чтобы избежать этого танца, у вас есть два варианта:

• Активировать 'after_commit' => true в конфигурации подключения к очереди. Таким образом, любое задание, отправленное в рамках транзакции, откладывается внутри системы до подтверждения.
• Используйте методы, учитывающие каждую задачу afterCommit() o beforeCommit() при диспетчеризации, чтобы принудительно задать поведение только в определенных заданиях.

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

Мониторинг, очистка очереди и тестирование

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

Полезные инструменты и команды:

• queue:clear для очистки определенной очереди (SQS, Redis, база данных).
• queue:monitor для отслеживания количества заданий в определенных очередях и запуска события QueueBusy если они превышают пороговое значение.
• Мероприятия по трудоустройству в качестве Queue::before, Queue::after, Queue::looping y Queue::failing для подключения пользовательской логики (журналов, метрик, откатов и т. д.).
• Подделки в тестировании использование Queue::fake() y Bus::fake() для проверки того, что задания, строки или пакеты были отправлены без их фактического выполнения.

В тестах можно указывать такие вещи, как «это задание было поставлено в очередь X раз» , «цепочка с этими заданиями была отправлена ​​в таком порядке» или «никакое конкретное задание не было поставлено в очередь», что обеспечивает высокую степень безопасности в сложных потоках.

С этой целой экосистемой — драйверами, такими как Redis или SQS, хорошо спроектированными заданиями, промежуточным программным обеспечением для тонкой настройки, пакетами, цепочками, конфигурацией повторных попыток, управляемыми Supervisor рабочими процессами и такими командами, как queue:pause y queue:continue— Laravel позволяет вам освоить фоновую обработку с точностью до миллиметра, от типичной отправки электронной почты до распределенных очередей с отказоустойчивостью, гарантируя, что ваше приложение будет продолжать быстро реагировать на уровне HTTP, пока внизу выполняется тяжелая работа, не блокируя пользователя.