Table of Contents

1. Overview

2. Prerequisites

3. Adding Dependenceis

4. Producer Implementation

5. Consumer Implementation

6. Execution Demo

7. Practical Application Example

8. Overview

RabbitMQ is an open-source message broker software that implements the Advanced Message Queuing Protocol (AMQP). It serves as an intermediary for receiving, storing, and forwarding messages between producers and consumers. The core architecture relies on queues for message storage and exchanges for message routing to appropriate queues.

This guide demonstrates how to use RabbitMQ's Java client library to build message producers and consumers, enabling asynchronous message sending and receiving in distributed applications.

2. Prerequisites

Before proceeding, ensure that:

• RabbitMQ server is installed and running
• Java Development Kit (JDK) 8 or higher is configured
• Maven is available for dependency management

3. Adding Dependencies

Add the following dependencies to your Maven project's pom.xml file within the <dependencies> section:

<!-- RabbitMQ Client -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.12.0</version>
</dependency>

<!-- SLF4J API -->
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>1.7.32</version>
</dependency>

<!-- Logback Implementation -->
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
    <version>1.2.6</version>
</dependency>

4. Producer Implementation

The producer is responsible for publishing messages to a RabbitMQ queue. Below is a complete implementation example:

package org.example.messaging;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MessagePublisher {
    private static final String QUEUE_NAME = "notification_queue";
    private static final Logger logger = LoggerFactory.getLogger(MessagePublisher.class);

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.100.146");
        factory.setUsername("guest");
        factory.setPassword("guest");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            String messageContent = "Hello, RabbitMQ!";
            
            channel.basicPublish("", QUEUE_NAME, null, messageContent.getBytes());
            logger.info("Message sent: {}", messageContent);
            System.out.println(" [✓] Published: '" + messageContent + "'");
        }
    }
}

Key operations in the producer:

• ConnectionFactory: Creates connections to the RabbitMQ server
• queueDeclare(): Declares a queue with specified parameters
• basicPublish(): Publishes a message to the default exchange

5. Consumer Implementation

The consumer subscribes to a queue and processes incoming messages. Here is the implementation:

package org.example.messaging;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MessageSubscriber {
    private static final String QUEUE_NAME = "notification_queue";
    private static final Logger logger = LoggerFactory.getLogger(MessageSubscriber.class);

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.100.146");
        factory.setUsername("guest");
        factory.setPassword("guest");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println("[*] Waiting for messages. Press Ctrl+C to exit.");

            DeliverCallback messageHandler = (consumerTag, delivery) -> {
                String receivedMessage = new String(delivery.getBody(), "UTF-8");
                logger.info("Message received: {}", receivedMessage);
                System.out.println("[✓] Received: '" + receivedMessage + "'");
            };

            channel.basicConsume(QUEUE_NAME, true, messageHandler, consumerTag -> {});
        }
    }
}

Key operations in the consumer:

• DeliverCallback: Defines the message processing logic
• basicConsume(): Starts consuming messages from the specified queue
• Auto-acknowledge mode (second parameter set to true)

6. Execution Demo

To test the producer-consumer implementation:

1. Start the consumer application first - it will block and wait for messages
2. Run the producer application to publish a message
3. Observe the console output confirming successful message delivery

The consumer should display the received message, demonstrating the message flow from producer to consumer through RabbitMQ.

7. Practical Application Example

In e-commerce platforms, order processing is a common scenario where message queuing provides significant benefits.

7.1 Scenario Description

Consider an online shopping platform where order fulfillment involves multiple steps: inventory verification, payment processing, shipping arrangement, and notification delivery. These operations can be time-consuming. By implementing asynchronous processing with RabbitMQ, the system can return immediate responses to users while processing orders in the background.

7.2 Architecture Design

• When an order is placed successfully, publish the order details to the order queue
• Backend order processing service subscribes to the queue and processes orders sequentially
• After processing, update order status and optionally publish completion events for other serviecs (notification service, inventory service, etc.)

7.3 Implementation Example

Order Publisher:

package org.example.ecommerce;

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

public class OrderPublisher {
    private static final String ORDER_QUEUE = "order_processing_queue";

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

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(ORDER_QUEUE, false, false, false, null);
            String orderData = "New order placed!";
            
            channel.basicPublish("", ORDER_QUEUE, null, orderData.getBytes());
            System.out.println(" [✓] Order published: '" + orderData + "'");
        }
    }
}

Order Processor (Consumer):

package org.example.ecommerce;

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

public class OrderProcessor {
    private static final String ORDER_QUEUE = "order_processing_queue";

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

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(ORDER_QUEUE, false, false, false, null);
            System.out.println("[*] Waiting for orders. Press Ctrl+C to exit.");

            DeliverCallback processingHandler = (consumerTag, delivery) -> {
                String orderDetails = new String(delivery.getBody(), "UTF-8");
                System.out.println("[✓] Order received: '" + orderDetails + "'");
                
                // Simulate order processing workflow
                processOrderWorkflow();
                
                System.out.println("[✓] Order processed successfully!");
            };

            channel.basicConsume(ORDER_QUEUE, true, processingHandler, tag -> {});
        }
    }

    private static void processOrderWorkflow() throws InterruptedException {
        Thread.sleep(2000);
    }
}

This implementation demonstrates how RabbitMQ enables asynchronous order processing, improving system responsiveness and scalability. The user receives immediate confirmation while order processing occurs in the background.