Rabbitmq如何保证消息可靠性的全面解析_资讯

Rabbitmq如何保证消息可靠性的全面解析

创始人

2025-05-31 17:03:37

0次

Rabbitmq如何保证消息可靠性的全面解析

- Rabbitmq保证消息的可靠性主要针对以下几种途径体现
- 1、ReturnCallback
- 2、ConfirmCallback
- 3、消息的持久化
- 4、失败策略——重新投递到新的交换机
- 5、死信队列
- 6、消息过期或者队列过期
- 7、延迟队列
- - 1、安装延迟队列插件
  - 2、DelayExchange原理
  - 3、声明延迟交换机
  - 4、发送延迟消息
- 8、惰性队列
- - 1、什么是惰性队列
  - 2、创建惰性队列
  - 3、总结

Rabbitmq保证消息的可靠性主要针对以下几种途径体现

生产者发送消息到交换机、生产者直接发送消息到队列、交换机投递消息到队列、交换机队列消息持久化问题、消费者消费消息情况。通过ack、nack确认消息是否发送成功或者消费失败，通过本地重试机制、失败策略避免循环消息重入队到失败的过程，并衍生死信交换机、延迟队列、惰性队列的使用场景/font>

1、生产者是否成功发送消息到交换机（RabbitMQ提供了publisher confirm机制，ConfirmCallback返回ack或nack，correlated异步、simple同步）

2、交换机是否成功将消息投递到队列（RabbitMQ提供了publisher return机制，通过全局ReturnCallback去处理）

3、生产者是否成功将消息直接发送给队列（RabbitMQ提供了publisher confirm机制）

4、交换机、队列是否持久化（SpringAmqp默认持久化）

5、消息的持久化（设置消息的属性（MessageProperties），指定delivery-mode）

6、消费者是否成功消费消息（返回ack）

manual 手动ack 不推荐
auto 根据Spring Aop检测是否出现异常，若出现异常则返回nack，没有异常返回ack
none 消费者获取到消息后立即返回ack，RabbitMQ立刻删除消息

7、消息失败的本地重试，消费者确认机制中使用了auto，返回nack后消息会被重新投递到队列中，此时java客户端依旧监听消息，抛出异常，消息又被重新投递到队列中，为避免重复的循环动作，需要开启本地重试，达到本地重试次数后走失败策略

RejectAndDontRequeueRecoverer：重试耗尽后，消息标记reject，丢弃消息，默认就是这种方式
ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队
RepushMessageRecoverer：重试耗尽后，将消息重新投递到指定的失败交换机

8、死信交换机，对于标记为死信的消息会发送到死信交换机，死信交换机投递到死信队列，有专门监听死信队列的消费者

标记为死信的情况：

消息被消费者reject或nack
消息过期
队列满了消息投递失败

9、消息过期或者队列过期标记为死信，交由死信交换机处理

10、延迟队列：消息过期或队列过期交由死信队列处理，达到延迟队列的目的，RabbitMQ原生也支持延迟队列的效果

11、惰性队列：生产者发送消息的速度大于消费者消费消息的速度，时间一长队列堆积满消息上限后，交换机投递到队列失败的消息会被标记为死信，交由死信交换机处理，可以通过惰性队列解决消息堆积的问题

spring:rabbitmq:host: 192.168.79.129 # rabbitMQ的ip地址port: 5672 # 端口username: rootpassword: rootvirtual-host: /publisher-confirm-type: correlated  # 异步回调  生产者发送消息的回调  ConfirmCallbackpublisher-returns: true  # 交换机投递消息到队列失败、消息直接发送到队列失败 ReturnCallbacktemplate:mandatory: true listener:simple:prefetch: 1acknowledge-mode: auto  # 由spring aop检测是否抛出异常retry:enabled: true # 开启消费者失败本地重试机制initial-interval: 1000 # 初次失败等待时长为1秒multiplier: 3 # 下次失败等待时长倍数max-attempts: 3 # 最大重试次数stateless: true # true代表无状态  false代表有状态 如果当前包含事务，必须改为false

