Вопрос по Laravel queue, как сделать чтоб очереди выполнялись друг за другом и не пересекались?

Question

[uniquerichboy]

У меня от 500-2000 очередей который запускаются в разное время на неделю вперёд и когда-то могут вызваться в одно время и превысить лимит запросов сервиса к которому обращаюсь, моя задача к которой я не могу найти решения кроме громких Kafka и RabbitMq, но и то не сильно уверен что мне нужны такие сервисы, искал готовые решения на гите, ничего не нашел, читал про Haystak, не то, Spatie тоже не то, тупо выкидывает задачи, но лимиты соблюдает, а мне надо чтоб они друг за дружкой шли, 1 задача выполняется, остальные ждут хоть и должны выполниться в данный момент, но найти решения не смог, надеюсь здесь найду решение!

Comments

[Дмитрий]

не совсем понятно. что мешает вам создать 1 очередь где будут задачи с обращениями к сервисам с лимитами и разбирать ее одним воркером? или где то хранить количество обращений и откладывать задачи если лимит превышен?

[uniquerichboy]

Дмитрий, вопрос в другом, в подходе, нужно ли изобретать велосипед и делать свой костыль когда по любому уже такое написали, может я просто не нашел, поэтому задаю вопрос

[Дмитрий]

uniquerichboy, https://laravel.com/docs/10.x/queues#rate-limiting

[nozzy]

uniquerichboy, нужно чтобы джобы выполнялись один за одним, по очереди?

[JhaoDa]

Документацию ты не читал, как обычно...

Answer 1

[pLavrenov]

В коробке есть WithoutOverlapping, добавь туда "ключ" группы необходимых задач и они будут выполняться по очереди.

Answer 2

[Илья Чубаров]

Поделюсь опытом, готового решения в Laravel из коробки нет.
Если надо остаться в рамках Laravel то есть несколько вариантов. Можно допилить функционал в самом Laravel, но код внутри не готов к такой кастомизации. (говорю заранее это будет жесткий костыль, он будет мешать нормальной работе очередей). Второй вариант, сделать еще одну базовую очередь.
Что я имею в виду?
Вариант №1
Мы берем простой кэш или redis(сразу) с его очередью(вроде это lists) , и добавляем туда наши jobs, каждый раз когда когда job выполняется, она смотрит должна ли именно она сейчас выполниться и если нет, то скрипт ее возвращает обратно в обычную очередь.
Соответственно запуск job регулируется другой очередью(то есть вы организуете добавление jobs в обычную очередь), из минусов, ваши jobs будут туда сюда гоняться скриптом и полезной работы не какой. Если например один уйдет в fail jobs - worker будет гонять туда сюда пока вы не выполните его и не восстановиться работа.

Вариант №2
Второй вариант сделать уже нормальную базовую очередь. К примеру, делаете таблицу в базе данных. Когда job в нормальном flow должен отправляться в очередь, вы отправляете его в эту таблицу. Дальше вы пишите своего демона(worker) который просматривает вашу таблицу на предмет выполнения.
Если что то готово выполняться, он отправляет его в обычную очередь. Добавляете в эту таблицу статусы и вперед.
Job добавилась в таблицу - "ready"
Job ушла - "в процессе"
Job выполнилась - "удалилась из таблицы или approved"
Возможно еще есть колонки для группировки(например принадлежность какой нибудь группе), теперь ваш демон может запускать их по очереди даже в рамках групп.
Соответственно все идут друг за другом, пока один "в процессе" остальные не берутся.

Comments

[Дмитрий]

Простите а зачем такие сложности? Выполнять определенный тип задач в один поток можно - просто отдав под это одну очередь и разбирать ее в один поток. Если нужно что то более сложное - то job middleware

[Илья Чубаров]

Я могу рассказать в чем у нас была проблема
Но как например бороться с тем что job может уйти из общего потока.
В самом простом случае нам надо стопорить все job если одни попала в fail jobs

[Дмитрий]

Илья Чубаров, ну было бы конечно интересно.

[Илья Чубаров]

