【MQ】RabbitMq的可靠性保证

消息队列中的可靠性主要是分为三部分：

• 消息不丢失：确保消息从生产者发送到消费者消息不丢失
• 消息不重复：确保消息不被重复消费
• 消息顺序性：确保消费的顺序性

解决方案主要有以下几部分：

• 消息不丢失

• • 生产者确认机制
  • 持久化机制
  • 消费者确认机制
  • 高可用

• 消息不重复：

• • 消费者确认机制
  • 消息重试机制
  • 幂等性设计

• 消息顺序性

• • 单消费者模式
  • 消息编号

生产者确认机制

作用

确保生产者发送的消息成功到达mq，避免消息在传输过程中丢失

实现原理

1. 生产者会发送消息之后，mq会向生产者返回确认（ack）或未确认（nack）信号
2. 如果生产者收到ack，说明消息已经成功发送

配置方式

• 开启生产者确认模式

Channel channel = connection.createChannel();
channel.confirmSelect(); // 开启确认模式

• 处理确认和未确认

channel.addConfirmListener((sequenceNumber, multiple) -> {// 消息确认System.out.println("Message confirmed: " + sequenceNumber);
}, (sequenceNumber, multiple) -> {// 消息未确认System.out.println("Message not confirmed: " + sequenceNumber);
});

适用场景

对消息可靠性要求高的场景，如金融交易、订单处理等

持久化机制

作用

将消息和队列持久化到磁盘，防止mq重启或者崩溃时消息丢失

实现原理

1. 队列持久化：将队列中的数据保存在磁盘
2. 消息持久化：将消息内容保存到磁盘

配置方式

• 队列持久化

boolean durable = true;
channel.queueDeclare("my_queue", durable, false, false, null);

• 消息持久化

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2) // 2 表示持久化消息.build();
channel.basicPublish("", "my_queue", properties, message.getBytes());

适用场景

需要保证消息不丢失的场景，如日志存储、重要数据同步

消费者确认机制

作用

确保消费者成功处理消息之后，mq才会将消息从消息队列中删除，避免消息丢失

实现原理

1. 消费者处理完成消息后，手动发送ack信号
2. 如果消费者未发送ack，mq会将消息重新投递给其他消费者

配置方式

• 开启消费者确认

boolean autoAck = false; // 关闭自动确认
channel.basicConsume("my_queue", autoAck, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 处理消息System.out.println("Received: " + new String(body));// 手动发送 ackchannel.basicAck(envelope.getDeliveryTag(), false);}
});

适用场景

需要确保消息成功处理的场景，如订单处理、消息通知等

死信队列

作用

处理无法正常消费的消息（被拒绝或者过期），避免消息丢失

实现原理

1. 当消息无法被正常消费时，mq将其转发到死信队列中
2. 死信队列可以配置独立的交换机和路由规则

配置方式

• 配置死信队列

Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "my_dlx_exchange"); // 设置死信交换机
args.put("x-dead-letter-routing-key", "my_dlx_routing_key"); // 设置死信路由键
channel.queueDeclare("my_queue", true, false, false, args);

适用场景

需要处理异常消息的场景，如消息重试，失败信息分析等

消息重试机制

作用

当消息处理失败的时候，通过重试机制重新投递消息，确保消息最终被成功处理

实现原理

1. 消费者捕获异常，并拒绝消息，同时设置requeue=true
2. 使用死信队列和TTL实现延迟重试

配置方式

• 拒绝消息，重新入队

channel.basicConsume("my_queue", false, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {try {// 处理消息System.out.println("Received: " + new String(body));channel.basicAck(envelope.getDeliveryTag(), false);} catch (Exception e) {// 处理失败，拒绝消息并重新入队channel.basicNack(envelope.getDeliveryTag(), false, true);}}
});

适用场景

需要重试机制的场景，如网络抖动、依赖服务不可用等

高可用性

作用

通过集群和经销队里额，确保mq在节点出现故障仍然可以要运行，避免消息丢失。

镜像队列结构是一主多从，所有操作都是主节点完成，然后同步给镜像节 点，如果主节点宕机后，镜像节点会替代成新的主节点

实现原理

1. 集群：多个mq节点组成的集群，共享数据和队列状态
2. 镜像队列：将队列镜像到多个节点，确保队列的高可用性

配置方式

• 集群配置

rabbitmqctl join_cluster rabbit@node1

• 镜像队列配置

Map<String, Object> args = new HashMap<>();
args.put("x-ha-policy", "all"); // 镜像到所有节点
channel.queueDeclare("my_queue", true, false, false, args);

适用场景

适用于高可用的场景，如分布式系统、关键业务系统等

高可用性数据丢失怎么解决

在镜像队列主从同步的过程中，会出现在主从同步完成前，主节点就已经宕机，可能出现数据丢失。

我们可以通过仲裁队列来进行解决，和镜像队列一样，都是主从模式，支持主从数据同步但是不一样的点就是主从同步基于Raft协议，强一致性

并且使用起来也非常简单，不需要额外的配置，在声明队列的时候只要指定 这个是仲裁队列即可