RabbitMQ 可靠性投递

---

前言

RabbitMQ 是一个流行的消息队列系统，用于在分布式系统中传递消息。其中一个重要特性是其可靠性投递（Reliable Message Delivery），保证消息在队列和消费者之间可靠的传递和处理。

---

一、RabbitMQ自带机制

RabbitMQ 的架构图

从架构图中，我们可以发现，消息投递的关键步骤在于如下四点

• 1.生产者发送消息
• 2.消息路由机制
• 3.消息存储持久化机制
• 4.消费者消费消息

接下来我们一步一步进行分析

1、生产者发送消息

当网络中断或者节点不存在等，均有可能导致生产者发送消息失败，对于失败，可以通过如下机制进行处理

• 事务（Transactions）
• 发布确认（Publisher Confirms）

事务（Transactions）和发布确认（Publisher Confirms）是两种确保消息持久性和可靠性的方法。

注意

事务提供了一种全有或全无的机制，但通常不建议在生产环境中使用，因为它们会显著降低性能。发布确认则提供了更轻量级的解决方案，具有更高的性能和灵活性。

1.1、事务（Transactions）

事务的工作原理

• 事务机制确保一组消息的发送要么全部成功，要么全部失败。如果提交事务失败，所有消息都会回滚，确保数据一致性。

使用方法

• 启动事务：在信道上启动事务模式。
• 提交事务：成功时提交所有消息。
• 回滚事务：发生错误时回滚所有消息。

示例

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class RabbitMQTransaction {private final static String QUEUE_NAME = "transaction_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, true, false, false, null);try {// 启动事务模式channel.txSelect();for (int i = 0; i < 10; i++) {String message = "Message " + i;channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("Sent: " + message);}// 提交事务channel.txCommit();System.out.println("Transaction committed");} catch (Exception e) {System.out.println("Transaction failed: " + e.getMessage());// 回滚事务channel.txRollback();System.out.println("Transaction rolled back");}}}
}

1.2、发布确认（Publisher Confirms）

1.2.1、同步

相较于事务，发布确认（Publisher Confirms）是更为推荐的方式，因为它可以提供类似的可靠性并且具有更好的性能表现：

• 发布：使用 channel.confirmSelect() 启用发布确认模式。
• 回调：在发布消息后调用 channel.waitForConfirmsOrDie 方法等待确认，或捕获 Exception 进行错误处理。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConfirmCallback;public class RabbitMQPublishConfirm {private final static String QUEUE_NAME = "confirm_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 启用发布确认模式channel.confirmSelect();// 发布消息for (int i = 0; i < 10; i++) {String message = "Message " + i;channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("Sent: " + message);}// 确认所有发布的消息// 批量确认结果， ACK 如果是 Multiple=True， 代表 ACK 里面的 Delivery-Tag 之前的消息都被确认了// 比如 5 条消息可能只收到 1 个 ACK， 也可能收到 2 个（抓包才看得到）// 直到所有信息都发布， 只要有一个未被 Broker 确认就会 Exceptionchannel.waitForConfirmsOrDie(5000);System.out.println("All messages confirmed");} catch (Exception e) {System.err.println("Message publishing failed: " + e.getMessage());}}
}

1.2.2、异步

异步确认模式（Asynchronous Confirmations）在发布确认模式的基础上提供了更高效和灵活的消息确认机制。相比于同步发布确认，异步模式允许发布者继续发送消息而不必等待每条消息的确认。这种模式更适合高吞吐量的生产环境。以下是实现异步确认模式的详细步骤和示例代码：

• 建立连接和信道：与RabbitMQ服务器建立连接并创建信道。
• 启用发布确认模式：使用 channel.confirmSelect() 方法。
• 设置确认回调（Confirms Callback）：定义消息确认和未确认的回调方法。
  • ConfirmCallback: 用于确认成功的回调。
  • ConfirmListener: 用于设定成功和失败的回调。
• 发送消息：发布者继续发送消息，不需要等待确认。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConfirmCallback;public class AsyncConfirmPublisher {private static final String QUEUE_NAME = "async_confirm_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, true, false, false, null);// 启用发布确认模式channel.confirmSelect();// 设置确认和未确认的回调处理事件ConfirmCallback ackCallback = (deliveryTag, multiple) -> {System.out.println("Message ACKed with delivery tag: " + deliveryTag + ", multiple: " + multiple);};ConfirmCallback nackCallback = (deliveryTag, multiple) -> {System.err.println("Message NACKed with delivery tag: " + deliveryTag + ", multiple: " + multiple);};channel.addConfirmListener(ackCallback, nackCallback);// 发布消息for (int i = 0; i < 10; i++) {String message = "Message " + i;channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("Sent: " + message);}}}
}