1、ReturnCallback

交换机投递消息到队列失败、生产者直接投递消息到队列失败

全局唯一

@Configuration
@Slf4j
public class CommonConfig implements ApplicationContextAware {@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {// 获取bean单例RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);// 设置ReturnCallbackrabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {// 记录日志log.info("消息发送失败，应答码:{},原因:{},交换机:{},路由键:{},消息:{}",replyCode, replyText, exchange, routingKey, message.toString());// 消息重发});// rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {//   @Override//   public void returnedMessage(Message message, int i, String s, String s1, String s2) {//    //   }// });}
}

2、ConfirmCallback

消息投递到交换机的回调机制

@Test
public void testSendMessageSimpleQueue() {// 消息体String message = "Hello,spring amqp";// 设置唯一消息idCorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 添加callbackcorrelationData.getFuture().addCallback(result -> {if (result.isAck()){// acklog.info("消息投递到交换机成功！消息id：{}",correlationData.getId());}else {// nacklog.error("消息投递到交换机失败！消息id：{}",correlationData.getId());}}, ex -> {// 记录日志log.error("消息发送失败：{}",ex.getMessage());});// 发送消息rabbitTemplate.convertAndSend("simple.direct.zs", "zs.*", message, correlationData);log.info("消息发送完成");
}

3、消息的持久化

@Test
public void testDurableMessage() {// 1、准备消息Message message = MessageBuilder.withBody("hello,zs".getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build();// 2、发送消息rabbitTemplate.convertAndSend("topic.exchange", "zs.*", message);
}public enum MessageDeliveryMode {NON_PERSISTENT, // 持久化PERSISTENT;  // 非持久化
}

4、失败策略——重新投递到新的交换机

声明交换机、队列、绑定队列到交换机上

@Bean
public DirectExchange errorMessageExchange(){return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){return new Queue("error.queue");
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

创建RepushMessageRecoverer的bean到IOC容器中

// 异常消息处理器
@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate) {return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

5、死信队列

当普通队列里的消息被消费者标记为reject（本地重试后还是失败走RejectAndDontRequeueRecoverer）, nack (本地重试后失败走ImmediateRequeueMessageRecoverer)，消息过期，或者队列满了，通过普通队列绑定的死信交换机，死信交换机下绑定的死信队列，将消息重新投递到死信队列，交由专门处理死信消息的消费者模块去处理

队列配置了dead-letter-exchange属性，指定了一个交换机，那么队列中的死信就会投递到这个交换机中，而这个交换机称为死信交换机（Dead Letter Exchange，检查DLX）

队列将死信投递给死信交换机时，必须知道两个信息：

死信交换机名称
死信交换机与死信队列绑定的RoutingKey

这样才能确保投递的消息能到达死信交换机，并且正确的路由到死信队列。

定义普通队列配置dead-letter-exchange属性，创建死信交换机、死信队列、将死信队列绑定到死信交换机

// 声明普通的 simple.queue队列，并且为其指定死信交换机：dl.direct
@Bean
public Queue simpleQueue2(){return QueueBuilder.durable("simple.queue") // 指定队列名称，并持久化.deadLetterExchange("dl.direct") // 指定死信交换机.build();
}
// 声明死信交换机 dl.direct
@Bean
public DirectExchange dlExchange(){return new DirectExchange("dl.direct", true, false);
}
// 声明存储死信的队列 dl.queue
@Bean
public Queue dlQueue(){return new Queue("dl.queue", true);
}
// 将死信队列 与 死信交换机绑定
@Bean
public Binding dlBinding(){return BindingBuilder.bind(dlQueue()).to(dlExchange()).with("simple");
}

死信队列的使用场景：

如果队列绑定了死信交换机，死信会投递到死信交换机中；
利用死信交换机手机所有消费者处理失败的消息（死信），交由人工处理，进一步提高消息队列的可靠性，但是也会提高系统的复杂性

6、消息过期或者队列过期

一个队列中的消息如果超时未消费，则会变为死信，超时分为两种情况：

消息所在的队列设置了超时时间
消息本身设置了超时时间

创建死信交换机、死信队列的消费者

@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "dl.ttl.queue", durable = "true"),exchange = @Exchange(name = "dl.ttl.direct"),key = "ttl"
))
public void listenDlQueue(String msg){log.info("接收到 dl.ttl.queue的延迟消息：{}", msg);
}

