How to Connect, Send, and Receive Messages with RabbitMQ in Java

RabbitMQ Basic Model

In the messaging model, a producer (P) sends messages to a queue, and a consumer (C) retrieves messages from that queue. The queue acts as a container storing messages until they are consumed.

Obtaining a RabbitMQ Connection in Java

Similar to acquiring a database connection, you first create a ConnectionFactory, configure the host, port, virtual host, username, and password, and then obtain a Connection instance.

package com.mmr.rabbitmq.conn;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;

public class ConnectionUtils {
    public static Connection getConnection() throws IOException, TimeoutException {
        // Define connection factory
        ConnectionFactory factory = new ConnectionFactory();
        // Set server address
        factory.setHost("127.0.0.1");
        // Set AMQP port (default 5672)
        factory.setPort(5672);
        // Set virtual host, username, and password
        factory.setVirtualHost("/vhost_mmr");
        factory.setUsername("user_mmr");
        factory.setPassword("admin");
        // Obtain connection
        Connection connection = factory.newConnection();
        return connection;
    }
}

Producer: Sending Messages to the Queue

The producer creates a channel, declares the target queue (if it does not already exist), and publishes a message.

public class SendMQ {
    private static final String QUEUE_NAME = "QUEUE_simple";

    @Test
    public void sendMsg() throws Exception {
        // Get a connection
        Connection connection = ConnectionUtils.getConnection();
        // Create a channel
        Channel channel = connection.createChannel();
        // Declare the queue (non‑durable, non‑exclusive, non‑auto‑delete)
        boolean durable = false;
        boolean exclusive = false;
        boolean autoDelete = false;
        channel.queueDeclare(QUEUE_NAME, durable, exclusive, autoDelete, null);
        String msg = "Hello  Simple QUEUE !";
        // Publish to the default exchange with the queue name as routing key
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        System.out.println("---------send ms :" + msg);
        channel.close();
        connection.close();
    }
}

Consumer: Receiving Messages from the Queue

The consumer also obtains a connection and channel, then registers a DefaultConsumer to handle incoming deliveries.

package com.mmr.rabbitmq.simple;
import java.io.IOException;
import com.mmr.rabbitmq.conn.ConnectionUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;

public class Consumer {
    private static final String QUEUE_NAME = "QUEUE_simple";

    public static void main(String[] args) throws Exception {
        // Get a connection
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        // Declare the queue if needed (optional)
        // channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            }
        };
        // Start listening
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }

    @SuppressWarnings("deprecation")
    private static void oldGet(Channel channel) throws IOException, InterruptedException {
        // Legacy QueueingConsumer example
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(QUEUE_NAME, true, consumer);
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String message = new String(delivery.getBody());
            System.out.println(" [x] Received '" + message + "'");
        }
    }
}

Limitations of a Simple Queue

The simple queue model has high coupling: each message is consumed by a single consumer, making it unsuitable when multiple consumers need the same message. Changing the queue name also requires updating all related code.