Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How does RabbitMQ send messages to consumers?

Tags:

rabbitmq

amqp

I am a newbie to RabbitMQ, hence need guidance on a basic question:

Does RabbitMQ send messages to consumer as they arrive?
OR
Does RabbitMQ send messages to consumer as they become available?

  • At message consumption endpoint, I am using com.rabbitmq.client.QueueingConsumer.
  • Looking at the sprint client source code, I could figure out that
    • QueueingConsumer keeps listening on socket for any messages the broker sends to it
    • Any message that is received is parsed and stored as Delivery in a LinkedBlockingQueue encapsulated inside the QueueingConsumer.
  • This implies that even if the message processing endpoint is busy, messages will be pushed to QueueingConsumer

Is this understanding right?

like image 429
Amit Sharma Avatar asked Jun 19 '14 14:06

Amit Sharma


People also ask

What is RabbitMQ and how it works?

Let’s briefly have a look at how RabbitMQ works. Let's first familiar with a few important concepts of RabbitMQ: Producer: Application that sends the messages. Consumer: Application that receives the messages. Queue: Buffer that stores messages. Message: Information that is sent from the producer to a consumer through RabbitMQ.

How are messages delivered by RabbitMQ?

When a consumer (subscription) is registered, messages will be delivered (pushed) by RabbitMQ using the basic.deliver method. The method carries a delivery tag, which uniquely identifies the delivery on a channel.

What is a message broker in RabbitMQ?

RabbitMQ is an open source message broker software. It accepts messages from producers, and delivers them to consumers. It acts like a middleman which can be used to reduce loads and delivery times taken by web application servers. Here P is Producer, C is Consumer and middle one is queue. String message = “Hello World!”;

When are consumers recovered from a RabbitMQ queue?

In other words, consumers are usually recovered last, after their target queues and those queues' bindings are in place. Applications can subscribe to have RabbitMQ push enqueued messages (deliveries) to them. This is done by registering a consumer (subscription) on a queue.


2 Answers

TLDR: you poll messages from RabbitMQ till the prefetch count is exceeded in which case you will block and only receive heart beat frames till the fetch messages are ACKed. So you can poll but you will only get new messages if the number of non-acked messages is less than the prefetch count. New messages are put on the QueueingConsumer and in theory you should never really have much more than the prefetch count in that QueueingConsumer internal queue.

Details: Low level wise for (I'm probably going to get some of this wrong) RabbitMQ itself doesn't actually push messages. The client has to continuously read the connections for Frames based on the AMQP protocol. Its hard to classify this as push or pull but just know the client has to continuously read the connection and because the Java client is sadly BIO it is a blocking/polling operation. The blocking/polling is based on the AMQP heartbeat frames and regular frames and socket timeout configuration.

What happens in the Java RabbitMQ client is that there is thread for each channel (or maybe its connection) and that thread loops gathering frames from RabbitMQ which eventually become commands that are put in a blocking queue (I believe its like a SynchronousQueue aka handoff queue but Rabbit has its own special one).

The QueueingConsumer is a higher level API and will pull commands off of that handoff queue mentioned early because if commands are left on the handoff queue it will block the channel frame gathering loop. This is can be bad because timeout the connection. Also the QueueingConsumer allows work to be done on a separate thread instead of being in the same thread as the looping frame thread mentioned earlier.

Now if you look at most Consumer implementations you will probably notice that they are almost always unbounded blocking queues. I'm not entirely sure why the bounding of these queues can't be a multiplier of the prefetch but if they are less than the prefetch it will certainly cause problems with the connection timing out.

like image 94
Adam Gent Avatar answered Sep 21 '22 15:09

Adam Gent


I think best answer is product's own answer. As RMQ has both push + pull mechanism defined as part of the protocol. Have a look : https://www.rabbitmq.com/tutorials/amqp-concepts.html

like image 32
siddhusingh Avatar answered Sep 17 '22 15:09

siddhusingh