2、消息路由机制

当 routingKey 错误，或者队列不存在的时候，就会出现无法路由，导致消息投递失败，解决方案

• 使用备份交换机（Alternate Exchanges）：如果消息无法路由到指定的交换机，RabbitMQ 可以将消息路由到一个备用交换机。
• 启用消息的确认回调（Return Callbacks）：当消息无法路由到任何队列时，可以使用回调函数捕获并处理这些未被路由的消息。
• 配置死信交换机（Dead Letter Exchanges）：当消息在队列中无法被消费或出现错误时，可将消息转发到死信交换机进行之后处理。

2.1、使用备份交换机（Alternate Exchanges）

备份交换机可以防止消息丢失，如果消息无法路由到主交换机，会被备份交换机存储。

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private static final String PRIMARY_EXCHANGE = "primary_exchange";private static final String PRIMARY_QUEUE = "primary_queue";private static final String ALTERNATE_EXCHANGE = "alternate_exchange";private static final String ALTERNATE_QUEUE = "alternate_queue";@Beanpublic ConnectionFactory connectionFactory() {// Configuration connection factory, assuming default settingsreturn new CachingConnectionFactory("localhost");}@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());rabbitTemplate.setMandatory(true);rabbitTemplate.setReturnCallback(returnCallback);return rabbitTemplate;}@Beanpublic DirectExchange primaryExchange() {return ExchangeBuilder.directExchange(PRIMARY_EXCHANGE).durable(true).alternate(ALTERNATE_EXCHANGE).build();}@Beanpublic DirectExchange alternateExchange() {return new DirectExchange(ALTERNATE_EXCHANGE);}@Beanpublic Queue primaryQueue() {return QueueBuilder.durable(PRIMARY_QUEUE).build();}@Beanpublic Queue alternateQueue() {return new Queue(ALTERNATE_QUEUE);}@Beanpublic Binding bindingPrimary() {return BindingBuilder.bind(primaryQueue()).to(primaryExchange()).with("primary_key");}@Beanpublic Binding bindingAlternate() {return BindingBuilder.bind(alternateQueue()).to(alternateExchange()).with("");}private final RabbitTemplate.ReturnCallback returnCallback = (message, replyCode, replyText,exchange, routingKey) -> {// Handle undeliverable messageSystem.err.printf("Message returned: %s, code: %d, text: %s, exchange: %s, routing key: %s%n",new String(message.getBody()), replyCode, replyText, exchange, routingKey);};
}

2.2、启用消息的确认回调（Return Callbacks）

启用消息的确认回调，当消息无法路由到任何队列时，可以使用回调函数捕获并处理这些未被路由的消息。

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private static final String PRIMARY_EXCHANGE = "primary_exchange";private static final String PRIMARY_QUEUE = "primary_queue";@Beanpublic ConnectionFactory connectionFactory() {return new CachingConnectionFactory("localhost");}@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());rabbitTemplate.setMandatory(true);rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 错误处理逻辑System.err.printf("Message returned: %s, code: %d, text: %s, exchange: %s, routing key: %s%n",new String(message.getBody()), replyCode, replyText, exchange, routingKey);});return rabbitTemplate;}@Beanpublic DirectExchange primaryExchange() {return ExchangeBuilder.directExchange(PRIMARY_EXCHANGE).durable(true).build();}@Beanpublic Queue primaryQueue() {return QueueBuilder.durable(PRIMARY_QUEUE).build();}@Beanpublic Binding bindingPrimary() {return BindingBuilder.bind(primaryQueue()).to(primaryExchange()).with("primary_key");}
}

2.3、配置死信交换机（Dead Letter Exchanges）

