Introduction
In this part of the series we’ll look at 4 basic message exchange patterns (MEP):
One way
Worker queues
Publish/Subscribe
Remote Procedure Call (RPC)
Most of the posts on RabbitMQ on this blog are based on the work of RabbitMQ guru Michael Stephenson.
For the demos you can start a new Visual Studio solution or re-use the one we’ve been working on so that you have all code references in one place.
A general note: we’ll write a lot of example code in this post. We’ll be concentrating on writing code that works and will not follow any software design principles such as SOLID or DRY. That would only slow us down in a large topic like this. Use the link provided to improve the library as you wish.
One way messaging
This is the simplest MEP: a message is sent to the broker which is then processed by the receiver.
Open the RabbitMQ management UI at http://localhost:15672/ and have it ready throughout the demo. Fire up Visual Studio and either open the same solution as before or create a new blank one. Add a C# class library called RabbitMqService. Add the NuGet RabbitMQ package to it as we did in the first part of this series. Add new class called AmqpMessagingService. Add the following private fields:
Add the following method to create a connection to the RabbitMQ server:
This method will set up the queue we’ll use for the one way message demo:
Next add a new Console application to the solution called OneWayMessageSender. Add the RabbitMQ NuGet package there as well and also add a reference to the RabbitMqService library. Insert the following code to Main and run the Sender app:
Check in the RabbitMQ console that the queue called “OneWayMessageQueue” has been set up. Comment out the call to…
Add the following code to send a single message to the queue in AmqpMessagingService.cs:
This code should be familiar from the previous part. Add the following method to Program.cs in the Sender application:
We send the message entered by the Sender to the appropriate queue.
Add a call to this method in Main:
Create another Console application called OneWayMessageReceiver to the solution. Add the NuGet RabbitMQ package to it. Add a project reference to RabbitMqService. Insert the following code to Main:
The first three lines of code should be familiar. ReceiveOneWayMessages has the following implementation in AmqpMessagingService:
BasicQos means basic quality of service. The parameters mean that we require one message at a time and we don’t want to process any additional messages until this one has been processed. You can use these parameters to receive messages in batches.
QueueingBasicConsumer is built into RabbitMQ and is used to consume messages from a specified queue. We use the IModel’s BasicConsume method to consume messages and specify the queue name and the consumer. With ‘false’ we also indicate that we don’t want to auto-acknowledge the messages. Then in the loop we constantly pull message from the queue and acknowledge them with BasicAck. The Queue.Dequeue() method will block the thread until a message has been delivered into the queue. We extract the message byte array from the BasicDeliverEventArgs object. The acknowledgement will release the message from the queue and will allow us to receive the next message.
Let’s see if this works. Set the Receiver as the start up project and start the application. The Receiver app will start. Then in VS right-click the Sender application, click Debug, Start new instance. Enter a message in the Sender windows and press Enter. If everything works fine then the message should show up in the Receiver window:
Send a couple more messages to confirm that the setup works. Set a breakpoint within the while-loop of ReceiveOneWayMessages. You’ll see that execution will stop at…
…and will only continue if there’s a message in the queue. In other words the loop won’t just continue asking for new data all the time.
Worker queues
In this MEP a message is sent by the sender. There will be many listeners waiting for messages from the same queue. However, those listeners compete to receive the message and only one of them will receive it. The purpose is that if an application is expecting to receive a large load of messages then it can create different threads/processes to process those messages. The benefit is better scalability. For the demo we’ll set up a sender and two receivers.
Add the following private field to AmqpMessagingService:
…and the following method to create the queue for this sample:
Add a new console application to the solution called WorkerQueueSender. Add the RabbitMQ NuGet package and a reference to the RabbitMqService library. Insert the following code in Main to set up the queue:
Set WorkerQueueSender as the startup project and run the application. Check the RabbitMQ UI that the queue has been set up. Comment out the call to SetUpQueueForWorkerQueueDemo in Main.
Add the following method in AmqpMessagingService:
…and the one below to receive messages from the worker queue:
It is identical to ReceiveOneWayMessages except for the queue name.
Back in WorkerQueueSender.Program.cs add the following method and add a call to it from Main:
As you see it is identical to what we had in the previous demo. We’ll create two Receivers and they will be identical to the receiver we had in the previous demo. Add two new Console apps: WorkerQueueReceiverOne and WorkerQueueReceiverTwo. In both projects do the following:
Add RabbitMQ package through NuGet
Add a library reference to RabbitMqService
Add the following code to Program.cs.Main:
Follow these steps to start the demo:
Set WorkerQueueSender as the startup project
Start the application
Right-click WorkerQueueReceiverOne, Debug, Start new instance
Do the same for WorkerQueueReceiverTwo
You should have 3 console windows up and running on your screen. Start sending messages in the Sender window. You should see that messages will alternate between receiver one and two:
You should never see that the same message is delivered to both receivers.
The Worker Queue MEP can be implemented with very little extra effort compared to the One Way Message MEP. This MEP helps you create a horizontally scalable server where multiple receivers are set up to collect the incoming messages.
Publish/Subscribe
In this MEP a message is sent to an exchange and the exchange distributes the message to all queues bound to it. Each queue will have its listener to process the message. If you recall the different exchange types then this sounds like the Fan-out type. We’ll set up a dedicated exchange for this, i.e. not use the default one in RabbitMQ.
Enter the following private fields in AmqpMessagingService.cs:
…and the following method where we set up the exchange, 2 queues and bind both queues to the exchange:
Consult the first part in this series if don’t recall what these methods do.
Add a new Console project to the solution called PublishSubscribeSender. Perform the usual actions:
Add RabbitMQ via NuGet
Add a reference to RabbitMqService
In Main insert the following code to set up the necessary infrastructure:
Set PublishSubscribeSender as the startup application and then run it. Check in the RabbitMQ UI whether the exchange and the two queues have been created and if the bindings are OK. Then comment out the call to messagingService.SetUpExchangeAndQueuesForDemo. Add the following method to start sending messages:
As you see it’s not much different from what we had in the previous demos. SendMessageToPublishSubscribeQueues looks like this in AmqpMessagingService:
We’re sending the message to the designated exchange with no routing key specified.
Add two new Console applications: PublishSubscribeReceiverOne and PublishSubscribeReceiverTwo. Apply the following to both:
Add RabbitMQ via NuGet
Add a reference to RabbitMqService
In PublishSubscribeReceiverOne.Program.cs.Main add the following code:
In PublishSubscribeReceiverTwo.Program.cs.Main add the following code:
…where ReceivePublishSubscribeMessageReceiverOne and ReceivePublishSubscribeMessageReceiverTwo look like this in AmqpMessagingService:
As you see there’s not much difference compared to how the Receiver extracted the messages before. The subscription model is represented by the Subscription object in RabbitMQ .NET. The BasicDeliverEventArgs object is returned by the Next() method of the subscription. We then show the message and acknowledge it.
To run this demo:
Run PublishSubscribeSender
Start a new instance of PublishSubscribeReceiverOne the way we did above with WorkerQueueReceiverOne
Start a new instance of PublishSubscribeReceiverTwo the way we did above with WorkerQueueReceiverTwo
You should have three black console screens up and running
Start sending messages on the Sender window. The message should appear on both receivers:
The receivers are listening on two different queues hence they are not competing with each other like in the Worker Queue MEP.
Remote Procedure Call (RPC)
RPC is slightly different from the above three MEPs in that there’s a response queue involved. The sender will first start listening on a response queue before sending any message. It then sends a message to a destination queue via the default exchange where the message includes a property indicating the response queue. The response queue will be dynamically created by the sender. The receiver processes the message and responds using the response queue extracted from the message. The sender then processes the response.
Add the following method to AmqpMessagingService.cs that sets up the queue for this demo:
…where _rpcQueueName is a new private field:
Add a new Console app called RpcSender. Add the usual references: RabbitMQ NuGet, RabbitMqService. Insert the following code to Main to set up the queue:
Set RpcSender as the startup project and run the application. Check in the RabbitMQ management UI that the queue has been set up. Comment out the call to messagingService.SetUpQueueForRpcDemo(model). This queue will be used as the default queue by the sender to send messages. The response queue will be dynamically set up.
Insert the following method to RpcSender.Program.cs to start sending messages:
This setup is very similar to what we’ve seen up to this point. Note, however, that the SendRpcMessageToQueue method returns a string, which will be the response from the Receiver. We also specify a timeout parameter for the response to arrive.
Declare a new method in AmqpMessagingService:
The sender in this case will also need to listen to messages so it will need a QueueingBasicConsumer object we saw before. Also, the response queue will be set up dynamically. The QueueDeclare() method without any parameter will create a temporary response queue. The name of the temporary queue will be randomly generated, e.g. “amq.gen-3tj4jtzMauwolYqc7CUj9g”. While you’re running the demo in a bit you can check the list of queues in the RabbitMQ management UI. The temporary queue will be available as long as the Sender is running. After that it will be removed automatically. Insert the following code to SendRpcMessageToQueue:
…where _rpcConsumer and _responseQueue are private variables:
The sender will listen on that temporary response queue. Append the following code to SendRpcMessageToQueue:
We create a message correlation ID to be able to match the sender’s message to the response from the receiver. If the receiver is responding to another message then it will be ignored. We then set up the IBasicProperties object and specify the temporary queue name to reply to and the correlation ID. Next we publish the message using BasicPublish like before.
Then we enter something that only receivers have done up to now: listen. The sender will listen for the duration of the timeout date. When a response comes then the correlation IDs must be compared. If there’s a match then the response is returned. Otherwise it’s ignored. If there’s no response before the timeout then an exception is thrown.
Let’s look at the receiver now. Add a new Console application called RpcReceiver, add RabbitMQ and RabbitMqService to the reference list. Insert the following code to Main:
…where ReceiveRpcMessage in AmqpMessagingService looks like this in AmqpMessagingService:
Most of this code looks familiar by now I hope. We extend the “normal” receiving logic with the ability to send a response. We extract the correlation ID from the sender’s message so that our response will have the same ID. We send the response to the ReplyTo queue which was also extracted from the sender’s message. We finally acknowledge the reception of the message from the sender.
Let’s run this:
Make sure that RpcSender is the startup project and run the application
Start RpcReceiver the same way as before (Run new instance)
You should have 2 console screens up and running
Send a message from the sender to the receiver. Then send a response. It looks like a very primitive chat application:
I hope you agree that it wasn’t too difficult to implement these 4 basic message exchange patterns.