Create Dynamic Queues in RabbitMQ

1. Introduction

RabbitMQ is a message broker that provides asynchronous communication between various components. It provides an implementation of the AMQP (Advanced Message Queuing Protocol), the most popular messaging protocol.

In this tutorial, we’ll look at how we can create queues dynamically in RabbitMQ using the Java client library.

2. RabbitMQ Messaging Model

Before we begin, let’s quickly overview how the RabbitMQ messaging works.

We first need to understand the building blocks of AMQP, also called AMQP entities. Exchanges, queues, and bindings are collectively referred to as AMQP Entities.

In RabbitMQ, a message producer never sends a message directly to a queue. Instead, it uses an exchange as a routing mediator. The message producer publishes the messages to the exchanges. Exchanges then route these messages to various queues based on the routing rules called bindings. The broker then delivers the messages to consumers subscribed to the queues, or consumers pull/fetch messages from the queue on demand. Delivery of messages to the consumers happens based on the FIFO Model.

3. Connection Initialization

RabbitMQ provides client libraries for all major programming languages. With Java, the standard way for a client to communicate with the RabbitMQ Broker is to use RabbitMQ’s amqp-client Java library. Let’s add this library’s Maven dependency to our project’s pom.xml file:

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.16.0</version>
</dependency>

For a client to be able to interact with the RabbitMQ Broker, we first need to establish a Connection. Once we have established a connection, we can create a Channel from the existing Connection. AMQP channels are basically lightweight connections that share a single TCP connection. A channel helps us multiplex multiple logical connections on top of a single TCP connection.

The Java RabbitMQ client library uses the Factory pattern to create new connections.

Firstly, let’s create a new ConnectionFactory instance. Next, we’ll set all the parameters needed to create connections. At a bare minimum, this requires informing the address of the RabbitMQ host:

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("amqp.baeldung.com");

Secondly, we will use the newConnection() factory method from the ConnectionFactory instance we created to get a new Connection instance:

Connection connection = factory.newConnection();

Lastly, let’s create a new Channel from an existing Connection using the createChannel() method:

Channel channel = connection.createChannel();

We successfully connected with the RabbitMQ Broker and created a Channel. We are now ready to send commands to the RabbitMQ server using the created Channel.

Also, we can set up different strategies for channels with single or multiple connections.

4. Creating Dynamic Queues

The Java RabbitMQ client library provides various easy-to-use methods to create and manage queues. Let’s have a look at some of the important methods:

4.1. Creating a Queue

To dynamically create a queue, we use the queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) method from the Channel instance we created earlier. This method creates a queue if it does not already exists. The method accepts the following arguments:

• queue – the name of the queue to be created
• durable – boolean flag denoting whether the queue to be created should be durable (i.e., the queue will survive a server restart)
• exclusive – boolean flag denoting whether the queue to be created should be exclusive (i.e., restricted to this connection)
• autoDelete – boolean flag denoting whether the queue to be created should be autodeleted (i.e., the server will delete it when no longer in use)
• arguments – other properties for the queue

Let’s have a look at the Java code to create a queue:

AMQP.Queue.DeclareOk declareOk = channel.queueDeclare("baeldung-queue", true, false, false, null);

The above code snippet creates a queue named ‘baeldung-queue’. Upon successful creation of the queue, the method will return an instance of AMQP.Queue.DeclareOk. The method throws an IOException in case of any errors in creating the queue.

Further, we will use the returned AMQP.Queue.DeclareOk instance to obtain information about the queue – such as the queue name, count of consumers for the queue, and the count of messages contained in the queue. Let’s have a look at the code snippet to get the queue name from the DeclareOk instance:

String queueName = declareOk.getQueue();

The above snippet will return the name of the created queue. Similarly, we can fetch the consumer count and message count for a queue:

int messageCount = declareOk.getMessageCount();
int consumerCount = declareOk.getConsumerCount();

We have seen how we can dynamically create queues using the RabbitMQ Java client library. Next, let’s have a look at how we can check for the existence of a queue.

4.2. Checking if a Queue Exists

The RabbitMQ Java client library also provides a method to check if the queue exists. We will use the method queueDeclarePassive(String queue) to check if the queue exists. This method returns an instance of AMQP.Queue.DeclareOk as a confirmation that the queue exists. If the queue does not exist, the queue is exclusive, or if there is any other error, the method will throw an IOException.

Let’s have a look at the Java code to check if the queue already exists:

AMQP.Queue.DeclareOk declareOk = channel.queueDeclarePassive("baeldung-queue");

The code snippet checks whether the queue “baeldung-queue” exists.

Lastly, we close the channel and the connection:

channel.close();
connection.close();

We can also use try-with-resources to initialize channel and connection objects so that they are auto-closed.

5. Conclusion

In this article, we first looked at how we can establish a connection with the RabbitMQ Server and open a communication channel.

We then used the RabbitMQ Java client library to demonstrate how we can create a queue dynamically and check for its existence.