RabbitMQ with Java

Java developers connect to RabbitMQ using the official RabbitMQ Java client library maintained by the RabbitMQ team. The library provides a full AMQP implementation with support for all queue types, exchanges, publisher confirms, and TLS connections. Spring AMQP builds on top of it to offer a higher-level abstraction for Spring-based applications.

Adding the Dependency

Maven

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

Gradle

implementation 'com.rabbitmq:amqp-client:5.20.0'

Connecting to RabbitMQ

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
factory.setRequestedHeartbeat(60);
factory.setConnectionTimeout(30000);

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

Declaring a Queue and Publishing

import com.rabbitmq.client.AMQP;
import java.nio.charset.StandardCharsets;

// Declare a durable queue
channel.queueDeclare("orders", true, false, false, null);

// Build message properties
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .contentType("application/json")
    .deliveryMode(2)   // persistent
    .build();

// Publish a message
String body = "{\"order_id\": 5001, \"product\": \"Gadget\"}";
channel.basicPublish("", "orders", props, body.getBytes(StandardCharsets.UTF_8));
System.out.println("Message published");

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

Consuming Messages

import com.rabbitmq.client.DeliverCallback;

channel.queueDeclare("orders", true, false, false, null);
channel.basicQos(1);  // prefetch count

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
    System.out.println("Received: " + message);
    try {
        processOrder(message);
        channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    } catch (Exception e) {
        System.err.println("Processing failed: " + e.getMessage());
        channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
    }
};

channel.basicConsume("orders", false, deliverCallback, consumerTag -> {});
System.out.println("Waiting for messages...");

Publisher Confirms

channel.confirmSelect();  // enable publisher confirms on this channel

String body = "{\"order_id\": 6001}";
channel.basicPublish("", "orders",
    MessageProperties.PERSISTENT_TEXT_PLAIN,
    body.getBytes(StandardCharsets.UTF_8));

// Wait for single confirmation (blocking)
if (channel.waitForConfirms(5000)) {
    System.out.println("Message confirmed");
} else {
    System.out.println("Message was nacked by broker!");
}

Async Confirms (high throughput)

channel.confirmSelect();

channel.addConfirmListener((deliveryTag, multiple) -> {
    // ack callback
    System.out.println("Confirmed: " + deliveryTag);
}, (deliveryTag, multiple) -> {
    // nack callback
    System.err.println("Nacked: " + deliveryTag + " - retry needed");
});

// Publish without blocking
channel.basicPublish("", "orders",
    MessageProperties.PERSISTENT_TEXT_PLAIN,
    body.getBytes(StandardCharsets.UTF_8));

Working with Exchanges

// Declare a topic exchange
channel.exchangeDeclare("app-events", "topic", true);

// Declare a queue
channel.queueDeclare("order-events", true, false, false, null);

// Bind queue to exchange with routing pattern
channel.queueBind("order-events", "app-events", "order.#");

// Publish to exchange
channel.basicPublish(
    "app-events",
    "order.placed",
    MessageProperties.PERSISTENT_TEXT_PLAIN,
    "{\"order_id\":7001}".getBytes(StandardCharsets.UTF_8)
);

Connection Recovery

The Java client supports automatic connection recovery. Enable it on the factory:

factory.setAutomaticRecoveryEnabled(true);
factory.setNetworkRecoveryInterval(5000);  // retry every 5 seconds
factory.setTopologyRecoveryEnabled(true);  // redeclare queues/exchanges after recovery

With these settings, the Java client automatically reconnects after a network failure and redeclares all queues, exchanges, and bindings that were declared on the previous connection.

Using a Thread Pool for Consumers

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

ExecutorService executor = Executors.newFixedThreadPool(4);
Connection connection = factory.newConnection(executor);

The Java client uses this thread pool for consumer callbacks. Using a thread pool prevents one slow consumer callback from blocking others.

TLS Connection

factory.useSslProtocol();  // uses default TLS settings

// Or with custom trust store
SSLContext sslContext = SSLContext.getInstance("TLSv1.3");
// ... configure TrustManagerFactory and KeyManagerFactory
factory.useSslProtocol(sslContext);
factory.setPort(5671);  // TLS port

Complete Producer Example

import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;

public class OrderProducer {
    private static final String QUEUE = "orders";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setAutomaticRecoveryEnabled(true);

        try (Connection conn = factory.newConnection();
             Channel ch = conn.createChannel()) {

            ch.queueDeclare(QUEUE, true, false, false, null);
            ch.confirmSelect();

            String order = "{\"order_id\": 9001}";
            ch.basicPublish("", QUEUE,
                MessageProperties.PERSISTENT_TEXT_PLAIN,
                order.getBytes(StandardCharsets.UTF_8));

            ch.waitForConfirmsOrDie(5000);
            System.out.println("Order sent and confirmed.");
        }
    }
}

Summary

Java connects to RabbitMQ using the official amqp-client library. Use ConnectionFactory to configure connections and Channel for all operations. Enable automatic connection recovery on the factory for production resilience. Use confirmSelect() and waitForConfirmsOrDie() for reliable publishing. Set basicQos(1) for fair dispatch across multiple consumers. Use async confirm listeners for high-throughput publishing. Enable TLS by calling factory.useSslProtocol() and connecting on port 5671.