第十二章 RabbitMQ之失败消息处理策略

一、引言

Spring AMQP提供了消费者失败重试机制，在消费者出现异常时利用本地重试，而不是无限地requeue到mq。我们可以通过在application.yaml文件中添加配置来开启重试机制：

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: Wangzhexiao
    password: Wangzhexiao
    virtual-host: /hangzhou
    listener:
      simple:
        prefetch: 1
        acknowledge-mode: manual # none，关闭ack；manual，手动ack；auto：自动ack
        # 消费者重试机制配置
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初始的失败等待时长为1秒
          multiplier: 1 # 下次失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

在开启重试模式后，重试次数耗尽，如果消息依然失败，则需要有MessageRecoverer接口来处理，它包含三种不同的实现：

RejectAndDontRequeueRecoverer：重试耗尽后，直接reject，丢弃消息（默认方式）

ImmediateRequeueMessageRecoverer：重试耗尽后，返回nack，消息重新入队

RepublishMessageRecoverer：重试耗尽后，将失败消息投递到指定的交换机（推荐）

二、RepublishMessageRecoverer 实现

在实际项目的生产环境中，通过 RepublishMessageRecoverer 方式我们可以定义一个异常队列和交换机，来接收其他交换机队列转发的无法处理的异常消息。然后我们可以查看其中的异常消息并进行人工处理。

2.1. 实现步骤

1. 将失败处理策略改为RepublishMessageRecoverer

2. 定义接收失败消息的交换机、队列及其绑定关系

3. 定义RepublishMessageRecoverer

2.2. 实现代码

2.2.1. 异常交换机队列回收期配置类

package com.example.consumer;

import jakarta.annotation.Resource;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 异常交换机/队列/消息回收器配置类
 * ConditionalOnProperty 通过yml中的重试配置来选择该配置类是否启用
 */
@Configuration
@ConditionalOnProperty(prefix = "spring.rabbitmq.listener.simple.retry", name = "enabled", havingValue = "true")
public class ErrorConfig {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Bean
    Queue errorQueue() {
        return new Queue("error.queue");
    }

    @Bean
    DirectExchange errorExchange() {
        return new DirectExchange("error.direct");
    }

    @Bean
    Binding errorBind(Queue errorQueue, DirectExchange errorExchange) {
        return BindingBuilder.bind(errorQueue).to(errorExchange).with("error");
    }

    @Bean
    public MessageRecoverer messageRecoverer() {
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}

2.2.2. 常规交换机队列配置类

package com.example.consumer;

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 常规的RabbitMQ 交换机/队列绑定配置类
 */
@Configuration
public class RabbitMQConfig {

    @Bean
    Queue simpleQueue() {
        // 使用 QueueBuilder 创建一个持久化队列
        return QueueBuilder.durable("simple.queue").build();
    }
}

2.2.3. 消费者代码

package com.example.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 消费者
 */
@Slf4j
@Component
public class SimpleListener {

    @RabbitListener(queues = "simple.queue")
    public void listener1(String msg) throws Exception {
//        System.out.println("消费者1：人生是个不断攀登的过程【" + msg + "】");
        throw new Exception();
    }
}

2.2.4. 消费者yml配置

# 消费者application.yml配置
spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: Wangzhexiao
    password: Wangzhexiao
    virtual-host: /hangzhou
    listener:
      simple:
        prefetch: 1
        acknowledge-mode: auto # none，关闭ack；manual，手动ack；auto：自动ack
        # 消费者重试机制配置
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 1000ms # 初始的失败等待时长为1秒
          multiplier: 1 # 下次失败的等待时长倍数，下次等待时长 = multiplier * last-interval
          max-attempts: 3 # 最大重试次数
          stateless: true # true无状态；false有状态。如果业务中包含事务，这里改为false

2.2.5. 生产者代码

package com.example.publisher;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;

/**
 * 生产者
 */
@Slf4j
@SpringBootTest
class PublisherApplicationTests {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Test
    void test() {
        rabbitTemplate.convertAndSend("simple.queue", "只要学不死，就往死里学！");
    }
}

2.2.6. 生产者yml配置

# 生产者application.yml配置
spring:
  rabbitmq:
    # MQ连接配置
    host: 127.0.0.1
    port: 5672
    username: Wangzhexiao
    password: Wangzhexiao
    virtual-host: /hangzhou

2.2.7. 运行效果

最终效果是，我们在消费者的代码逻辑中会抛出异常，消息在反复投递消费失败后被重新入列到我们定义的异常交换机队列中：