前言

消费者在处理消息的过程中可能会发生异常，那么此时此刻该如何处理这个异常的消息呢？

RabbitMQ有两个方法 channel.basicNack 或 channel.basicReject能够让消息重新回到原队列中，这样子可以实现重试。但是如果第二次消费又发生了异常，一直消费一直异常。由于没有明确重试次数，会造就无限重试，这是一个致命的问题。

本文就来使用 spring-rabbit中自带的 retry功能来解决这个问题。

编码

依赖

starter-amqp中包含了 spring-rabbit。

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

配置

需要进行简单的配置即可开启

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: auto  # 消息确认方式，其有三种配置方式，分别是none、manual(手动ack) 和auto(自动ack) 默认auto
        retry:
          enabled: true  #监听重试是否可用
          max-attempts: 5   #最大重试次数 默认为3
          initial-interval: 2000  # 传递消息的时间间隔 默认1s
    host: 47.105.*
    port: 5672
    virtual-host: /*-1
    username: *
    password: *
mq:
  queueBinding:
    queue: prod_queue_pay
    exchange:
      name: exchang_prod_pay
      type: topic
    key: prod_pay

创建业务队列、交换机

@Configuration
public class RabbitConfig {
    @Value("${mq.queueBinding.queue}")
    private String queueName;
    @Value("${mq.queueBinding.exchange.name}")
    private String exchangeName;
    @Value("${mq.queueBinding.key}")
    private String key;
    /**
     * 业务队列
     * @return
     */
    @Bean
    public Queue payQueue(){
        Map<String,Object> params = new HashMap<>();
        return QueueBuilder.durable(queueName).withArguments(params).build();
    }
    
    @Bean
    public TopicExchange payTopicExchange(){
        return new TopicExchange(exchangeName,true,false);
    }
    //队列与交换机进行绑定
    @Bean
    public Binding BindingPayQueueAndPayTopicExchange(Queue payQueue, TopicExchange payTopicExchange){
        return BindingBuilder.bind(payQueue).to(payTopicExchange).with(key);
    }
}

生产者

@Component
@Slf4j
public class RabbitSender {
    @Value("${mq.queueBinding.exchange.name}")
    private String exchangeName;
    @Value("${mq.queueBinding.key}")
    private String key;
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void send(String msg){
        log.info("RabbitSender.send() msg = {}",msg);
        // 将消息发送给业务交换机
        rabbitTemplate.convertAndSend(exchangeName,key,msg);
    }
}

消费者

@Component
@Slf4j
public class RabbitReceiver {
  int count  = 0;
    //测试重试
    @RabbitListener(queues = "${mq.queueBinding.queue}")
    public void infoConsumption(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        log.info("重试次数 = {}",count++);
        int i = 10 /0;
        channel.basicAck(tag,false);
    }
}

提供对外方法

@Controller
public class TestController {
    @Autowired
    private RabbitSender rabbitSender;
    @GetMapping
    public void test(@RequestParam String msg){
        rabbitSender.send(msg);
    }
}

然后调用接口： http://localhost:8080/?msg=红红火火，消息会被发送到 prod_queue_pay这个队列。然后重试 5 次。每次重试时间间隔为2秒，与配置相符。

注意：重试并不是 RabbitMQ 重新发送了消息到了队列，仅仅是消费者内部进行了重试，换句话说就是重试跟mq没有任何关系。上述消费者代码不能添加try{}catch(){}，一旦捕获了异常，在自动 ack 模式下，就相当于消息正确处理了，消息直接被确认掉了，不会触发重试的。 当然并不是说不能添加 try{}catch(){}，而是不能将异常给处理了。可以如下写：

 @RabbitListener(queues = "${mq.queueBinding.queue}")
    public void infoConsumption(String data, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        log.info("重试次数 = {}",count++);
        try {
            // 处理主要业务
            int i = 10 /0;
        } catch (Exception e) {
            // 处理业务失败，还要进行其他操作，比如记录失败原因
            log.info("记录失败原因 ====>");
            throw new RuntimeException("手动抛出");
        }
        channel.basicAck(tag,false);
    }

MessageReCoverer

上面的例子在测试中我们还发现了一个问题，就是经过 5 次重试以后，控制台输出了一个异常的堆栈日志，然后队列中的数据也被 ack 掉了（因为我配置了 auto，自动ack模式）。如果你配置的是 manual(手动ack)，结果就会如下：

