app开发定制的基本使用、ACKapp开发定制确认机制这里就不赘述了,app开发定制这里主要是想实现一个应用场景:
app开发定制消息消费失败后重试至多三次,app开发定制仍失败则加入死信队列
一、重试机制
app开发定制首先说一下RabbitMQapp开发定制的消息重试机制,顾名思义,app开发定制就是消息消费失败后进行重试,重试机制的触发条件是消费者显式的抛出异常,这个很类似@Transactional,如果没有显式地抛出异常或者try catch起来没有手动回滚,事务是不会回滚的。
以下代码可以触发重试机制
还有一种情况就是消息被拒绝后重新加入,比如basic.reject和basic.nack,并且requeue = true,但是个人认为这个不算是触发了重试机制,这个是重新进入到了消息队列然后重新被消费,并且也不会触发我们重试机制的配置(如重试间隔、最大重试次数等等)。
重试机制是默认开启的,但是如果没有重试机制相关的配置会导致消息一直无间隔的重试,直到消费成功,所以要使用重试机制一定要有相关配置。
- spring:
- rabbitmq:
- host: 127.0.0.1
- port: 5672
- virtual-host: mq-test
- username: ********
- password: ********
- listener:
- simple:
- # ACK模式(none,auto,manual,默认为auto)
- acknowledge-mode: auto
- # 开启重试
- retry:
- # 是否开启重试机制
- enabled: true
- # 最大重试次数
- max-attempts: 5
- # 重试间隔(ms)
- initial-interval: 5000
二、死信队列
说到死信队列,首先需要知道什么是死信
死信就是消息在特定场景下的一种表现形式,这些场景包括:
- 消息被拒绝(basic.reject / basic.nack),并且requeue = false
- 消息的 TTL 过期时
- 消息队列达到最大长度
- 达到最大重试限制
消息在这些场景中时,被称为死信。
死信队列就是用于储存死信的消息队列,在死信队列中,有且只有死信构成,不会存在其余类型的消息。死信队列也是一个普通队列,也可以被消费者消费,区别在于业务队列需要绑定在死信队列上,才能正常地把死信发送到死信队列上。
业务队列绑定死信队列
三、重试+死信的实现
文章开头的场景有两种方案可以实现
方案一:使用自动ACK + RabbitMQ重试机制
方案二:使用手动ACK + 手动重试机制
3.1 自动ACK + RabbitMQ重试机制
配置
- spring:
- rabbitmq:
- host: 127.0.0.1
- port: 5672
- virtual-host: mq-test
- username: ********
- password: ********
- listener:
- simple:
- # ACK模式(默认为auto)
- acknowledge-mode: auto
- # 开启重试
- retry:
- enabled: true
- max-attempts: 5
- initial-interval: 5000
消费者
- @RabbitListener(queues = RabbitMqConfig.USER_ADD_QUEUE, concurrency = "10")
- public void userAddReceiver(String data, Message message, Channel channel) throws Exception {
- UserVo vo = OBJECT_MAPPER.readValue(data, UserVo.class);
- boolean success = messageHandle(vo);
- // 通过业务控制是否消费成功,消费失败则抛出异常触发重试
- if (!success) {
- log.error("消费失败");
- throw new Exception("消息消费失败");
- }
- }
需要说明的是,上述的方法一定要开启自动ACK,才会在到达最大重试上限后发送到死信队列,而且在重试过程中会独占当前线程,如果是单线程的消费者会导致其他消息阻塞,直至重试完成,所以可以使用@RabbitListener上的concurrency属性来控制并发数量。
3.2 手动ACK + 手动重试
配置
- spring:
- rabbitmq:
- host: 127.0.0.1
- port: 5672
- virtual-host: mq-test
- username: ********
- password: ********
- listener:
- simple:
- # ACK模式(默认为auto)
- acknowledge-mode: manual
需要说明的是,如果是手动ACK配置了重试机制,在抛出异常的时候仍会触发重试,但是达到重试上限之后,会永远处于Unacked状态,不会进入到死信队列,必须要手动拒绝才可以进入死信队列,所以说这里不用配置重试机制而是采用手动重试的方式
消费者
- /**
- * 消息最大重试次数
- */
- private static final int MAX_RETRIES = 3;
-
- /**
- * 重试间隔(秒)
- */
- private static final long RETRY_INTERVAL = 5;
-
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
- @RabbitListener(queues = RabbitMqConfig.USER_ADD_QUEUE, concurrency = "10")
- public void userAddReceiver(String data, Message message, Channel channel) throws IOException, InterruptedException {
- UserVo vo = OBJECT_MAPPER.readValue(data, UserVo.class);
- // 重试次数
- int retryCount = 0;
- boolean success = false;
- // 消费失败并且重试次数<=重试上限次数
- while (!success && retryCount < MAX_RETRIES) {
- retryCount++;
- // 具体业务逻辑
- success = messageHandle(vo);
- // 如果失败则重试
- if (!success) {
- String errorTip = "第" + retryCount + "次消费失败" +
- ((retryCount < 3) ? "," + RETRY_INTERVAL + "s后重试" : ",进入死信队列");
- log.error(errorTip);
- Thread.sleep(RETRY_INTERVAL * 1000);
- }
- }
- if (success) {
- // 消费成功,确认
- channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
- log.info("创建订单数据消费成功");
- } else {
- // 重试多次之后仍失败,进入死信队列
- channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
- log.info("创建订单数据消费失败");
- }
- }
总结:两种方案都可以达到我们的预期效果,相比起来方案一会更加的方便简洁,方案二的可控性更高