Работа с очередями

Очереди нужны, когда работу не стоит выполнять в том же HTTP-запросе или artisan-команде, в которых она возникла. Типичные случаи — долгий пересчёт, пакетная миграция данных из мастер-сервисов, тяжёлое обращение к внешнему API, индексация пачки сущностей. При асинхронном соединении задача передаётся в backend очереди, а отдельный воркер подхватывает её позже. Соединение sync выполняет работу сразу в вызывающем процессе.

В Goravel-сервисах Ensi очереди строятся на штатном Queue-фасаде фреймворка, а обвязка запуска воркеров (встроенный runner рядом с web и команда queue:work) лежит в app/adapters/queue. Сами джобы принадлежат модулям: app/modules/{module}/jobs. Список всех известных джоб собирается в bootstrap/jobs.go, настройки соединений и слушаемых очередей — в config/queue.go.

Очередь служит для фоновой работы внутри сервиса; интеграционные события между сервисами передаются через Kafka. Ниже описаны Goravel 1.18 и очередь Go-шаблона. В существующем сервисе проверьте наличие config/queue.go, provider, регистрации jobs, команды и runner. Все команды выполняют из корня сервиса через elc.

Как выглядит джоба со стороны фичи

Джоба — обычная структура с двумя обязательными методами. Signature() возвращает уникальное строковое имя задачи. Handle(args ...any) error содержит логику. Имя нужно воркеру, чтобы понять, какую структуру создавать и запускать: именно по signature задача лежит в Redis или в таблице очереди. Его лучше выбирать стабильным и говорящим, в духе catalog:reindex_offers или common:entities:migrate, и после появления в проде без крайней нужды не переименовывать — иначе уже отправленные, но ещё не выполненные задачи перестанут находиться.

Минимальный шаблон такой:

type ReindexOffersJob struct{}

func (j *ReindexOffersJob) Signature() string {
    return "catalog:reindex_offers"
}

func (j *ReindexOffersJob) Handle(args ...any) error {
    // аргументы приходят в том же порядке, в котором их передали при Dispatch
    return nil
}

В реальных модулях джоба часто не держит всю бизнес-логику внутри Handle, а делегирует её action’у — так же, как HTTP-хэндлер не ходит в ORM напрямую. Например, джоба миграции сущностей создаёт (или получает) action и вызывает Execute. В Handle остаётся только обвязка: разобрать аргументы, вызвать действие, вернуть ошибку наружу.

Джобу обязательно зарегистрировать. Иначе воркер при разборе задачи из бэкенда не узнает signature и не сможет её выполнить. Удобный приём: в пакете jobs модуля собрать функцию All() []queue.Job, а в bootstrap/jobs.go склеить списки всех модулей в общее Jobs(). Новые джобы добавляют и в All() своего модуля, и тем самым — в общий перечень при старте приложения.

В шаблоне сервиса обычно оставляют учебные джобы в app/modules/examples/jobs (короткий и длинный sleep) — ими удобно проверить, что воркер жив, ещё до написания своей фичи.

Как отправить задачу

Из любого места, где уже поднято приложение — HTTP-хэндлер, artisan-команда, другая джоба, Kafka-handler — задачу отправляют через фасад:

err := facades.Queue().
    Job(&jobs.ReindexOffersJob{}, []queue.Arg{
        {Type: "int", Value: offerID},
        {Type: "string", Value: "full"},
    }).
    OnQueue("default").
    Dispatch()

Job принимает экземпляр джобы и слайс аргументов. OnQueue выбирает именованную очередь — логическую «ленту», из которой воркер будет забирать задачи. В типовом конфиге воркеры по умолчанию слушают очередь default. Если тяжёлые задачи не должны тормозить короткие, заводят отдельное имя (например heavy), кладут туда соответствующие джобы через OnQueue("heavy") и следят, чтобы воркер эту ленту реально слушал: либо добавляют имя в queue.listen, либо передают его аргументом в queue:work.

Аргументы описывают слайсом queue.Arg: у каждого элемента есть тип (string, int и другие значения, которые ожидает фреймворк) и значение. По типу значение сериализуется в бэкенд очереди и потом восстанавливается при вызове Handle. Порядок элементов при отправке совпадает с порядком в Handle(args ...any), поэтому в Handle обычно делают аккуратный type assert и не полагаются на «магический» разбор.

Иногда задачу нужно выполнить не сразу. Перед Dispatch() можно вызвать .Delay(time.Now().Add(5 * time.Minute)) — воркер возьмёт её только после указанного момента (для соединений, которые поддерживают отложенные задачи). Если в этом же процессе нужно дождаться результата и не класть ничего во внешнюю очередь, используют DispatchSync(): Handle выполнится синхронно в текущем потоке. Это полезно в тестах или в редких командах, где асинхронность мешает, но код отправки хочется оставить тем же.