创建带过期时间的队列

@Bean
public Queue ttlQueue(){return QueueBuilder.durable("ttl.queue") // 指定队列名称，并持久化.ttl(10000) // 设置队列的超时时间，10秒.deadLetterExchange("dl.ttl.direct") // 指定死信交换机.build();
}

创建交换机绑定带过期时间的队列

@Bean
public DirectExchange ttlExchange(){return new DirectExchange("ttl.direct");
}
@Bean
public Binding ttlBinding(){return BindingBuilder.bind(ttlQueue()).to(ttlExchange()).with("ttl");
}

消费者发送不带过期时间的消息

@Test
public void testTTLQueue() {// 创建消息String message = "hello, ttl queue";// 消息ID，需要封装到CorrelationData中CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 发送消息rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);// 记录日志log.debug("发送消息成功");
}

消费者发送带过期时间的消息

@Test
public void testTTLMsg() {// 创建消息Message message = MessageBuilder.withBody("hello, ttl message".getBytes(StandardCharsets.UTF_8)).setExpiration("5000").build();// 消息ID，需要封装到CorrelationData中CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());// 发送消息rabbitTemplate.convertAndSend("ttl.direct", "ttl", message, correlationData);log.debug("发送消息成功");
}

7、延迟队列

1、安装延迟队列插件

利用TTL结合死信交换机，我们实现了消息发出后，消费者延迟收到消息的效果。这种消息模式就称为延迟队列（Delay Queue）模式。

延迟队列的使用场景包括：

延迟发送短信
用户下单，如果用户在15 分钟内未支付，则自动取消
预约工作会议，20分钟后自动通知所有参会人员

安转RabbitMQ 延迟队列插件：DelayExchange

执行命令开启插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

2、DelayExchange原理

DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时，流程如下：

接收消息
判断消息是否具备x-delay属性
如果有x-delay属性，说明是延迟消息，持久化到硬盘，读取x-delay值，作为延迟时间
返回routing not found结果给消息发送者
x-delay时间到期后，重新投递消息到指定队列

发送延迟消息到延迟交换机后，会立刻被ReturnCallback捕获消息发送失败异常，是因为延迟交换机发现消息头携带x-delay属性，则把消息持久化到硬盘后，等延迟时间到在投递到延迟队列，交由监听延迟队列的消费者去消费

3、声明延迟交换机

