Как сохранить очередность выполнения заданий в очереди?

Question

[kategg]

Есть проблема с порядком обработки заданий в очереди. Допустим, что я, с интервалом 2-3 секунды, занесла в очередь 3 задания.
Первое задание выполняется 30 секунд и записывает в лог цифру 1

public function handle()
    {
        sleep(30);
        file_put_contents(storage_path('/logs/journal.log'), 1, FILE_APPEND);
    }

второе - выполняется 20 секунд и записывает цифру 2

public function handle()
    {
        sleep(10);
        file_put_contents(storage_path('/logs/journal.log'), 2, FILE_APPEND);
    }

третье - выполняется 2 секунды и записывает цифру 3

public function handle()
    {
        sleep(2);
        file_put_contents(storage_path('/logs/journal.log'), 3, FILE_APPEND);
    }

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

Comments

[Lynn «Кофеман»]

Не изобретать велосипед, а, например, взять БД в которую писать время начала и окончания задачи.

А потом сортировать полученные данные как угодно.

[Дмитрий]

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

[kategg]

Дмитрий, вкратце опишу суть задачи. Пользователи Авито отправляют сообщения в мессенджере, а к моему приложению приходят вебхуки (1 вебхук = 1 сообщение). Нужно выводить эти сообщения в crm, сохраняя очередность. Но, если это новый чат, то перед выводом сообщений необходимо сделать несколько запросов к апи crm для создания сущностей, а уже затем выводить сообщения. Также, одни сообщения могут обрабатываться дольше других, что может повлечь за собой нарушение порядка вывода сообщений

[Дмитрий]

kategg, ну а приходящие сообщения вы прям из очереди отправляете в crm? или они у вас кладутся в бд, а потом crm их берет оттуда?

[kategg]

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

[Дмитрий]

kategg, ага. теперь более менее понятно.
Я бы предложил бы вам посмотреть в сторону ShouldBeUniqueUntilProcessing. Этот трейт будет вам гарантировать что в момент у вас будет обрабатываться 1 задача с заданным ключом, остальные будут отложены. Если вы в ключ уникальности положите что то-то что будет идентифицировать пользователя то получится, что все остальные сообщения от этого пользователя будут отложены пока не отработается первое. При этом сообщения остальных пользователей будут отрабатываться

[kategg]

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

[Дмитрий]

kategg, значит она должна вернуться в очередь. однако это вроде как будет зависить от attempts и прочего.

Answer 1

[JhaoDa]

Надо начать читать документацию и не изобретать велосипедов с БД и прочим.

Comments

[Lynn «Кофеман»]

Это если мы хотим что бы джобы запускались последовательно.

Хотя, конечно, автор не уточнял этот момент.

[JhaoDa]

Lynn «Кофеман»,

а надо получить 123

Да всё он уточнил.

[Lynn «Кофеман»]

JhaoDa, ну лично я это понял как «хочется ответ в правильном порядке», а не «хочется что бы задачи выполнялись последовательно»

[kategg]

Про цепочки я знаю, но это не решает проблему. Я написала, что задания попадают в очередь не сразу пачками, а с временным интервалом

[JhaoDa]

kategg, ну так или очерёдность работы, или скорость работы. Для первого цепочки, для второго пачки. Выбирай

Answer 2

[Роман Олегович]

Ну раз chain не устраивает, то есть в доке и другой раздел: "Предотвращение дублирования работы"
#preventing-job-overlaps

Надо сделать примерно такой trait:

<?php
namespace App\Jobs;

use Illuminate\Queue\Middleware\WithoutOverlapping;

trait WithoutOverlappingFileWrite
{

    public function middleware(): array
    {
        return [
            (new WithoutOverlapping('file_write'))
                ->shared() // https://laravel.com/docs/11.x/queues#sharing-lock-keys
        ];
    }

    public function retryUntil(): \Illuminate\Support\Carbon
    {
        return now()->addSeconds(60);
    }

}

и use WithoutOverlappingFileWrite; во всех Job'ах