How to Prevent Message Loss in RabbitMQ: Producer, Server, and Consumer Strategies

In RabbitMQ, message loss can occur during three stages: when the producer sends a message, while the message is transferred inside the RabbitMQ server, and when the consumer receives the message. To guarantee no loss, measures must be taken at each stage.

1. Ensure the Producer Sends Messages Successfully

Enable Confirm Mode : The producer calls channel.confirmSelect() to enable confirm mode. After sending a message, RabbitMQ returns a confirmation indicating whether the message reached the server.

Transaction Mechanism : The producer can start a transaction with channel.txSelect(). After publishing, commit the transaction; if an exception occurs, roll back. Note that transactions significantly impact performance and are not recommended for high‑concurrency scenarios.

public class Producer {
    private static final String QUEUE_NAME = "test_queue";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.confirmSelect();
            // Add confirm listener
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws Exception {
                    System.out.println("Message confirmed, deliveryTag: " + deliveryTag + (multiple ? "，multiple" : ""));
                }
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws Exception {
                    System.out.println("Message NOT confirmed, deliveryTag: " + deliveryTag + (multiple ? "，multiple" : ""));
                }
            });
            String message = "Hello, RabbitMQ!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
        }
    }
}

public class ProducerWithTx {
    private static final String QUEUE_NAME = "test_queue";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.txSelect();
            String message = "Hello, RabbitMQ!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
            channel.txCommit();
        } catch (Exception e) {
            if (channel != null) {
                try { channel.txRollback(); } catch (Exception ex) { ex.printStackTrace(); }
            }
            e.printStackTrace();
        }
    }
}

2. Ensure Server‑Side Message Persistence

Exchange Persistence : Declare the exchange with the durable flag, e.g., channel.exchangeDeclare(EXCHANGE_NAME, "direct", true); Queue Persistence : Declare the queue as durable, e.g., channel.queueDeclare(queueName, true, false, false, null); Message Persistence : Publish with MessageProperties.PERSISTENT_TEXT_PLAIN so the message is written to disk, e.g.,

channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));

public class PersistentMessageSender {
    private static final String EXCHANGE_NAME = "persistent_exchange";
    private static final String QUEUE_NAME = "persistent_queue";
    private static final String ROUTING_KEY = "persistent_key";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
            String message = "这是一条持久化消息";
            channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,
                    message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

3. Ensure the Consumer Receives Messages Correctly

Manual Acknowledgment : Consume with channel.basicConsume(queueName, false, ...). After processing, call channel.basicAck(deliveryTag, false). If the consumer crashes without ack, RabbitMQ will re‑queue the message for another consumer.

Consumer Retry Mechanism : On failure, log and retry; after a maximum number of retries, route the message to a dead‑letter exchange/queue.

public class Consumer {
    private static final String QUEUE_NAME = "test_queue";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME, autoAck, "myConsumerTag", false, false, null,
                new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String message = new String(body, "UTF-8");
                        System.out.println("收到消息: '" + message + "'");
                        try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); }
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                });
        }
    }
}

public class MessageConsumerWithRetryAndDLX {
    private static final String QUEUE_NAME = "main_queue";
    private static final String DLX_EXCHANGE_NAME = "dlx_exchange";
    private static final String DLX_QUEUE_NAME = "dlx_queue";
    private static final int MAX_RETRIES = 3;
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(DLX_EXCHANGE_NAME, "direct", true);
            channel.queueDeclare(DLX_QUEUE_NAME, true, false, false, null);
            channel.queueBind(DLX_QUEUE_NAME, DLX_EXCHANGE_NAME, "dlx_routing_key");
            Map<String, Object> args = new HashMap<>();
            args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
            args.put("x-dead-letter-routing-key", "dlx_routing_key");
            channel.queueDeclare(QUEUE_NAME, true, false, false, args);
            boolean autoAck = false;
            channel.basicConsume(QUEUE_NAME, autoAck, "myConsumerTag", false, false, null,
                new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                                               AMQP.BasicProperties properties, byte[] body) throws IOException {
                        String message = new String(body, "UTF-8");
                        System.out.println(" [x] Received '" + message + "'");
                        int retryCount = getRetryCount(properties);
                        boolean success = processMessage(message, retryCount);
                        if (success) {
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        } else {
                            if (retryCount < MAX_RETRIES) {
                                AMQP.BasicProperties newProps = properties.builder()
                                        .headers(getUpdatedHeaders(retryCount))
                                        .build();
                                channel.basicPublish("", envelope.getRoutingKey(), newProps, body);
                                channel.basicAck(envelope.getDeliveryTag(), false);
                            } else {
                                channel.basicPublish(DLX_EXCHANGE_NAME, "dlx_routing_key", properties, body);
                                channel.basicAck(envelope.getDeliveryTag(), false);
                                System.out.println("达到最大重试次数，消息转入死信队列: " + message);
                            }
                        }
                    }
                });
        }
    }
    private static int getRetryCount(AMQP.BasicProperties properties) {
        if (properties.getHeaders() == null || !properties.getHeaders().containsKey("retryCount")) {
            return 0;
        }
        return (int) properties.getHeaders().get("retryCount");
    }
    private static boolean processMessage(String message, int retryCount) {
        try {
            if (message.contains("error")) {
                throw new RuntimeException("模拟处理失败");
            }
            System.out.println("消息处理成功: " + message);
            return true;
        } catch (Exception e) {
            System.out.println("消息处理失败: " + e);
            return false;
        }
    }
    private static Map<String, Object> getUpdatedHeaders(int retryCount) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("retryCount", retryCount + 1);
        return headers;
    }
}