Давай-те вместе попробуем решить три базовые проблемы
- job разбиты на десятки групп, в каждой группе надо соблюдать очередность, в том порядке в котором приходят события - ровно в том же их надо выполнять
- мы хотим иметь возможность увеличить производительность и запускать не один worker а больше в зависимости от кол-ва jobs
- если job попадает в fail table мы стопорим все остальные

[Дмитрий]

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

Ну первое банальное решение класть задачи от этого конкретного сервиса в конкретную очередь. И разбирать это одним воркером. И мы никогда не получим что у нас несколько воркеров полезут в один сервис и его задолбает. Оно будет их разбирать последовательно.
Второе решение, мы хотим зачем то запускать несколько воркеров на такие сервисы. Хотя вообщем то это не надо. Ну ок пишем свой миддлеваре - и свой rate limit который будет основаться скажем на основе количества обращений к сервису, ну который придется где вести - редис, мемкеш, база, файл. И откладывать его согласно каким то правилам.
Третье. Зачем нам fail table. мы можем отлавливать конкретную ситуацию когда мы ловим 429 - и шарашить delay().

У вас то какие проблемы были?

З.Ы. Если вы не нажимаете кнопку ответить - мне уведомление не сваливается.

[JhaoDa]

Илья Чубаров, есть изкоробочные механизмы для batch, chain, rate limit; есть куча настроек для определения поведения при ошибках; есть Horizon для скэйлинга. Вообще, из описания проблем складывается впечатление, что был изобретён какой-то велосипед.

[Илья Чубаров]

Дмитрий,

Ну первое банальное решение класть задачи от этого конкретного сервиса в конкретную очередь. И разбирать это одним воркером. И мы никогда не получим что у нас несколько воркеров полезут в один сервис и его задолбает. Оно будет их разбирать последовательно.

ну проблема не то что несколько worker лезут в один сервис
а в том что эти jobs должны выполняться строго один за другим

Например
пришло 3 запроса A,B,C
создано 3 jobs А,B,C
мы должны выполнить их как A,B,C
Если A выполниться, а B нет, то он уйдет в таблицу fail jobs, далее выполниться job C
Получается последовательность нарушена

[Дмитрий]

Илья Чубаров, откуда такие ограничения? я честно говоря не прочитал сего в вопросе автора. Ну да ладно - у нас есть Chain Jobs в которые мы можем запихать любую последовательность и прописать что будет происходить при падении какой то задачи из списка. И обрабатывать эти цепочки в один поток.

[Илья Чубаров]

JhaoDa, напишите подробнее как предложенное вами решит эти проблемы, возможно я не все объяснил

[JhaoDa]

Илья Чубаров, вы знаете об этих возможностях? Пробовали их? Они вам не подошли? Или ответ «нет» на все три вопроса?

[Илья Чубаров]

Дмитрий, мы сейчас мою проблему рассматриваем или автора
В моей ситуации были именно такие ограничения.
То есть последовательность запросов влияло на логику в целом
Я пока не понимаю как Batch Jobs поможет в этой ситации.
У нас есть асинхронные запросы, после каждого запроса надо создавать job в очередь.
Запросы должны выполняться строго в той последовательности в которой пришли.

[Илья Чубаров]

JhaoDa, У нас тут конструктивный диалог, все три да.
Я стараюсь тут подробно все расписывать и опираться на факты.
Если говорить "попробуйте N вы просто не знали", не очень конструктивно.
Напишите подробно как это сделать.

[Дмитрий]

Илья Чубаров, перепутал c job chains. уже исправил. Chains это цепочка из задач которые будут выполняться последовательно, и в случае отвала какой либо задачи - отвалится вся цепочка.

[Илья Чубаров]

Дмитрий, верно, но цепочка создается же сразу на месте
То есть при создании цепочки у нас на руках есть уже все jobs - а тут асинхронные запросы
Jobs добавляются постепенно через запросы.

[Дмитрий]

Илья Чубаров,
1. Ну насколько я помню вы можете расширять цепочку в момент обработки задачи.
2. Во вторых если у вас джобы создаются в изначальной джобе то вообще не понятно в чем проблемы - не смогли - не создали остальные, пошли на обработку такой ситуации

[Илья Чубаров]

