Is it possible to configure consumer for multiple queues with priority?

[dimanzver]

Is your feature request related to a problem? Please describe.

Is it possible to set up so that the consumer receives messages first from the first queue, then from the second, then from the third? That is, so that messages in different queues have different priorities. As soon as the tasks in the first queue run out, we consume messages from second queue, when the tasks in it run out, we consume messages from third queue. If at this time a message suddenly appears in the first queue, then we process it. Yes, I know that there are message priorities, but we have plans to use queues with the "quorum" type that do not support message priorities.

Describe the solution you'd like

I tried this, but not works correctly (consumer receives 1 message from queue p1, then 1 message from queue p2, then 1 message from queue p1, then 1 message from queue p2, etc...)

$connection = new AMQPConnection(compact('host', 'port', 'login', 'password'));
$connection->connect();

$channel = new AMQPChannel($connection);
$channel->qos(0, 1);

$queue = new AMQPQueue($channel);
$queue->setName('p1');
$queue->consume();

$queue = new AMQPQueue($channel);
$queue->setName('p2');

$queue->consume(function (\AMQPEnvelope $extEnvelope, \AMQPQueue $q) use ($connection) {
    var_dump($extEnvelope->getBody());
    sleep(1);
    $q->ack($extEnvelope->getDeliveryTag());
});

Describe alternatives you've considered

No response

Additional context

No response

[michaelklishin]

@dimanzver please never again use issues for questions in repositories that have Discussions enabled.

It is technically possible to reuse a consumer function (or object) across N queues. Consumers on a shared channel also can have priorities.

However, there is no mechanism for "priorities across N queues", and there never will be. The design you have in mind is a terrible idea (with 11 years on the RabbitMQ core team and some ≈ 14 years as a contributors I feel confident about those strong words). Using a single queue for N message types or a single consumer for N message types often runs into a wall. Imitating priorities across N queues for consumer coordination is guaranteed to not be worth the effort.

This is a consumer coordination problem. Use N consumers across N queues, with a prefetch they need (it could be 1 if you understand the extent to which this option will affect throughput), and coordinate them however you like in your own code.

Or use a stream which will allow consumers to consume and filter out what they want from a shared source. Again, if they need to coordinate, how to best do that is up to you but repeatable reads of streams will generally help. I haven't been following RabbitMQ Stream protocol PHP client development, to my knowledge our team does not maintain one.

[dimanzver]

"Use N consumers across N queues, with a prefetch they need" - how do it? It tried this code:
consume.php:

$connection = new AMQPConnection(compact('host', 'port', 'login', 'password'));
$connection->connect();

$channel = new AMQPChannel($connection);
$channel->qos(0, 1);

$queue = new AMQPQueue($channel);
$queue->setName('p1');
$queue->consume();

$queue = new AMQPQueue($channel);
$queue->setName('p2');

$queue->consume(function (\AMQPEnvelope $extEnvelope, \AMQPQueue $q) use ($connection) {
    echo $extEnvelope->getBody() . "\n";
    sleep(1);
    $q->ack($extEnvelope->getDeliveryTag());
});

publish.php:

$connection = new AMQPConnection(compact('host', 'port', 'login', 'password'));
$connection->connect();

$channel = new AMQPChannel($connection);
$ex = new AMQPExchange($channel);

for ($i = 0; $i < 3; $i++) {
    $ex->publish("p1 $i", 'p1', AMQP_NOPARAM, ['priority' => 7]);
    $ex->publish("p2 $i", 'p2');
}

