Dispatch to consumers - cannot use round robin

[Doug Sievers]

Currently, I have a single queue with a single consumer that process messages in order (FIFO).

We need to scale out our application and add more consumers... at least two, maybe more.

In our application FIFO is necessary, as the messages represent operations on objects.

Think of building a car... it must be assembled in a specific order.

According to tutorial #2 on works queues...

"By default, RabbitMQ will send each message to the next consumer, in sequence.

On average every consumer will get the same number of messages.

This way of distributing messages is called round-robin."

Round-robin will not work for us in this case, as the queue may have the following elements

(head of queue, next to process)

1 - Car serial # 100, insert engine

2 - Car serial # 100, mount hood

3 - Car serial # 101, insert engine

4 - Car serial # 102, insert engine

In this case, with 2 consumers named A and B, message #1 would get dispatched to Consumer A, and message #2 would get dispatched to Consumer B.

But, inserting the engine must be guaranteed to complete processing before mounting the hood, that is I must make sure message #1 is processed before message #2.

If they are processed by different consumers, I have no way to guarantee that will happen.

My idea to solve this problem would be to dispatch messages to a consumer based on the last digit of the serial number.

For example

- 1 consumer, all digits 0-9 go to single consumer

- 2 consumers, digits 0-4 go to consumer A, digits 5-9 go to consumer B

- 3 consumers (you get the idea)

I have seen a few proposed solutions

1 - write a "man-in-the-middle" type client that reads messages from the original queue and publishes them to separate queues based on serial number

I think this is probably the easiest to write, but I need to make sure that messages are not building up in the separate workers queues in case one of them fails.

In the event of failure of one consumer, we must switch back to the single consumer processing all messages so they stay in order.

We are already using prefetch, but I am not sure if prefetch would be safe here?

I would need to pass the ack backward from the consumer, to the "man-in-the-middle", then all the way back to the main queue, I think.

2 - it might be possible to write a custom exchange that distributes messages to the proper queue

However, again I need to make sure messages do not get backed up in the queue, for same reasons as #1

3 - is there a way to customize how messages are distributed from the queue to the consumers?

The tutorial says "by default" but I couldn't find info about changing the default or if there are settings to adjust that.

What classes/code would I need to modify / extend to make a plugin for such a change?

If anyone knows of a plugin or other software (open source) that offers this, that would work great.

I don't know Erlang but could learn if necessary.

Thank you! :)

- Doug

[Daniil Fedotov]

Hi,

The AMQP protocol separates message routing and delivery. The routing part is performed on exchanges and bindings, while delivery is done by queues. If you want to filter messages based on some data, you do that with exchanges and bindings. Queues will deliver messages to all consumers connected.

Round-robin does not guarantee causal order, when a message is delivered only after another one was processed.

You might achieve causal order on message broker, but you will have to have a single queue with a single consumer for a sequence, as you've mentioned.

You can route messages with bindings and routing keys. This page explains the exchanges basics https://www.rabbitmq.com/tutorials/amqp-concepts.html

I would not recommend moving your control logic to RabbitMQ, both for performance reasons and complexity.

You can have additional signals from workers informing that stage X for sequence Y has been completed and act on that message.

You can make worker for stage 1 publish a message, that the stage is completed and the next stage is available, or an RPC response to the controller, which will publish a new stage message after that.

You can check out RPC scenario for example http://www.rabbitmq.com/tutorials/tutorial-six-python.html

It's up to you how you do that in the end.

[Doug Sievers]

Hi Daniil,

Thank you for the e-mail.

There are some problems with what you wrote:

On Monday, September 24, 2018 at 2:01:10 AM UTC-7, Daniil Fedotov wrote:

Queues will deliver messages to all consumers connected.

That is not true. Tutorial #2 (example in python https://www.rabbitmq.com/tutorials/tutorial-two-python.html ) explicitly states that messages are dispatched to consumers as round-robin.

I would not recommend moving your control logic to RabbitMQ, both for performance reasons and complexity.

As mentioned, I already have a queue - we are not moving to rabbitMQ, we already use it, and it works awesome so far.

 

You can make worker for stage 1 publish a message, that the stage is completed and the next stage is available, or an RPC response to the controller, which will publish a new stage message after that.

You can check out RPC scenario for example http://www.rabbitmq.com/tutorials/tutorial-six-python.html

As you can read on Tutorial #6, they clearly recommend AGAINST using RPC.

 

