In this example we are showing how to create a dynamic load balancer, and how to distribute messages to many consumers. We are going to create a file downloader.
You can find the source at Chapter01/Recipe10/Java_10/.
To use this recipe we need to set up the Java development environment as indicated in the Introduction section.
In order to let two or more RabbitMQ clients properly balance consuming messages, you need to follow the given steps:
- Declare a named queue, and specify the
basicQosas follows:channel.queueDeclare(myQueue, false, false, false,null); channel.basicQos(1);
- Bind a consumer with explicit
ack:channel.basicConsume(myQueue, false, consumer);
- Send one or more messages using
channel.basicPublish(). - Execute two or more consumers.
The publisher sends a message with the URL to download:
String messageUrlToDownload= "http://www.rabbitmq.com/releases/rabbitmq-dotnet-client/v3.0.2/rabbitmq-dotnet-client-3.0.2-user-guide.pdf";
channel.basicPublish("",MyQueue,null,messageUrlToDownload.getBytes());The consumer gets the message and downloads the referenced URL:
System.out.println("Url to download:" + messageURL);
downloadUrl(messageURL);Once the download is terminated, the consumer sends the ack back to the broker and is ready to download the next one:
getChannel().basicAck(envelope.getDeliveryTag(),false);
System.out.println("Ack sent!");
System.out.println("Wait for the next download...");By default, messages are heavily prefetched. Messages are retrieved by the consumers in blocks, but are actually consumed and removed from the queue when the consumers send the ack, as already seen in the previous recipe.
On the other hand, using many consumers as in this recipe, the first one will prefetch the messages, and the other consumers started later won't find any available in the queue. In order to equally distribute the work among the active consumers, we need to use channel.basicQos(1), specifying to prefetch just one message at a time.
You can find more information about load balancing in Chapter 8, Performance Tuning for RabbitMQ.