If I run php publish.php and then consume.php, priority not works. Output of consume.php isn`t correct:

p1 0
p2 0
p1 1
p2 1
p1 2
p2 2

But expected output:

p1 0
p1 1
p1 2
p2 0
p2 1
p2 2

If I publish to p1 queue only, It works.
publish.php:

$connection = new AMQPConnection(compact('host', 'port', 'login', 'password'));
$connection->connect();

$channel = new AMQPChannel($connection);
$ex = new AMQPExchange($channel);

for ($i = 0; $i < 3; $i++) {
    $ex->publish("p1 $i", 'p1', AMQP_NOPARAM, ['priority' => 7]);
    $ex->publish("p2 $i", 'p1');
}

Output:

p1 0
p1 1
p1 2
p2 0
p2 1
p2 2

[michaelklishin]

Use N channels, each with a different prefetch if you want, and N queues. Then messages in the "high priority" queue won't be blocked by others, which is usually the key or the only goal of message priorities.

Since prefetch can be considered a per-channel setting to make things simpler for everyone, you must use N channels to consume from N queues. That's it.

[rbro]

I was going to post a similar question, but I found this discussion. I have a similar setup with a single consumer reading from a single queue with 3 levels of message priorities and a prefetch count of 1. This setup allows me to process the highest priority messages first, and then only if none are remaining, process the lower priority messages in order by priority number.

I see that quorum queues no longer support message priorities and instead the recommendation on https://www.rabbitmq.com/docs/quorum-queues#priorities is:

"To prioritize messages with Quorum Queues, use multiple queues; one for each priority."

I'm trying to figure out how to structure my consumers to prioritize messages if I have multiple queues.

Based on the discussion above, I believe the recommendation is to setup a separate consumer for each queue. Does that mean that all 3 levels of priorities would be processed at the same time through each of the consumers? I only want to process the lower priority messages if there are no high priority messages.

Can you please assist in an example of how to coordinate the consumers to support message priority with multiple queues?

Thank you for your help.

[mkuratczyk]

Please note that in RabbitMQ 4.0, quorum queues will have priority support, although significantly different than how classic queues implement it. You can read more about it in the soon-to-be-published blog post: https://qq-v4.rabbitmq-website.pages.dev/blog/2024/08/08/quorum-queues-in-4.0

[michaelklishin]

@rbro how your consumers process messages from different queues is up to your application, why would RabbitMQ dictate that.

And yes, in 4.0 there will be QQ priority support that implements the same basic idea and supports 3 or so? priorities, which is enough for the proverbial "80-90%" of cases. We will produce a new 4.0 beta release soon and you would be most welcome to try it.

[rbro]

@mkuratczyk and @michaelklishin - thanks for the updates and sending the blog post in advance. A few questions, and I understand if any of these are not possible. I'm just trying to evaluate options for how I need to change my logic:

1. For message priorities with quorum queues, would it be possible to support 5 priorities? This matches existing priority queues where it says "values between 1 and 5 are highly recommended". If not, 3 would be better than 2.

2. The blog post mentions that the approach avoids "starvation" of the low priority messages. Could an option be added that would disable that 2:1 ratio? In my scenarios, I'm fine if progress is not made on the low priority messages if there are higher priority messages in the queue.

3. The blog post also mentions "You can develop a consumer that subscribes to multiple queues and then decides which queue to consume from".

Would you be able to give an example of how such a consumer works? Psuedo-code or any language would be fine. I agree that RabbitMQ shouldn't dictate to my application how consumers should process messages from different queues, but I'm trying to figure out how I would even be able to implement that. Would my consumer receive messages from multiple queues at the same time in some form, judge their priority, and then possibly reject and requeue lower priority messages? I'm not sure how that would be implemented. Or is there another way?

Thanks again for your help.

[mkuratczyk]

1. At this point we only support 2 priorities (there can be more values used, but they all map to low/normal or high)
2. Starvation is also a problem on the server side, which is why we don't want to allow it. A quorum queue that has even 1 old message, cannot truncate its RAFT log and would therefore use more and more disk space.
3. I don't have any examples at this point. Using AMQP-1.0 can be easier for this use case since link credits can be dynamically adjusted to allow messages from one queue or another.

[michaelklishin]

https://github.com/rabbitmq/rabbitmq-server/releases/tag/v4.0.0-beta.4 is out for those who want to try priority support in quorum queues. Note: it is not exactly the same behavior when there is a backlog of messages with different priorities but should be enough for many cases.

[dimanzver]

@michaelklishin This result can be obtained if consume one queue some times. For example

$queue = new AMQPQueue($channel);
$queue->setName('p1');
$queue->consume();
$queue->consume();
$queue->consume();

$queue = new AMQPQueue($channel);
$queue->setName('p2');
$queue->consume($callback);

So we get the ratio 3:1 (3 processed message from p1 queue and 1 processed message from queue p2).

[dimanzver]

@rbro FYI

[dimanzver]

Other varianl - use multiple channels with different prefetch count

$channel = new AMQPChannel($connection);
$channel->qos(0, 3);

$channel2 = new AMQPChannel($connection);
$channel2->qos(0, 1);

$queue = new AMQPQueue($channel);
$queue->setName('p1');
$queue->consume();

$queue = new AMQPQueue($channel2);
$queue->setName('p2');
$queue->consume($callback);

It gives the same result.
@michaelklishin Do I understand correctly that this variant is better?

[michaelklishin]

@dimanzver I don't know how you define "better". There's nothing inherently wrong in having N consumers on N channels on a shared queue but you must be aware of two potential problems:

1. You will lose total ordering of the messages because they will be delivered on different channels concurrently and (potentially, depending on what your consumer does) processed in parallel. This is not something negative but something your code must be ready to handle
2. You won't be able to ack multiple deliveries at a time (by setting the multiple property of basic.ack to true) or very soon your consumers will overlap on the ranges they confirm, which will lead to a channel exception

You also don't get much in terms of relative priorities but you do avoid the fundamental head-of-queue blocking, that's true.