声明一个交换机，交换机的类型可以是任意类型，只需要设定delayed属性为true即可，然后声明队列与其绑定即可。

  @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "delay.queue", durable = "true"),exchange = @Exchange(name = "delay.direct", delayed = "true"),key = "delay"))public void listenDelayedQueue(String msg) {logger.info("接收到delay.queue的延迟消息：{}", msg);HashMap hashMap = JSONUtil.toBean(msg, HashMap.class);if (ObjectUtil.equal(hashMap.get("delayType"), MessageConstant.DELAY_TYPE_ACTIVITY)) {// 业务处理logger.info("listenDelayedQueue update activity result:{}", result);} else if (ObjectUtil.equal(hashMap.get("delayType"), DELAY_TYPE_TOPIC)) {// 业务处理logger.info("listenDelayedQueue update topic result:{}", result);}}

4、发送延迟消息

Map map = new HashMap<>();
map.put("delayType", MessageConstant.DELAY_TYPE_TOPIC);Message message = MessageBuilder.withBody(JSONUtil.toJsonStr(map).getBytes(StandardCharsets.UTF_8)).setHeader("x-delay", delayTime) // 设置延迟时间，单位ms.build();
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend("delay.direct", "delay", message, correlationData);
logger.info("delayType:{},消息发送成功！", MessageConstant.DELAY_TYPE_TOPIC);

消息发送到延迟交换机以后，在官方交换机的基础上功能的改造，理论上交换机应该立即转发消息到队列，它不具备存储功能，但是延迟交换机帮我们把消息存储起来，因此消息没有被转发路由到队列，所以就报错了，失败原因就是NO_ROUTE，消息如果没有被发送到队列，则会报NO_ROUTE，但是事实上消息并没有错，它只不过被暂存在交换机中，等过了ttl时间后就发送到消息队列里边receivedDelay=5000 延迟接收5s中

23:35:07:499  INFO 13741 --- [nectionFactory1] cn.mq.config.CommonConfig         : 消息发送失败，应答码:312,原因:NO_ROUTE,交换机:dl.zs.exchange,路由键:dl.zs,消息:(Body:'[B@7d928713(byte[25])' MessageProperties [headers={}, contentType=application/octet-stream, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, receivedDelay=5000, deliveryTag=0])

8、惰性队列

解决消息堆积有两种思路：

增加更多消费者，提高消费速度。也就是我们之前说的work queue模式
消费者内开启线程池加快消息处理速度
扩大队列容积，提高堆积上限

要提升队列容积，把消息保存在内存中显然是不行的。

1、什么是惰性队列

从RabbitMQ的3.6.0版本开始，就增加了Lazy Queues的概念，也就是惰性队列。惰性队列的特征如下：

接收到消息后直接存入磁盘而非内存避免触发内存预警，但是带来一定读写延迟，增加的多余IO操作）
消费者要消费消息时才会从磁盘中读取并加载到内存
支持数百万条的消息存储

2、创建惰性队列

命令行的方式设置队列的惰性属性
```
rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues 
```
命令解读：
- rabbitmqctl ：RabbitMQ的命令行工具
- set_policy ：添加一个策略
- Lazy ：策略名称，可以自定义
- “^lazy-queue$”` ：用正则表达式匹配队列的名字
- '{"queue-mode":"lazy"}' ：设置队列模式为lazy模式
- --apply-to queues ：策略的作用对象，是所有的队列

基于@Bean声明lazy_queue

@Bean
public Queue lazyQueue(){return QueueBuilder.durable("lazy.queue").lazy()  // 开启x-queue-mode为lazy.bulid();
}

基于@RabbitListener声明LazyQueue

@RabbitListener(queuesToDeclare = @Queue(name = "lazy.queue",durable = "true",arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String name){}

3、总结

消息堆积问题的解决方案？

队列上绑定多个消费者，提高消息速度
使用惰性多列，通过硬盘持久化保存更多的消息

惰性队列的优点有哪些？

基于磁盘存储，成本低，消息上限高
没有间隙性的page-out，性能比较稳定，普通队列有间隙性的page-out

惰性队列的缺点有哪些？

增加系统复杂度
基于磁盘存储，消息时效性会下降
性能受限于磁盘的IO

以上便是Rabbitmq如何保证消息可靠性的全面解析的全部内容，如有解读不当，欢迎在评论区指出！

词库加载错误:未能找到文件“E:\highferrum_mysql\Configuration\Dict_Stopwords.txt”。

上一篇：差速巡线机器人设计-良好（80+）的报告-2023

下一篇：Linux syslog 日志服务

Rabbitmq如何保证消息可靠性的全面解析

Rabbitmq如何保证消息可靠性的全面解析

Rabbitmq保证消息的可靠性主要针对以下几种途径体现

1、ReturnCallback

2、ConfirmCallback

3、消息的持久化

4、失败策略——重新投递到新的交换机

5、死信队列

6、消息过期或者队列过期

7、延迟队列

1、安装延迟队列插件

2、DelayExchange原理

3、声明延迟交换机

4、发送延迟消息

8、惰性队列

1、什么是惰性队列

2、创建惰性队列

3、总结

相关内容

热门资讯