How to Build Reliable Order Processing with RabbitMQ – A Complete Guide

Brief Introduction

Open‑source AMQP implementation written in Erlang, supporting multiple clients.

Distributed, high‑availability, persistent, reliable, and secure.

Supports AMQP, STOMP, MQTT, HTTP protocols.

Ideal for decoupling business logic between heterogeneous systems.

Basic Concepts

1. Exchange

An exchange receives messages and routes them to bound queues. Four types:

direct – exact match routing.

topic – pattern‑based routing.

fanout – broadcast.

headers – key‑value pair routing.

Exchange attributes:

Durable – survives broker restarts.

Auto‑delete – removed when all bound queues are deleted.

2. Queue

A RabbitMQ internal object that stores messages; it can also be durable or auto‑delete. Consumers pull messages from a queue; multiple consumers share the load.

3. Binding

Associates an exchange with a queue using a routing rule.

4. Routing Key

The keyword used for routing decisions.

Message Reliability

Message acknowledgment: Messages are removed only after the consumer sends an ack; otherwise they are re‑delivered.

Message durability: Persist messages to survive broker restarts; both exchange and queue must be durable.

Prefetch count: Number of messages delivered to a consumer at a time (default 1).

For high‑throughput scenarios you may disable durability, use no‑ack, and enable auto‑delete.

Simple Application Scenario

When a user places an order, the system needs to increase the user's points and send a notification – a typical e‑commerce use case. In a micro‑service architecture the order service, points service, and notification service are decoupled via RabbitMQ.

High performance thanks to Erlang’s concurrency.

Message persistence prevents loss on server failure.

Ack mechanism guarantees reliable processing.

Supports high‑availability clusters.

Flexible routing.

Implementation idea: after order creation, publish a message to EXCHANGE.ORDER_CREATE which is bound to two queues – QUEUE.ORDER_INCREASESCORE and QUEUE.ORDER_NOTIFY. Additional services can bind their own queues to the same exchange.

Order Creator (Publisher)

package com.robot.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

public class OrderCreator {
    private static final String EXCHANGE = "EXCHANGE.ORDER_CREATE";
    private static String msg = "create order success";

    public void createOrder() {
        System.out.println("Order placed, sending RabbitMQ message");
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.12.44");
        connectionFactory.setPort(56720);
        connectionFactory.setUsername("baibei");
        connectionFactory.setPassword("baibei");
        try {
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            boolean durable = true;
            String type = "topic";
            channel.exchangeDeclare(EXCHANGE, type, durable);
            String messageId = UUID.randomUUID().toString();
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .deliveryMode(2)
                .messageId(messageId)
                .build();
            String routingKey = "order_create";
            channel.basicPublish(EXCHANGE, routingKey, props, msg.getBytes("utf-8"));
            connection.close();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }
}

Score Service Consumer

package com.robot.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class IncreaseScoreConsumer implements Consumer {
    private Connection connection;
    private Channel channel;
    private static final String EXCHANGE = "EXCHANGE.ORDER_CREATE";
    private static final String QUEUENAME = "QUEUE.ORDER_INCREASESCORE";

    public void consume() {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.12.44");
        connectionFactory.setPort(56720);
        connectionFactory.setUsername("baibei");
        connectionFactory.setPassword("baibei");
        try {
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE, "topic", true);
            channel.queueDeclare(QUEUENAME, true, false, false, null);
            channel.queueBind(QUEUENAME, EXCHANGE, "order_create");
            channel.basicConsume(QUEUENAME, false, this);
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }

    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String msg = new String(body, "UTF-8");
        System.out.println("[Score Service] Received order message: " + msg + ", adding points...");
        channel.basicAck(envelope.getDeliveryTag(), false);
    }

    public void handleConsumeOk(String consumerTag) {}
    public void handleCancelOk(String consumerTag) {}
    public void handleCancel(String consumerTag) throws IOException {}
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {}
    public void handleRecoverOk(String consumerTag) {}
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {}
}

Notification Service Consumer

package com.robot.rabbitmq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class NotifyConsumer implements Consumer {
    private Connection connection;
    private Channel channel;
    private static final String EXCHANGE = "EXCHANGE.ORDER_CREATE";
    private static final String QUEUENAME = "QUEUE.ORDER_NOTIFY";

    public void consume() {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.12.44");
        connectionFactory.setPort(56720);
        connectionFactory.setUsername("baibei");
        connectionFactory.setPassword("baibei");
        try {
            connection = connectionFactory.newConnection();
            channel = connection.createChannel();
            channel.exchangeDeclare(EXCHANGE, "topic", true);
            channel.queueDeclare(QUEUENAME, true, false, false, null);
            channel.queueBind(QUEUENAME, EXCHANGE, "order_create");
            channel.basicConsume(QUEUENAME, false, this);
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
    }

    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String msg = new String(body, "UTF-8");
        System.out.println("[Notify Service] Received order message: " + msg + ", sending notification...");
        channel.basicAck(envelope.getDeliveryTag(), false);
    }

    public void handleConsumeOk(String consumerTag) {}
    public void handleCancelOk(String consumerTag) {}
    public void handleCancel(String consumerTag) throws IOException {}
    public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {}
    public void handleRecoverOk(String consumerTag) {}
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {}
}

Test Runner

package com.robot.rabbitmq;

public class Test {
    public static void main(String[] args) {
        IncreaseScoreConsumer increaseScoreConsumer = new IncreaseScoreConsumer();
        increaseScoreConsumer.consume();
        NotifyConsumer notifyConsumer = new NotifyConsumer();
        notifyConsumer.consume();
        OrderCreator orderCreator = new OrderCreator();
        for (int i = 0; i < 3; i++) {
            orderCreator.createOrder();
        }
    }
}

Sample Output

Order placed, sending RabbitMQ message
[Score Service] Received order message: create order success, adding points...
[Notify Service] Received order message: create order success, sending notification...
Order placed, sending RabbitMQ message
[Score Service] Received order message: create order success, adding points...
[Notify Service] Received order message: create order success, sending notification...
Order placed, sending RabbitMQ message
[Score Service] Received order message: create order success, adding points...
[Notify Service] Received order message: create order success, sending notification...

Original source: https://www.jianshu.com/p/2f55cd7a3e1c (author: 会跳舞的机器人)