Дмитрий, Видимо мы о чем то разном говорим просто
не понял что значит расширять цепочку в момент обработки и job не создаются внутри других job.
Надо попробовать другой пример подобрать
Допустим к вам на сервер приходят запросы одного характера. По каждому запросу вы создаете один job.
Все просто один request - один job.
Пришло 10 запросов - 10 jobs в очереди.
Их надо выполнять в той последовательности как они попали в сервер.
Как я могу тут использовать цепочку например?

[Дмитрий]

Илья Чубаров, не говоря о том что есть абсолютно твердолобый и тупой вариант аля

class SuperDuperJob implements ShouldQueue
{
    public function handle(){
          $jobA = new JobA($this->initalData);
          $jobA->handle();

          $jobB = new JobB($jobA->nextData);
          $jobB->handle();
           
          ...............
    }
}

Хрен тут чего не последовательно выполнится.

[Дмитрий]

Илья Чубаров, ну в какой последовательности они попадут в очередь в такой они и выполнятся.

[Илья Чубаров]

Дмитрий,
ну смотрите так мы сделать не можем, потому что job ваши выполняются вместе и они имеют связь.
Job которые добавляются в очередь по каждому запросу такую связь не имеют.
В вашем коде можно написать так, только в случае если мы имеем заранее известное кол-во, но мы его не имеем.
Запрос -> Job -> очередь

Потом тут в теле два job в очереди не находятся и могут отказать, соотвественно SuperDuperJob будет находиться не в консистентном состоянии, JobA выполнилась(отправила например запрос на списание), а JobB нет(не начислила деньги)

Но все это имеем смысл когда у вас есть заранее подготовленный список связанных jobs
Как в документации, например обработка audio
- Форматирование
- Сжатие
- Отправка
Вот это ложиться на chain

А у нас другой flow
Пришло аудио -> цепочка jobs которые выше A
Пришло еще аудио -> цепочка jobs которые выше B
Пришло еще аудио -> цепочка jobs которые выше C

Это все надо выполнить строго последовательно A,B,C

[Илья Чубаров]

Дмитрий,

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

нет не выполниться, можете посмотреть в исходниках демона worker
он берет первый доступный из очереди. В этой логике если например jobs попал в fails, worker возьмет следующей - последовательность нарушена.
Если мы добавим workers - они будут брать конкуретно все доступные jobs.

[Дмитрий]

Илья Чубаров,

нет не выполниться, можете посмотреть в исходниках демона worker

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

[Илья Чубаров]

Дмитрий, В том числе если один job помечен как "reserved" мы не должны брать из очереди следующие job

[Дмитрий]

Илья Чубаров,

А у нас другой flow
Пришло аудио -> цепочка jobs которые выше A
Пришло еще аудио -> цепочка jobs которые выше B
Пришло еще аудио -> цепочка jobs которые выше C

Это все надо выполнить строго последовательно A,B,C

1. Ну используйте очередь с приоритетами. Какой нибудь бинсталк имеет это из коробки.
2. У вас эта логика и так может развалиться. У вас прилетит аудио, в момент когда у вас уже идет обработка B. И все.

З.Ы. А давайте вы нормально опишете свой флоу и что вы вообще делаете. Потому что играть в ситуацию когда на каждый ответ чуть менять требования - можно долго, да и ни один программист этого не любит.

[Дмитрий]

Илья Чубаров,

В том числе если один job помечен как "reserved" мы не должны брать из очереди следующие job

пишете rate limit который проверяет есть ли в наличии хоть что то в обработке, и откладывает все задачи.

[Илья Чубаров]

Вобщем у меня не получается объяснить мою проблему, наверное ответ был не в тему вопроса.
Решить проблему вопроса можно всеми способами что вы предложили.Я пытался объяснить именно мою проблему, видимо было не в тему.
Могу удалить свой ответ

[Дмитрий]

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

Answer 3

[vrusua]

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

См. Laravel Workflow - оркестратор задач на базе Laravel jobs и UI тузла Waterline к нему для мониторинга выполнения:
https://github.com/laravel-workflow/laravel-workflow
https://github.com/laravel-workflow/waterline

Также еще альтернативы:
https://github.com/zerodahero/laravel-workflow
https://github.com/sebdesign/laravel-state-machine
https://github.com/asantibanez/laravel-eloquent-st...