Ответ от GPT, не проверял. :)

moko 28.09 20:13

В обычной очереди просто поставить задаче "не выполнять до такого-то времени" нельзя. expiration, который есть у publish, это TTL сообщения, а не задержка доставки. Если у очереди есть consumer, он получит сообщение сразу.

Но текущего API Parser3 уже достаточно, чтобы сделать отложенную очередь стандартным способом через TTL + dead letter exchange. Например, отдельная очередь tasks-delay без consumer'ов, у нее задаются x-dead-letter-exchange и x-dead-letter-routing-key, а сообщение публикуется туда с:

$.expiration(60000)

Через минуту оно протухнет и RabbitMQ переложит его в обычную рабочую очередь.

Есть нюанс: если каждой задаче задавать свой expiration и складывать их все в одну такую очередь, это плохо работает для произвольных задержек. RabbitMQ удаляет/перекладывает протухшие сообщения, когда они доходят до головы очереди. Поэтому задача на 10 секунд может оказаться за задачей на час и фактически ждать час.

Обычно без дополнительных плагинов делают несколько delay-очередей с фиксированными задержками, например 10 секунд, минута, 10 минут и т. п.

Если нужен именно свой delay для каждого сообщения, то это умеет rabbitmq_delayed_message_exchange: создается exchange типа x-delayed-message, обычная рабочая очередь привязывается к нему, а при публикации передается:

$.headers[
    $.x-delay(60000)
]

Текущий класс amqp со стороны Parser3 это уже позволяет: у declare есть произвольные arguments и type, у publish есть произвольные headers.

Например:

^o.declare[
    $.exchange[tasks-delay]
    $.type[x-delayed-message]
    $.durable(true)
    $.arguments[
        $.x-delayed-type[direct]
    ]
]

^o.declare[
    $.queue[tasks]
    $.durable(true)
]

^o.bind[
    $.exchange[tasks-delay]
    $.queue[tasks]
    $.routing_key[tasks]
]

^o.publish[$message;
    $.exchange[tasks-delay]
    $.routing_key[tasks]
    $.headers[
        $.x-delay(60000)
    ]
]

Тогда сама tasks остается обычной очередью, а задержка происходит до попадания сообщения в нее.

Но rabbitmq_delayed_message_exchange требует серверный plugin. Кроме того, сейчас upstream этого plugin уже закрыт и для новых RabbitMQ 4.3 он не подходит, так что для новой системы я бы скорее смотрел на TTL/DLX с фиксированными delay-очередями либо отдельный scheduler.

---