五次重试后，消费处于一个未被确认的状态。因为需要你手动 ack！下次服务重启的时候，会继续消费这条消息。

首先我们先来看一下这个异常日志是什么：

org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Retry Policy Exhausted

出现上述异常的原因是因为在构建 SimpleRabbitListenerContainerFactoryConfigurer类时使用了 MessageRecoverer接口，这个接口有一个 recover方法，用来实现重试完成之后对消息的处理，源码如下：

public final class SimpleRabbitListenerContainerFactoryConfigurer
		extends AbstractRabbitListenerContainerFactoryConfigurer<SimpleRabbitListenerContainerFactory> {
	@Override
	public void configure(SimpleRabbitListenerContainerFactory factory, ConnectionFactory connectionFactory) {
		PropertyMapper map = PropertyMapper.get();
		RabbitProperties.SimpleContainer config = getRabbitProperties().getListener().getSimple();
		configure(factory, connectionFactory, config);  >> 1
		map.from(config::getConcurrency).whenNonNull().to(factory::setConcurrentConsumers);
		map.from(config::getMaxConcurrency).whenNonNull().to(factory::setMaxConcurrentConsumers);
		map.from(config::getBatchSize).whenNonNull().to(factory::setBatchSize);
	}
}

注意标记为 >> 1 的 configure方法

