RabbitMQ （三）消息重试

1 RabbitMQ自带的重试机制

1 示例代码

rabbitMQ为自带了消息重试机制：当消费者消费消息失败时，可以选择将消息重新“推送”给消费者，直至消息消费成功为止。

开启自带的重试机制，需要如下几个配置：

1 开启消费者手动应答机制，对应的springboot配置项：

spring.rabbitmq.listener.simple.acknowledge-mode=manual

2 消费异常时，设置消息重新入列

 boolean multiple = false; // 单条确认boolean requeue  = true; // 重新进入队列,谨慎设置！！！很容易导致死循环，cpu 100%channel.basicNack(tag, multiple, requeue);

以下是运行例子：

消费者代码如下：

package com.fmi110.rabbitmq;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicInteger;

/*** @author fmi110* @description 消息消费者* @date 2021/7/1 16:08*/
@Component
@Slf4j
public class RabbitConsumer {
AtomicInteger count = new AtomicInteger();
@RabbitListener(queues="my-queue")public void consumer1(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
log.info(">>>> consumer1 消费 tag = {},次数count={},消息内容 : {}",tag, count.incrementAndGet(),data);try {Thread.currentThread().sleep(1000);int i = 1/0;channel.basicAck(tag,true); // 确认消息消费成功} catch (Exception e) {log.error(">>>>消费异常，消息重新进入队列并消费");boolean multiple = false; // 单条确认boolean requeue  = true; // 重新进入队列,谨慎设置！！！channel.basicNack(tag, multiple, requeue);}}
@RabbitListener(queues="my-queue")public void consumer2(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{log.info(">>>> consumer2 消费 tag = {},次数count={},消息内容 : {}",tag, count.incrementAndGet(),data);
try {Thread.currentThread().sleep(1000);int i = 1/0;channel.basicAck(tag,true); // 确认消息消费成功} catch (Exception e) {log.error(">>>>消费异常，消息重新进入队列并消费");boolean multiple = false; // 单条确认boolean requeue  = true; channel.basicNack(tag, multiple, requeue);}}
}

这里模拟了两个消费者 consumer1、consumer2 ，并在逻辑中人为设置异常 int 1/0 , 在异常捕获中通过

 channel.basicNack(tag, false, true);

设置消息重新进入队列，最终推给消费者再次消费。运行结果如下：

日志里包含了几个信息点：

消费者每次只消费一条消息，因为我设置了 spring.rabbitmq.listener.simple.prefetch=1

消息推送使用的 round-robin 算法

rabbitMQ的消费方式有推和拉两种方式，springboot创建的消费者模式使用的推的方式消费 this.channel.basicConsume()

2 潜在问题

如运行日志所示，重进进入队列的消息，会在队列头部，直接再次推送给消费者消费，如果是因为代码逻辑问题，将会导致消息一直消费失败，导致死循环！！！

比较合理的做法是，重试一定次数消费后，如果仍然失败，则终止重试，将消费异常的消息保存，并上报异常，由人工介入处理。

2 结合spring-retry和死信队列实现消息重试

一个比较合理的重试机制如下：

消息消费出现异常时，借助springboot提供的重试机制进行重试

因为使用的spring-retry，所以方法中必须抛出异常，否则spring-retry不会被触发！！！
重试仍然失败时，消息转发到死信队列，死信队列的消费者记录并上报异常信息

要实现消息消费失败自动转发到死信队列，则rabbitmq在创建消息队列时，需要指定与之绑定的死信队列

完整的实例代码如下：

1 配置文件 application.properties：

这里注释掉了 spring.rabbitmq.listener.simple.acknowledge-mode=manual ，这样在消息消费失败时，会自动转到死信队列，如果开启手动确认机制，必须调用 chanel.basicNack(tag,false,false) 消息才会进入死信队列！！！

# 应用名称
spring.application.name=rabbitmq
server.port=8080
server.servlet.context-path=/

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
# 指定连接的虚拟主机，可以在rabbitMQ控制台查看对应的虚拟主机的名字
spring.rabbitmq.virtual-host=my_vhost
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin

spring.rabbitmq.listener.simple.prefetch=1

# 开启 publish-comfirm 机制和消息路由匹配失败退回机制
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.publisher-confirm-type=correlated
# 开启消费者应答 ack 机制
# spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 开启spring提供的retry
spring.rabbitmq.listener.simple.retry.enabled=true
spring.rabbitmq.listener.simple.retry.max-attempts=3
spring.rabbitmq.listener.simple.retry.initial-interval=3000

2 RabbitConfig

主要在程序启动时，做如下设置：

创建死信队列和死信交换器，并将死信队列绑定到死信交换器。
创建普通队列和普通交换器，并将普通队列绑定到普通交换器，同时将死信队列与普通队列关联，这样当消息消费失败时，消息会进入死信队列（使用了自动 ack模式）。

package com.fmi110.rabbitmq.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;


/*** @author fmi110* @description rabbitMQ 配置类* @date 2021/7/1 15:08*/
@Configuration
@Slf4j
public class RabbitConfig {
String dlQueueName  = "my-queue-dl"; // 普通队列名称String dlExchangeName = "my-exchange-dl"; // 死信交换器名称String dlRoutingKey   = "rabbit.test";
String queueName = "retry-queue";String exchangeName = "my-exchange"; // 普通交换器名称
/*** 创建死信队列** @return*/@Beanpublic Queue queueDL() {
return QueueBuilder.durable(dlQueueName) // 持久化队列.build();}
/*** 创建死信交换机** @return*/@Beanpublic TopicExchange exchangeDL() {return new TopicExchange(dlExchangeName, true, false);}
/*** 绑定操作*/@Beanpublic Binding bindQueueDL2ExchangeDL(Queue queueDL, TopicExchange exchangeDL) {log.info(">>>> 队列与交换器绑定");return BindingBuilder.bind(queueDL).to(exchangeDL).with(dlRoutingKey);}
/*** 创建持久化队列,同时绑定死信交换器** @return*/@Beanpublic Queue queue() {log.info(">>>> 创建队列 retry-queue");HashMap<String, Object> params = new HashMap<>();params.put("x-dead-letter-exchange", dlExchangeName);params.put("x-dead-letter-routing-key", dlRoutingKey);
return QueueBuilder.durable(queueName) // 持久化队列.withArguments(params) // 关联死信交换器.build();}

/*** 创建交换机** @return*/@Beanpublic TopicExchange exchange() {log.info(">>>> 创建交换器 my-exchange");boolean durable    = true; // 持久化boolean autoDelete = false; // 消费者全部解绑时不自动删除return new TopicExchange(exchangeName, durable, autoDelete);}
/*** 绑定队列到交换机** @param queue* @param exchange* @return*/@Beanpublic Binding bindQueue2Exchange(Queue queue, TopicExchange exchange) {log.info(">>>> 队列与交换器绑定");return BindingBuilder.bind(queue).to(exchange).with("rabbit.test");}

//    /**
//     * spring-retry重试机制：当重试次数达到最大，消息仍然消费失败时回调。
//     * 如果开启这个类，则死信队列失效，消息消费失败，即使配置了死信队列，消息也不会进入死信队列。
//     * 重试失败回调和死信队列只能二选一！！！spring 提供回调实现类有如下几个：
//     * RejectAndDontRequeueRecoverer ：消费失败，并且消息不再入列，spring默认使用。
//     * ImmediateRequeueMessageRecoverer ：将消息重新入列
//     * RepublishMessageRecoverer：转发消息到指定的队列，
//     * @return
//     */
//    @Bean
//    public MessageRecoverer messageRecoverer(){
//        return new MessageRecoverer() {
//            @Override
//            public void recover(Message message, Throwable cause) {
//                log.info(message.toString());
//                log.info("spring-retry重试次数达到最大，消息仍然失败的回调");
//                // TODO: 记录错误信息并上报
//            }
//        };
//    }
}

3 消息生产者 RabbitProducer

这里为了保证消息能确保消息发送，配置了 confirm 确认机制

package com.fmi110.rabbitmq;

import com.rabbitmq.client.AMQP;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/*** @author fmi110* @description 消息生产者* @date 2021/7/1 15:08*/
@Component
@Slf4j
public class RabbitProducer {@AutowiredRabbitTemplate rabbitTemplate;
/*** 1 设置 confirm 回调，消息发送到 exchange 时回调* 2 设置 return callback ,当路由规则无法匹配到消息队列时，回调** correlationData：消息发送时，传递的参数，里边只有一个id属性，标识消息用*/@PostConstructpublic void enableConfirmCallback(){// #1/*** 连接不上 exchange或exchange不存在时回调*/rabbitTemplate.setConfirmCallback((correlationData,ack,cause)->{if (!ack) {log.error("消息发送失败");// TODO 记录日志，发送通知等逻辑}});
// #2/*** 消息投递到队列失败时，才会回调该方法* message:发送的消息* exchange:消息发往的交换器的名称* routingKey:消息携带的路由关键字信息*/rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) ->{log.error("消息路由失败");// TODO 路由失败后续处理逻辑});}
public void send(String msg){String exchangeName = "my-exchange";// String routingKey   = "aaa.xxx";String routingKey   = "rabbit.test";rabbitTemplate.convertAndSend(exchangeName, routingKey, msg);}
}

4 消息消费者 RabbitConsumer

package com.fmi110.rabbitmq;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicInteger;


/*** @author fmi110* @description 消息消费者* @date 2021/7/1 16:08*/
@Component
@Slf4j
public class RabbitConsumer {
AtomicInteger count = new AtomicInteger();
/*** 普通队列消费者* @param data* @param channel* @param tag* @throws Exception*/@RabbitListener(queues="retry-queue")public void consumer(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{
log.info(">>>> consumer 消费 tag = {},次数count={},消息内容 : {}",tag, count.incrementAndGet(),data);// TODO 消息处理逻辑throw new RuntimeException("抛出异常，模拟消费失败，触发spring-retry");}
/*** 死信队列消费者* @param data* @param channel* @param tag* @throws Exception*/@RabbitListener(queues="my-queue-dl")public void consumeDL(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception{log.info(">>>> 死信队列消费 tag = {},消息内容 : {}",tag,data);
//        channel.basicNack(tag, false, false);}
}

5 Controller

用于触发发送消息

package com.fmi110.rabbitmq.controller;


import com.fmi110.rabbitmq.RabbitProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;

@RestController
public class TestController {@AutowiredRabbitProducer rabbitProducer;
@GetMapping("/test")public Object test() {
rabbitProducer.send("this is a message");
HashMap<String, Object> result = new HashMap<>();result.put("code", 0);result.put("msg", "success");return result;}
}

6 运行结果

运行日志如下:

: >>>> consumer 消费 tag = 1,次数count=1,消息内容 : this is a message
: >>>> consumer 消费 tag = 1,次数count=2,消息内容 : this is a message
: >>>> consumer 消费 tag = 1,次数count=3,消息内容 : this is a message
o.s.a.r.r.RejectAndDontRequeueRecoverer  : Retries exhausted for message
(Body:'this is a message' MessageProperties
[headers={spring_listener_return_correlation=2840e95b-8544-4ed8-b3ed-8ba02aee2729},
contentType=text/plain, contentEncoding=UTF-8, contentLength=0,
receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false,
receivedExchange=my-exchange, receivedRoutingKey=rabbit.test,
deliveryTag=1, consumerTag=amq.ctag-a5AZEb9AYpOzL6mQJQIvaQ,
consumerQueue=retry-queue])

...
Caused by: java.lang.RuntimeException: 抛出异常，模拟消费失败，触发spring-retryat com.fmi110.rabbitmq.RabbitConsumer.consumer(RabbitConsumer.java:36) ~[classes/:na]at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_181]
....

: >>>> 死信队列消费 tag = 1,消息内容 : this is a message

从日志可看出，普通队列的消费者一共消费了三次仍然失败，最后回调spring提供 RejectAndDontRequeueRecoverer ，然后消息进入死信队列被消费。

7 pom依赖

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><scope>compile</scope></dependency></dependencies>