马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。
您需要 登录 才可以下载或查看，没有账号？立即注册
x
一、消息大概丢失的场景
消息丢失大概发生在以下几个阶段：
- 生产者阶段：
- RabbitMQ 服务器阶段：
- 消费者阶段：
二、RabbitMQ 消息可靠性保障机制
RabbitMQ 提供了一系列机制，保证消息在各个阶段的可靠通报：
2.1 生产者阶段：消息确认机制（Publisher Confirms）
- 原理：
- 开启消息确认机制后，RabbitMQ 会在消息乐成到达队列后向生产者发送确认相应。
- 实现方式：
- 开启 publisher-confirms 模式，并在代码中监听确认相应。
- 应用：
- 如果 RabbitMQ 返回确认消息，表示消息已乐成到达队列。
- 如果未返回，生产者可以重试发送或记录失败日志。
2.2 RabbitMQ 服务器阶段：持久化机制
- 原理：
- 将队列和消息持久化到磁盘，纵然 RabbitMQ 重启，消息仍旧存在。
- 实现方式：
- 队列设置为持久化（durable）。
- 消息设置为持久化（deliveryMode=2）。
- 注意：
- 持久化会增长消息发送的耽误，需在性能和可靠性之间权衡。
2.3 消费者阶段：消息确认机制（Acknowledgments）
- 原理：
- 消费者在乐成处理消息后，向 RabbitMQ 返回 ACK，RabbitMQ 才会删除该消息。
- 实现方式：
- 利用手动确认模式（autoAck=false），确保消息被精确处理后再确认。
- 应用：
- 如果消费者未返回 ACK 或消费者挂掉，RabbitMQ 会将消息重新投递给其他消费者。
2.4 死信队列（Dead Letter Queue，DLQ）
- 原理：
- 将无法被消费或处理的消息投递到死信队列，以便后续分析和处理。
- 实现方式：
- 配置队列的 x-dead-letter-exchange 和 x-dead-letter-routing-key 属性。
- 应用：
- 常用于处理消费者拒绝的消息或消息超时未被消费的情况。
三、RabbitMQ 消息不丢失的实践
以下通过 Java 示例，展示怎样在生产者、RabbitMQ 服务器和消费者阶段分别实现消息的可靠性保障。
四、Java 示例：保证消息不丢失
4.1 环境准备
Maven 依赖：
- <dependencies>
- <dependency>
- <groupId>com.rabbitmq</groupId>
- <artifactId>amqp-client</artifactId>
- <version>5.17.0</version>
- </dependency>
- </dependencies>
- import com.rabbitmq.client.*;
- public class ReliableProducer {
- private static final String EXCHANGE_NAME = "reliable_exchange";
- private static final String QUEUE_NAME = "reliable_queue";
- private static final String ROUTING_KEY = "reliable_routing_key";
- 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.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true); // 持久化交换机
- channel.queueDeclare(QUEUE_NAME, true, false, false, null); // 持久化队列
- channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
- // 开启消息确认
- channel.confirmSelect();
- String message = "Hello, Reliable RabbitMQ!";
- AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
- .deliveryMode(2) // 消息持久化
- .build();
- // 发送消息
- channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props, message.getBytes());
- System.out.println("Message sent: " + message);
- // 等待确认
- if (channel.waitForConfirms()) {
- System.out.println("Message confirmed by RabbitMQ");
- } else {
- System.err.println("Message not confirmed, consider retrying.");
- }
- }
- }
- }
4.3 消费者：手动确认模式
- import com.rabbitmq.client.*;
- public class ReliableConsumer {
- private static final String QUEUE_NAME = "reliable_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()) {
- // 设置手动消息确认
- boolean autoAck = false;
- // 消费消息
- channel.basicConsume(QUEUE_NAME, autoAck, (consumerTag, delivery) -> {
- String message = new String(delivery.getBody(), "UTF-8");
- System.out.println("Received message: " + message);
- try {
- // 模拟消息处理
- processMessage(message);
- // 消息处理成功，返回 ACK
- channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
- System.out.println("Message acknowledged");
- } catch (Exception e) {
- // 消息处理失败，拒绝并重新投递
- channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
- System.err.println("Message requeued due to processing failure");
- }
- }, consumerTag -> {});
- }
- }
- private static void processMessage(String message) {
- // 模拟消息处理逻辑
- System.out.println("Processing message: " + message);
- }
- }
4.4 死信队列配置
- import com.rabbitmq.client.*;
- import java.util.HashMap;
- import java.util.Map;
- public class DLQConsumer {
- private static final String MAIN_QUEUE = "main_queue";
- private static final String DLX_EXCHANGE = "dlx_exchange";
- private static final String DLQ = "dlq_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.exchangeDeclare(DLX_EXCHANGE, BuiltinExchangeType.DIRECT, true);
- channel.queueDeclare(DLQ, true, false, false, null);
- channel.queueBind(DLQ, DLX_EXCHANGE, "dlx_routing_key");
- // 配置主队列与死信交换机绑定
- Map<String, Object> argsMap = new HashMap<>();
- argsMap.put("x-dead-letter-exchange", DLX_EXCHANGE);
- argsMap.put("x-dead-letter-routing-key", "dlx_routing_key");
- channel.queueDeclare(MAIN_QUEUE, true, false, false, argsMap);
- // 消费主队列消息
- channel.basicConsume(MAIN_QUEUE, false, (consumerTag, delivery) -> {
- String message = new String(delivery.getBody(), "UTF-8");
- System.out.println("Received message from main queue: " + message);
- // 模拟处理失败
- channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);
- System.out.println("Message sent to DLQ");
- }, consumerTag -> {});
- }
- }
- }
五、总结
通过上述机制和代码示例，RabbitMQ 可以有效保证消息不丢失：
- 生产者阶段：
- 利用消息确认机制（Publisher Confirms）和消息持久化。
- 服务器阶段：
- 消费者阶段：
- 死信队列：
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