ListenerRetry retryConfig = configuration.getRetry();
if (retryConfig.isEnabled()) {
    RetryInterceptorBuilder<?, ?> builder = (retryConfig.isStateless()) ? RetryInterceptorBuilder.stateless()
        : RetryInterceptorBuilder.stateful();
    RetryTemplate retryTemplate = new RetryTemplateFactory(this.retryTemplateCustomizers)
        .createRetryTemplate(retryConfig, RabbitRetryTemplateCustomizer.Target.LISTENER);
    builder.retryOperations(retryTemplate);
    MessageRecoverer recoverer = (this.messageRecoverer != null) ? this.messageRecoverer
        : new RejectAndDontRequeueRecoverer(); //<1>
    builder.recoverer(recoverer);
    factory.setAdviceChain(builder.build());

注意看 <1>处的代码，默认使用的是 RejectAndDontRequeueRecoverer类，这个类已经出现过了，注意笔者前面的那几张图。根据类的名字我们就可以看出来该实现类的作用就是 拒绝并且不会将消息重新发回队列，也就是说，重试之后如果还没有成功，就认为该消息没救了，放弃它了。我们可以看一下这个实现类的具体内容：

public class RejectAndDontRequeueRecoverer implements MessageRecoverer {
    protected Log logger = LogFactory.getLog(RejectAndDontRequeueRecoverer.class); // NOSONAR protected
    @Override
    public void recover(Message message, Throwable cause) {
        if (this.logger.isWarnEnabled()) {
                this.logger.warn("Retries exhausted for message " + message, cause);
        }
        throw new ListenerExecutionFailedException("Retry Policy Exhausted",
                                new AmqpRejectAndDontRequeueException(cause), message);
    }
}

上述源码给出了异常的来源。

MessageRecoverer 接口中就一个 recover方法，回调已消费但所有重试尝试失败的消息。

重写了 recover方法的有四个类， MessageBatchRecoverer这个不在本文范围内。而 RejectAndDontRequeueRecoverer的功能已经看到过了，毕竟是默认的。那还有另外两个实现类，分别是 RepublishMessageRecoverer和 ImmediateRequeueMessageRecoverer，意思大意分别是 重新发布消息和立即重新返回原队列，下面我们分别测试一下这两个实现类的效果。

RepublishMessageRecoverer

将消息重新发送到指定队列。先创建一个队列，然后与交换机绑定进行绑定,绑定之后设置 MessageRecoverer。在 RabbitConfig类中增加代码。跟 死信队列看起来差不多。

@Autowired
private RabbitTemplate rabbitTemplate;
private static String errorTopicExchange = "error-topic-exchange";
private static String errorQueue = "error-queue";
private static String errorRoutingKey = "error-routing-key";
//创建异常交换机
@Bean
public TopicExchange errorTopicExchange(){
    return new TopicExchange(errorTopicExchange,true,false);
}
//创建异常队列
@Bean
public Queue errorQueue(){
    return new Queue(errorQueue,true);
}
//队列与交换机进行绑定
@Bean
public Binding BindingErrorQueueAndExchange(Queue errorQueue,TopicExchange errorTopicExchange){
    return BindingBuilder.bind(errorQueue).to(errorTopicExchange).with(errorRoutingKey);
}
//设置MessageRecoverer
@Bean
public MessageRecoverer messageRecoverer(){
    //AmqpTemplate和RabbitTemplate都可以
    return new RepublishMessageRecoverer(rabbitTemplate,errorTopicExchange,errorRoutingKey);
}

启动服务，重新调用接口，查看结果：

通过控制台可以看到，使用了我们所配置的 RepublishMessageRecoverer，并且消息重试 5 次以后直接以新的 routingKey发送到了配置的交换机中，此时再查看监控页面，可以看原始队列中已经没有消息了，但是配置的异常队列中存在了一条消息。

ImmediateRequeueMessageRecoverer

使用 ImmediateRequeueMessageRecoverer，重试失败的消息会立马回到原队列中。修改 messageRecoverer方法

@Bean
public MessageRecoverer messageRecoverer(){
    return new ImmediateRequeueMessageRecoverer();
}

启动服务，重新调用接口，查看结果：

重试5次之后，返回队列，然后再被消费，继续重试5次，周而复始直到消息被正常消费为止。

总结

通过上面的测试，对于重试之后仍然异常的消息，可以采用 RepublishMessageRecoverer，将消息发送到其他的队列中，再专门针对新的队列进行处理。

死信队列

除了可以采用上述 RepublishMessageRecoverer，还可以采用死信队列的方式处理重试失败的消息。这也是我们常用的方式。

创建死信交换机、死信队列以及两者的绑定

继续在 RabbitConfig中增加配置

private static String dlTopicExchange = "dl-topic-exchange";
private static String dlQueue = "dl-queue";
private static String dlRoutingKey = "dl-routing-key";
//创建交换机
@Bean
public TopicExchange dlTopicExchange(){
    return new TopicExchange(dlTopicExchange,true,false);
}
//创建队列
@Bean
public Queue dlQueue(){
    return new Queue(dlQueue,true);
}
//队列与交换机进行绑定
@Bean
public Binding BindingDlQueueAndExchange(Queue dlQueue, TopicExchange dlTopicExchange){
    return BindingBuilder.bind(dlQueue).to(dlTopicExchange).with(dlRoutingKey);
}

死信交换机的定义和普通交换机的定义完全相同，队列绑定死信交换机与绑定普通交换机的方式完全相同，死信交换机就是一个普通的交换机，只是换了一个叫法而已，没有什么特殊之处。

修改配置

修改业务队列的配置，还有将之前提供的 MessageReCoverer进行注释，不然死信交换机不会生效，会以我们所配置的 MessageReCoverer为主。

    /**
     * 绑定死信交换机需要给队列设置如下两个参数     
     * 业务队列
     * @return
     */
    @Bean
    public Queue payQueue(){
        Map<String,Object> params = new HashMap<>();
        //声明当前队列绑定的死信交换机
        params.put("x-dead-letter-exchange",dlTopicExchange);
        //声明当前队列的死信路由键
        params.put("x-dead-letter-routing-key",dlRoutingKey);
        return QueueBuilder.durable(queueName).withArguments(params).build();
    }
    
     //设置MessageRecoverer
    //@Bean
    //public MessageRecoverer messageRecoverer() {
        //AmqpTemplate和RabbitTemplate都可以
        //return new ImmediateRequeueMessageRecoverer();
   // }

启动服务之前，需要将之前创建的队列进行删除，因为本次队列的配置有所改动。启动成功，可以看到同时创建了业务队列以及死信队列、业务交换机、死信交换机。

在业务队列上出现了 DLX 以及 DLK 的标识，代表已经绑定了死信交换机以及死信路由键，此时调用生产者发送消息，消费者在重试5次后，由于 MessageCover默认的实现类是 RejectAndDontRequeueRecoverer，又因为业务队列绑定了死信队列，因此消息会从业务队列中删除，同时发送到死信队列中。

最后

最近我整理了整套 《JAVA核心知识点总结》,说实话 ,作为一名Java程序员,不论你需不需要面试都应该好好看下这份资料。拿到手总是不亏的~我的不少粉丝也因此拿到腾讯字节快手等公司的Offer

进 【 Java进阶之路群】，找管理员获取哦-！

ITPub博客

RabbitMQ进阶——RabbitMQ 重试机制，看这篇就够了

前言

编码

依赖

配置