Цепочку «сначала A, потом B» можно собрать через facades.Queue().Chain(...). Для большинства фич достаточно одной джобы с понятными аргументами.

Соединения: sync, redis, database

То, куда физически уходит задача после Dispatch(), задаёт QUEUE_CONNECTION (ключ queue.default в конфиге).

Если соединение sync (так часто в локальном .env.example), отдельный воркер не нужен: Dispatch() сразу вызывает Handle в том же процессе. Это удобно для отладки логики джобы «здесь и сейчас», но поведение не как в бою. Нет настоящего отложенного выполнения через бэкенд, нет отдельного воркера, падение джобы выглядит как ошибка вызывающего кода. Перед сдачей фичи, которая зависит от асинхронности, имеет смысл хотя бы раз прогнать сценарий на redis или database.

Для redis или database задача действительно уходит «в сторону» и должна быть забрана воркером. В конфиге для соединений задают драйвер, имя очереди по умолчанию и concurrent — сколько задач из одной очереди можно обрабатывать параллельно на одном воркере, если не переопределили это флагом --threads.

После исчерпания повторов асинхронный worker пытается записать задачу в failed jobs согласно queue.failed (обычно таблица failed_jobs). Для этого нужны применённая миграция и доступное соединение БД; ошибку записи worker логирует. Сохранённые задачи можно просмотреть и переотправить командами queue:failed и queue:retry. Ошибка синхронного Dispatch возвращается вызывающему коду и не проходит через этот worker.

Кто выполняет джобы

Для асинхронного соединения после Dispatch задачу выполняет worker. При sync обработчик вызывается сразу, без worker.

Первый способ — вместе с web-приложением. Если runner зарегистрирован в bootstrap/app.go, QUEUE_WORKER_ENABLED=true и соединение не sync, при старте HTTP-процесса запускается app:queue на очередях из queue.listen. Локально это удобно: подняли сервис — и короткие джобы уже кто-то обрабатывает. На stage/prod встроенный воркер обычно выключают и гоняют queue:work отдельным процессом или деплоем, чтобы масштаб воркеров не был жёстко привязан к числу HTTP-подов.

Второй способ — команда queue:work. Её запускают рядом с приложением или вместо встроенного runner:

elc ./artisan queue:work
elc ./artisan queue:work default
elc ./artisan queue:work default heavy --threads=4

Отдельная команда queue:work не зависит от QUEUE_WORKER_ENABLED. Без аргументов слушаются очереди из queue.listen. Имена после команды — конкретные очереди. Флаг --threads задаёт, сколько задач из одной очереди можно обрабатывать параллельно; если передать 0 или не указать флаг, берётся значение concurrent из конфига соединения (часто это 1).

В stdout для обрабатываемых задач выводятся статусы: RUNNING, затем DONE или FAIL и длительность.

Остановка — Ctrl+C локально или SIGTERM в Kubernetes. Worker прекращает получение новой работы и ждёт выполняемые задачи. terminationGracePeriodSeconds должен учитывать длительность обработки вместе с повторами и задержками. Принудительное завершение процесса не гарантирует запись в failed jobs; восстановление зарезервированной задачи зависит от драйвера. Обработчик должен допускать повторный запуск.

Ошибки и повторы

В асинхронном worker ошибка Handle сначала участвует в выборе повтора. Если джоба реализует queue.JobWithShouldRetry, вызывается метод ShouldRetry(err error, attempt int) (bool, time.Duration). Он задаёт необходимость следующей попытки и паузу; номер первой неудачной попытки — 1. Без этого метода worker использует своё число попыток Tries. Команда Ensi queue:work предоставляет флаг --threads, но не публикует отдельный --tries.

FAIL и запись в failed jobs относятся к окончательному отказу после прекращения повторов. При собственной политике задайте ограничение числа попыток и учитывайте тип ошибки. Ошибка при записи failed jobs не восстанавливает уже выполненные внешние эффекты.

Как и для Kafka-хэндлеров, Handle лучше сразу писать идемпотентным. Воркер может взять задачу ещё раз после сбоя, рестарта пода или ручного queue:retry. Повтор не должен плодить дубли в БД, повторно слать одно и то же внешнее «разрушающее» действие без защиты или оставлять данные в полуприменённом состоянии без возможности дожать их следующим запуском.

Аргументы джобы тоже стоит проектировать с учётом повторов: передавайте устойчивые идентификаторы сущностей, а не «случайный снимок», который через пять минут уже не имеет смысла, если только это не осознанная часть контракта задачи.

При переносе с Laravel сохраните контракт job и план обработки накопленных задач отдельно: одинаковое имя очереди не делает PHP- и Go-payload совместимыми. Порядок подключения метрик для worker описан в метриках.