消息在队列中无法被消费或出现错误时，可将其转发到死信交换机进行特定处理。

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private static final String PRIMARY_EXCHANGE = "primary_exchange";private static final String PRIMARY_QUEUE = "primary_queue";private static final String DLX_EXCHANGE = "dlx_exchange";private static final String DLX_QUEUE = "dlx_queue";@Beanpublic ConnectionFactory connectionFactory() {return new CachingConnectionFactory("localhost");}@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());rabbitTemplate.setMandatory(true);rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 错误处理逻辑System.err.printf("Message returned: %s, code: %d, text: %s, exchange: %s, routing key: %s%n",new String(message.getBody()), replyCode, replyText, exchange, routingKey);});return rabbitTemplate;}@Beanpublic DirectExchange primaryExchange() {return new DirectExchange(PRIMARY_EXCHANGE);}@Beanpublic DirectExchange dlxExchange() {return new DirectExchange(DLX_EXCHANGE);}@Beanpublic Queue primaryQueue() {return QueueBuilder.durable(PRIMARY_QUEUE).withArgument("x-dead-letter-exchange", DLX_EXCHANGE).build();}@Beanpublic Queue dlxQueue() {return QueueBuilder.durable(DLX_QUEUE).build();}@Beanpublic Binding bindingPrimary() {return BindingBuilder.bind(primaryQueue()).to(primaryExchange()).with("primary_key");}@Beanpublic Binding bindingDLX() {return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("#");}
}

3、消息存储持久化机制

如果消息没有持久化，当RabbitMQ服务重启、节点故障等情况下就会丢失消息，解决方案

3.1、队列持久化（Durable Queue）

创建队列时将其设置为持久化

channel.queueDeclare("queue_name", durable=True)

3.2、消息持久化（Persistent Message）

发送消息时将其标记为持久化

channel.basicPublish(exchange='', routingKey='queue_name', body=msg, properties=pika.BasicProperties(delivery_mode=2,))

3.3、集群部署

保证节点的高可用

4、消费者消费消息

在使用 RabbitMQ 作为消息队列系统时，消费者确认机制（Consumer Acknowledge）对于确保消息可靠消费非常重要。当消费者从队列中接收到消息时，必须明确确认消息已被成功处理。这种机制不仅防止消息丢失，还避免了消息重复消费的问题。
RabbitMQ 提供了两种主要的消息确认机制：

• 手动确认（Manual Acknowledgement）：消费者显式地向 RabbitMQ 发送确认消息，表明消息已被成功处理。
• 自动确认（Automatic Acknowledgement）：RabbitMQ 在消息被发送给消费者后立即认为消息已经成功处理，无需等待显式确认。（默认）

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {private static final String EXCHANGE_NAME = "direct_exchange";private static final String QUEUE_NAME = "example_queue";@Beanpublic DirectExchange directExchange() {return new DirectExchange(EXCHANGE_NAME);}@Beanpublic Queue queue() {return new Queue(QUEUE_NAME, true);}@Beanpublic Binding binding(Queue queue, DirectExchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("routing_key");}@Beanpublic SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(QUEUE_NAME);container.setMessageListener(listenerAdapter);// 自动确认 AcknowledgeMode.AUTO// 手动确认 AcknowledgeMode.MANUALcontainer.setAcknowledgeMode(org.springframework.amqp.core.AcknowledgeMode.MANUAL); return container;}@Beanpublic MessageListenerAdapter listenerAdapter(Consumer consumer) {return new MessageListenerAdapter(consumer, "consumeMessage");}
}

如果是手动确认，消费者处理消息的时候实现手动确认机制：

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;@Component
public class Consumer implements ChannelAwareMessageListener {@Overridepublic void onMessage(Message message, Channel channel) throws Exception {try {// 处理消息逻辑String body = new String(message.getBody());System.out.println("Received message: " + body);// 手动确认消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 消息处理失败，拒绝消息并重新入队channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);System.err.println("Failed to process message: " + e.getMessage());}}
}

二、业务保证

1、最终一致性

• 消息持久化处理：在发出消息之前，将消息存储到持久化存储设备，如数据库。 并使用消息状态进行追踪，每个状态表示消息的处理进度。
• 补偿机制：根据超时和失败记录进行消息补偿。
  • 怎么触发重发（定时任务）
  • 多久触发一次（参考业务属性）
  • 触发多少次（参考业务属性）
• 幂等性：确保消费者在处理消息时是幂等的，即多次处理不会导致副作用。

通过以上的方案结合，保证消息的最终一致性

2、监控和告警

建立监控系统，及时发现和处理异常情况，然后及时作出相应，增加客户的良好体验
例如：
使用 Prometheus 和 Grafana 对消息队列和数据库进行监控，及时发现并告警异常情况。
也可以结合业务属性，自己创建一套简单的监控系统。