I still need someone to give me some advice on how I might extend rabbitMQ or implement middleware that handles the problems originally outlined.

Thanks!

[Daniil Fedotov]

Round-robin is a manner to deliver messages, and they will be delivered to all of the consumers, given enough messages on the queue. There is no filtering on the queue-consumer level.

To be fair, there is no guarantee for round-robin delivery. There is a prefetch setting for consumer, which controls how many messages it can receive from the queue before acknowledging. This is for manual acknowledgements mode. When consumer exceeds the prefetch and does not acknowledge messages, it will not receive new messages from the queue. Other consumers though can continue acknowledging and receiving new messages. So the order of consumers getting messages depends on their performance and is not guaranteed.

You should not rely on queue-consumer level if you want to get specific messages to specific consumers, the AMQP protocol requires you to use exchanges and bindings for that.

The part which discourages RPC recommends to use asynchronous pipeline. I don't know how your code is controlled, so I cannot recommend one or another, but in general scenario both should work.

The pipeline is when there is a queue per stage instead of sequences. A stage 1 worker publishes a message to stage 2 queue with a sequence X metadata as soon as it finishes the stage 1. A stage N worker publishes message to stage N+1 queue. This will allow asynchronous stage processing.

This can be implemented using RPC if your control code is separated from workers code.

Cheers.

[Doug Sievers]

Hi All,

I would appreciate if anyone else has suggestions or solutions for my use case.

It seems there are issues with round-robin, but there must be a way to change this.

The consumer priority plugin (https://www.rabbitmq.com/consumer-priority.html) can modify how messages are distributed to consumers.

Maybe I can build my own plugin with this as a starting point.

Currently all messages use the same exchange and routing key. We cannot change the publishers, as that software is written by many different third parties for different machines sending data - cost and time prohibitive.

Even if I could change the publishers to send to different queues (with each queue being serviced by a single consumer) then I have messages piling up if one consumer dies.

Also, adding more scaling with additional consumers down the road would require doing everything all over again - cost and time prohibitive.

If there is some exchange/topic/routing/binding setting that I am not aware of, please give me a specific example of how I would set it up to do content-based routing as initially stated.

 Thank you.

--
You received this message because you are subscribed to a topic in the Google Groups "rabbitmq-users" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/rabbitmq-users/Knk5JOL-XyM/unsubscribe.
To unsubscribe from this group and all its topics, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Michael Klishin]

There is no way to to customise how messages are distributed to consumers besides

QoS [1] and consumer priorities. By the way, consumer priorities are not a plugin, it's a built-in feature.

You *can* have competing consumers on a queue and make sure only one message of each type is being processed

at a time, with any messaging technology. The problem here is that of consumer coordination. We highly recommend using

a service built for coordinating other services (ZooKeeper, etcd, even a key/value store with suitable consistency guarantees might work).

I don't think routing is really the problem here, since no matter what already available exchange type you use you will run into

more or less the same problem.

Manual acknowledgements and QoS [1] then can be used to reduce or eliminate concurrent deliveries to the leader consumer

on a given queue.

Note that consumers in RabbitMQ can also be made exclusive (see the "exclusive" argument to Channel#basicConsume in your client

or its equivalent). That does not eliminate the general problem of consumer coordination but might be a decent safety measure to add.

If you do use competing consumers, make sure you read [2] to understand how connection failures are detected since it becomes

a critically important aspect for leader consumer failure tolerance.

HTH.

1. http://www.rabbitmq.com/confirms.html

2. http://www.rabbitmq.com/heartbeats.html

To unsubscribe from this group and all its topics, send an email to rabbitmq-user...@googlegroups.com.
To post to this group, send email to rabbitm...@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-user...@googlegroups.com.
To post to this group, send email to rabbitm...@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

--

MK

Staff Software Engineer, Pivotal/RabbitMQ

[Srinath C]

Doug,

Do you think a consistent hash exchange like [1] would help?

The messages need to be published with the car serial # as the routing key.

Each consumer can bind a queue to the exchange to receive all messages from the same set of serial #s. 

In the case of a consumer failure, you probably need to relaunch or launch a new consumer (via process monitoring) so that the messages in the queue get processed.

[1] https://github.com/rabbitmq/rabbitmq-consistent-hash-exchange

To unsubscribe from this group and all its topics, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

--

MK

Staff Software Engineer, Pivotal/RabbitMQ

--

You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.