SpringBoot4系列09 - 使用消息队列RabbitMQ

上一篇讲缓存,结尾留了个尾巴:缓存解决的是「读得快」,系统里另一半问题「通知得可靠」还空着,这一篇来补。场景还是订单。订单支付成功之后要发短信、要推邮件、要给积分系统记账,这几件事没有一件是下单主链路必须等着的,也没有一件是挂了能装作没看见的。

这个坑我早年在一家电商公司结结实实踩过一回。支付成功的方法里挨个调用短信服务、邮件服务、积分服务,功能是对的,上线第一周就出事了。短信服务商那边抖了三十秒,支付接口跟着超时了三十秒,用户盯着转圈的进度条以为支付失败,又付了一次。投诉进来我翻日志,支付服务自己一行错误都没有,它在等一个跟它不相干的下游。后来我把通知改成异步:订单主链路只做一件事,把「支付成功」这件事喊一嗓子,谁关心谁去听。喊完就返回,短信晚到十秒用户无感;短信服务挂了,消息在队列里排队,等它缓过来接着发。主链路的生死从此不跟下游绑在一起。

有人会问,为什么不用更轻的进程内事件总线。我踩过的坑给的答案是:进程内的方案解决不了「下游服务挂了」的问题,订阅者和发布者在同一个 JVM 里,进程一挂大家一起没,进程内的异步只是把超时换成了丢事件。消息队列的价值就是把这件事搬到进程外,broker 的持久化替你扛住下游的宕机窗口,服务重启的间隙消息照样在。

https://static.xiongneng.me/rabbitmq-message-lifecycle-20260922005534.png

干这件事的基础设施,这一篇我选 RabbitMQ。不是因为它新,恰恰是因为它老,老到所有的坑前人都踩过了,老到 Spring 生态里那套 spring-amqp 的抽象成熟得几乎不用操心协议层。说实话我看重的就是这份成熟。Spring Boot 4 这边 starter 极简化,自动配置独立成模块,yml 开关照常好使。整篇里有三处细节我打算单独展开:一个在消息转换器里,一个在重试机制里,还有一个是发送方确认的定制口子。4.1.1 的制品我翻过一遍,要点先列在这儿,后面各自展开。

配套工程在 springboot4-rabbitmq 目录,场景是订单事件的发布与消费,拓扑里带完整的死信兜底。六个测试用例我全连着真实 RabbitMQ 4.3.6 跑,broker 在我树莓派的 Docker 里,不是内嵌的内存替身。超时、重试、死信这些时序敏感的行为,只有真 broker 才作数。

工程很小,一个 REST 服务加两个监听器。POST /api/orders 创建订单,POST /api/orders/{orderNo}/pay 支付并发布 ORDER_PAID 事件,GET /api/notify/state 是观测口,吐出 published、consumed、retried、deadLettered 四个计数和最后一次发送方确认的结果。为了做实验我还埋了两个后门:一个能对指定订单注入一次消费故障,一个能发布一条永远消费失败的 POISON 事件。观测归观测,注入归注入,计数才干净。

4.1.1 里 AMQP 的几个要点

starter 极简化

先看依赖。spring-boot-starter-amqp 的 POM 我拆开数过,就两层

1
2
3
4
5
6
7
8
spring-boot-starter-amqp
├── spring-boot-starter
└── spring-boot-amqp            ← 自动配置本体
    ├── spring-boot
    ├── spring-messaging
    ├── spring-rabbit           ← Spring AMQP 的实现主体(4.1.1)
    │   └── amqp-client         ← RabbitMQ 的 Java 客户端(5.30.0)
    └── spring-boot-transaction

跟缓存篇的 starter-cache 一个套路:starter 本身不带任何中间件客户端,客户端和实现都收进 spring-boot-amqp 模块里。BOM 里登记的版本我逐个数了一遍:spring-amqp 4.1.1、amqp-client 5.30.0、rabbit-stream-client 1.6.0。这些数字不用背,需要的时候去 BOM 里查就行,我是写这篇的时候特意对过一遍,怕文章里的版本和仓库里的对不上。

https://static.xiongneng.me/rabbitmq-starter-anatomy-20260922005534.png

还有一个跟 Redis 篇不一样的点我单独说。Redis starter 也不带序列化器,但 Redis 的 JSON 序列化要显式处理类型信息;AMQP 这边协议本身就带 __TypeId__ 头,类型映射是框架的事,要操心的只剩「信任哪些包」。这个区别讲转换器的时候我会展开。

自动配置在 amqp 模块

包名规律跟 MongoDB、Redis、缓存三篇完全一致,RabbitMQ 的自动配置在独立模块里,包名是

1
org.springframework.boot.amqp.autoconfigure

这个模块 47 个类,自动配置注册文件里有三条记录,RabbitAutoConfiguration、健康检查、metrics 各一条。触发条件是 @ConditionalOnClass({RabbitTemplate.class, Channel.class}),classpath 上有 spring-rabbit 就装配;broker 连不上只影响运行,不影响启动,应用照常起来。这个行为我特意把 broker 停掉试过,启动日志干干净净。类名是 RabbitAutoConfiguration,检索资料按这个名字走就行。

这个自动配置类我翻了内容,装配的东西不少:连接工厂、RabbitTemplate、RabbitAdmin、监听容器工厂,一条链全齐。连接工厂默认就是带缓存的实现,常规用量不必另配;要调连接参数,同样走定制口子,思路跟后面的 RabbitTemplateCustomizer 一致。整条链唯一需要我动手的就是那一个转换器 Bean。

属性前缀 spring.rabbitmq

前缀我从 RabbitProperties 的字节码常量池里确认过,是 spring.rabbitmq。对比 Redis 的 spring.data.redis.*(多了个 data)和缓存的 spring.cache.*,AMQP 属于前缀原样保留的那一类。常用的几组属性我直接贴工程里这份

属性不配会怎样我也试过:host 不给就落到默认的 localhost,应用照样启动,发消息连不上才报错。所以连不上 broker 的问题都发生在运行期,排查的时候别光盯着启动日志。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
spring:
  rabbitmq:
    host: 192.168.1.97
    port: 5672
    username: admin
    password: admin123456
    virtual-host: /
    publisher-confirm-type: correlated   # 发送方确认
    publisher-returns: true              # 路由失败回执
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 500ms
          multiplier: 1.5

publisher-confirm-type 和 publisher-returns 这两个发送方可靠性开关都在,listener.simple.retry.* 这一组重试配置照用。底层重试机制是另一回事,它跟你想的可能不太一样,放在核心代码里细说。

这份 yaml 我按组拆开讲。host/port/username/password/virtual-host 是连接五件套,broker 在我树莓派上,virtual-host 用默认的根路径。publisher-confirm-type: correlated 让 broker 对每条消息回执确认,correlated 模式下回执带关联数据,能把确认结果跟具体那次发送对上号。publisher-returns: true 打开路由失败回执。listener.simple.retry.* 四个参数凑成一个完整的重试策略:开关、次数上限、初始间隔、退避倍率,少一个都不完整。

消息转换器,Jackson 2/3 双轨

spring-amqp 4.1.1 的 jar 我解开数过,org.springframework.amqp.support.converter 包里两套 Jackson 转换器并存

Jackson 2(旧,com.fasterxml)Jackson 3(新,tools.jackson)
Jackson2JsonMessageConverterJacksonJsonMessageConverter
Jackson2XmlMessageConverterJacksonXmlMessageConverter
DefaultJackson2JavaTypeMapperDefaultJacksonJavaTypeMapper

跟 Redis 篇序列化器的双轨一个形态:旧的留着,新的用 Jackson 3 的 tools.jackson.databind.ObjectMapper。新转换器的签名我在字节码里确认过,构造方法可以直接传 JsonMapper,也可以传一个可变参数的信任包列表。

web 工程自带 Jackson 3,所以这一篇我的结论很干脆,**新工程用 JacksonJsonMessageConverter,别用 Jackson2JsonMessageConverter。**后者照样能跑,但会把 com.fasterxml 的依赖拽进来,两套 Jackson 同 classpath 的日子,能避就避。

选型的判断依据我展开一下。__TypeId__ 头的映射行为两套转换器都一样,差别在底下的 ObjectMapper:Jackson 3 的 ObjectMapper 在 Boot 4 的 web 栈里就是默认那个,消息体序列化出来的 JSON 风格和 HTTP 接口返回的完全一致,排查问题的时候两边可以对着看。这是我坚持选 Jackson 3 转换器最实际的理由,不只是为了避开重复依赖。

AMQP 有专属测试 starter

缓存篇的规律在 AMQP 这里也成立。spring-boot-starter-amqp-test 在 4.1.1 的 BOM 里登记着,组成是 starter-amqp 加 spring-boot-starter-test 加 spring-rabbit-test,一行引入。这一篇的测试就用的它,省了我自己凑依赖的工夫。

依赖和配置

pom 全家福,Web 和校验照旧,AMQP 一个 starter,测试两个

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webmvc</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 专属测试 starter = starter-amqp + starter-test + spring-rabbit-test -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webmvc-test</artifactId>
    <scope>test</scope>
</dependency>

没有 Jackson 依赖。JSON 转换器用的 Jackson 3 由 web starter 顺路带进来,web 工程里 AMQP 的 JSON 化是零额外依赖的。我第一次列依赖清单的时候还犹豫要不要单独加 jackson,翻了依赖树发现早就有了,白犹豫一场。

核心代码

拓扑声明,Bean 写好,剩下的交给 RabbitAdmin

消息队列的第一道坎不是代码,是概念。交换机、队列、绑定,三个东西各管各的。我的记法:交换机是邮局分拣机,队列是信箱,绑定写的是「什么样的信进哪个信箱」。生产者永远只面对交换机,不需要知道信箱在哪。

这套拓扑在 Spring 里全部用 Bean 声明

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
@Configuration
public class RabbitTopologyConfig {

    public static final String ORDER_EXCHANGE = "sb4.order.exchange";
    public static final String QUEUE_NOTIFY = "sb4.order.notify";
    public static final String DLX_EXCHANGE = "sb4.order.dlx";
    public static final String QUEUE_DEAD = "sb4.order.dead";
    public static final String RK_ORDER_CREATED = "order.created";
    public static final String RK_ORDER_PAID = "order.paid";
    public static final String RK_ORDER_DEAD = "order.dead";

    @Bean
    public TopicExchange orderExchange() {
        return ExchangeBuilder.topicExchange(ORDER_EXCHANGE).durable(true).build();
    }

    @Bean
    public Queue notifyQueue() {
        return QueueBuilder.durable(QUEUE_NOTIFY)
                .deadLetterExchange(DLX_EXCHANGE)      // 挂死信参数的关键一行
                .deadLetterRoutingKey(RK_ORDER_DEAD)
                .build();
    }

    @Bean
    public Binding notifyBinding() {
        // order.* 同时匹配 order.created 与 order.paid
        return BindingBuilder.bind(notifyQueue()).to(orderExchange()).with("order.*");
    }

    // 死信交换机、死信队列、绑定同理,DirectExchange + order.dead
}

https://static.xiongneng.me/rabbitmq-topology-20260922005534.png

这些声明不需要手动执行。自动配置好的 RabbitAdmin 会在连接建立之后,把容器里所有的交换机、队列、绑定 Bean 统一向 broker 声明一遍。声明是幂等的,已经存在的不会重建,我重启过好几轮,从没见过重复声明报错。拓扑跟着应用走,换一个环境部署,拓扑自动就位,挺省心。

我第一次用的时候犯过嘀咕:应用起来之前,broker 上是不是得先手工把队列建好?不用。开发环境里我随时可以把 broker 上的队列删干净,应用一连上,交换机、队列、绑定全体重新就位。拓扑声明因此成了代码的一部分,它跟着 jar 走,不跟着环境走,换机器部署省掉了手工建队列的仪式,也不会出现「我本地有这个队列,线上没有」的悬案。

业务队列上挂的 x-dead-letter-exchange 参数是死信机制的地基,QueueBuilder 的 deadLetterExchange 方法就是给它准备的便捷入口。这一行的意思是,凡是进了这个队列之后被「拒收且不回队」的消息,broker 自动转发到指定的死信交换机去。这个语义先记住,后面讲重试的时候它是最后一环。

routing key 的设计我多说两句。业务队列绑的是 order.*,一个通配符同时接住 order.created 和 order.paid,以后要加 order.refunded,事件照发,绑定不用动。死信那边是 DirectExchange,routing key 写死 order.dead,一一对应,不存在歧义。两个交换机两种类型,各取所需:业务侧要扩展性,死信侧要确定性。

拓扑 Bean 我全声明成 durable,broker 重启之后拓扑还在。这跟消息持久化是两码事,消息要不要落盘由发送端的投递模式决定,spring-rabbit 发消息默认就是持久化的,两边凑齐,重启树莓派上的 broker 我验证过,队列和消息都还在。

消息转换器,一个 Bean,两端生效

发送和消费都要 JSON,机制藏在 RabbitAutoConfiguration 内部的 RabbitTemplateConfiguration:它拿着一个 ObjectProvider<MessageConverter>,只要容器里存在唯一的 MessageConverter Bean,就自动把这个转换器 set 进 RabbitTemplate,监听容器工厂同样处理。

所以整个配置就一个 Bean,机制上发送端和消费端一起生效

1
2
3
4
5
6
7
8
9
@Configuration
public class RabbitClientConfig {

    @Bean
    public MessageConverter jacksonJsonMessageConverter() {
        // 信任包前缀:只允许反序列化自己工程的类型(对应 __TypeId__ 头)
        return new JacksonJsonMessageConverter("com.xncoding.");
    }
}

没有手动 new RabbitTemplate,没有在 @RabbitListener 上指定 containerFactory,两端全是自动的。发送时 OrderEvent 对象被序列化成 JSON 字节写进消息体,同时挂上 __TypeId__ 头;消费时容器拿着这个头反查类型,反序列化好再把对象递给监听方法。监听方法签名里写的就是领域对象,不是原始字节,这是我特别喜欢这套抽象的地方。

OrderEvent 里就几个字段,订单号、事件类型、事件 ID、时间戳,都是字符串和整型,Jackson 序列化毫无悬念。比较容易疏忽的是字段演进:异步链路两端的部署是错开的,消费端可能还跑着老结构。加字段没问题,改字段语义要慎重,这是异步化之后必须养成的习惯。

那个信任包参数我多说一句。__TypeId__ 头里写的是全限定类名,消费端会照着它反序列化。默认情况下这个映射是放行的,意味着谁往队列里塞一条 __TypeId__ 指向恶意类的消息,理论上是能被反序列化的。构造参数里传 "com.xncoding." 之后,只有这个前缀下的类型能通过,攻击面就收住了。成本是一个单词,收益是一类漏洞,怎么算都划算。

发送方确认,RabbitTemplateCustomizer 这个口子

yml 里开了 publisher-confirm-type: correlated 之后,broker 会为每条消息回执确认。回调往哪挂,这里我闹过笑话:给 RabbitTemplate 写个 @Bean 自己 new 一个,等于把自动配置的模板整个顶掉,确认回调悄悄就失灵了。正确姿势是自动配置留的定制口子

1
2
3
4
5
6
7
8
9
@Bean
public RabbitTemplateCustomizer reliabilityCustomizer() {
    return template -> {
        template.setConfirmCallback((correlationData, ack, cause) ->
                lastConfirmResult = ack ? "ACK" : "NACK: " + cause);
        template.setReturnsCallback(returned ->
                lastConfirmResult = "RETURNED: " + returned.getReplyText());
    };
}

自动配置在创建 RabbitTemplate 之后,会把容器里所有的 RabbitTemplateCustomizer 依次执行一遍,只改关心的属性,其余默认值一个不动。这个类在自动配置包 org.springframework.boot.amqp.autoconfigure 里。这里还有个坑:IDE 会提示 org.springframework.amqp.rabbit.core 里有一个同名的类,import 到那个,编译直接报找不到符号,自动配置不认它。两个同名类分属两个包,这个形态在缓存篇的 RedisCacheConfiguration 已经见过一次,第二次撞上我就没再犹豫。

confirm 和 return 的分工记一句话就够,**confirm 管「消息到没到 broker」,return 管「到了 broker 但路由不到任何队列」。**两个都开了,发送端可靠性才算配齐。

这段代码里的 lastConfirmResult 是个简单的字符串变量,demo 里专门用来观测确认结果,生产上更常见的做法是打日志加监控埋点。形态不重要,重要的是两个回调都挂在定制口子里,不碰自动配置的模板,升级 Boot 也不用跟着改。

return 回调平时安静得很,它只在消息到了 broker 却路由不到任何队列的时候才说话。我把 routing key 改成一个不存在的值试了一回,RETURNED 立刻出现在统计面板里,replyText 把原因写得明明白白。这类问题要没有 return 回调,消息就悄无声息地消失在 broker 里,日志里什么都查不到,这也是我坚持两个开关都要开的原因。

消费端一个 @RabbitListener 就够

消费端比发送端还省事,一个注解一个方法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
@RabbitListener(queues = RabbitTopologyConfig.QUEUE_NOTIFY)
public void onOrderEvent(OrderEvent event) {
    notificationService.onDeliveryAttempt(event);
    // ... 业务校验,失败抛异常
    doNotify(event);
    notificationService.onConsumed(event);
}

@RabbitListener(queues = RabbitTopologyConfig.QUEUE_DEAD)
public void onDeadLetter(OrderEvent event) {
    notificationService.onDeadLettered(event);
    log.error("[死信兜底] 事件进入死信队列,待人工处理: orderNo={}", event.getOrderNo());
}

两个监听方法,一个管正常业务,一个管死信兜底。acknowledge-mode 用默认的 auto,方法正常返回自动 ack,抛异常自动 nack,心智负担最小。手动 ack 那套方案我想了一下就放下了,能交给框架的就不自己攥在手里。

监听容器默认单消费者。通知这种场景吞吐要求不高,默认配置就够;要提并发,yml 里 listener.simple.concurrency 一行的事。我保持默认的原因是演示工程要观测计数,多消费者并发会让 attempts 的时序变得难读,先保可读,再谈吞吐。

重试,引擎是 spring-core 的 RetryTemplate

这一篇最重要的实测发现在这里。消费失败分两类:网络抖一下、下游服务抖一下的偶发失败,直接进死信太浪费,先重试几次是标准动作。yml 里用的是 listener.simple.retry.* 这一组配置,测试日志里栈顶的类是这样的

1
2
3
at org.springframework.core.retry.RetryTemplate.execute(RetryTemplate.java:174)
   ~[spring-core-7.0.9.jar:7.0.9]
at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterceptor.invoke(...)

https://static.xiongneng.me/rabbitmq-retry-flow-20260922005534.png

看清楚这个包名,org.springframework.core.retry,RetryTemplate 来自 Framework 7 的 spring-core,不是 Spring Retry 那个独立的 spring-retry 依赖。AMQP 消费重试的引擎就是它。这个发现直接影响两处。

第一处是依赖,重试不需要显式引 spring-retry,spring-core 自带,我的 pom 里一个重试相关的坐标都没有。第二处是语义,实测确认过,**max-attempts: 3 计的是重试次数,加上初始投递,一条消息最多被投递 4 次。**毒丸消息的日志里我数得清清楚楚,四次投递,间隔 500 毫秒、750 毫秒、1125 毫秒,标准的 1.5 倍退避。把 max-attempts: 3 理解成「总共投 3 次」,重试的计数就永远差一次,我第一版测试断言就是这么写错的。

为了把这个语义钉死,我做了一轮专门的实验:purge 掉两个队列,发一条毒丸消息,盯着日志数投递。第 1 次投递抛异常,等 500 毫秒;第 2 次还是失败,等 750 毫秒;第 3 次失败,等 1125 毫秒;第 4 次失败之后日志里出现 recoverer 的拒收记录,业务队列深度归零,消息出现在死信队列里。四次投递、三次重试、1.5 倍退避,三个数字互相印证,这一页日志我留了底。

还有一点要清楚,重试发生在监听线程里,重试期间这条线程被占着,别的消息在队列里排队。演示场景无所谓,高吞吐场景就得掂量:重试次数给多了,一条毒丸消息能占住一条线程好几秒。这也是死信兜底要尽快接手的原因,别让毒丸在业务队列里久留。

重试耗尽之后走哪条路,AbstractRabbitListenerContainerFactoryConfigurer 的字节码里确认过,默认的 recoverer 是 RejectAndDontRequeueRecoverer,抛异常拒收并且不再回队。拓扑那一节里业务队列挂着 x-dead-letter-exchange,拒收的消息被 broker 自动转进死信队列。三段剧本就此连起来了:偶发失败重试抢救,永久失败拒收,死信队列兜底等人工。

https://static.xiongneng.me/rabbitmq-dead-letter-path-20260922005534.png

这套链路有个前提我必须敲黑板。**死信转发是 broker 的行为,前提是声明队列的时候挂了死信参数。**只开重试不配 DLX 的工程不少见,重试耗尽的消息默认回队无限循环,一条毒丸消息能把消费者钉死。拓扑里那两个参数,不是可选项。

幂等,异步化之后必须还的债

消息从「函数调用」变成「网络投递」,一个物理事实就绕不开了:at-least-once。broker 不丢消息的代价是,ack 在网络上丢了它就再投一次,监听方法会收到两条一模一样的订单事件。这不是 bug,是协议的设计取舍,只能在业务代码里消化。

消化的方式就是幂等。**消费前按事件的唯一标识查一遍台账,处理过的直接跳过。**演示工程里的 NotificationService 用 ConcurrentHashMap 按 eventId 记台账,真实工程里这张表落库,加个唯一索引,幂等就是一条 insert 语句的事。顺序必须是先想幂等再上队列,反过来做的话,凌晨三点的第二条相同短信会把这笔债翻出来,这种电话我不想再接第二次。

我拿演示工程验证过这条路径:同一条事件手动投两次,第二次进来台账里 eventId 已经在了,监听方法直接跳过,统计面板里只有一条消费记录。几十行代码换来这种踏实,值。

完整案例,跑起来看

照例把整条链路的实测数据摆出来。服务连的是树莓派上的 RabbitMQ 4.3.6,所有数字来自实际运行,工程里那份验证脚本可以原样复现,我每一轮跑完都拿它的输出对过正文。

这一轮演示我按「先正常、再抖动、后毒丸」的顺序跑,每一步之间都 purge 过队列复位计数,所以三段数字既可以独立看,也可以合起来对总账。

创建订单,201 返回订单号,主链路到此为止,通知在后台异步走

1
2
POST /api/orders  {"product":"机械键盘","receiver":"熊大"}
→ 201 {"orderNo":"SO1001","product":"机械键盘","receiver":"熊大","status":"CREATED"}

https://static.xiongneng.me/rabbitmq-first-publish-stats-20260919181702.png

此刻查统计面板,一条事件已发布、已消费、只投递了一次,发送方确认是 ACK,broker 的回执到了,消息落在了队列里并被消费掉

接着演示重试。给另一个订单注入一次模拟故障,然后支付。故障注入后第一次投递失败,500 毫秒后重试成功,台账里这条事件的 attempts 是 2。统计面板里的 eventId 是每条事件的身份证,发布时生成,消费、重试、死信全程带着它走,对账就认它。下面各段记录里的 eventId 我原样照贴,读者复现的时候数值肯定不一样,格式和字段是一致的。

用户视角是零感知的,支付接口照常返回,短信晚到了半秒。这就是重试存在的意义,把基础设施的抖动挡在用户看不见的地方。

最后是毒丸。我发布一条永远消费失败的事件,看它走完全程

https://static.xiongneng.me/rabbitmq-poison-deadletter-20260919181717.png

四次投递全部失败,重试耗尽,RejectAndDontRequeueRecoverer 拒收不回队,broker 按死信参数把消息转进 sb4.order.dead,死信监听器记录台账,deadLettered 计数从 0 变 1。管理台里死信队列的深度此刻是 0,因为兜底消费者把它接走了,consumers 是 1。整条链路,发布、确认、重试、拒收、死信、兜底,每一环都拿数字当铁证。

死信兜底消费者干的事很朴素:记台账、打 error 日志、留给人工。我没有在这个监听器里做任何自动重试,能进死信的消息说明重试已经救不活了,再自动重试一轮只是把同样的失败重复一遍。人工介入之前消息已经安全落在死信队列里,这个设计把「救不活」和「不知道」分开,前者有记录可查,后者才是事故。

最终的全局统计

retried 是 4,毒丸贡献 3 次,PAID 事件贡献 1 次,对得上,我自己核过一遍。错误语义照例是 RFC 9457,404 带 resourceType 和 resourceId,400 的 errors 数组逐字段列原因,跟前面几篇一个规格,就不占篇幅了。

这一篇的测试怎么写

**异步链路的测试和同步最大的区别是,断言时机不由测试代码控制。**发完请求立刻断言,消费者可能还没跑到,我最早的几个用例就这么红过,检查半天断言本身一点错没有,错的是时机。做法是一个 untilAsserted 轮询助手,给定超时时间,断言失败就等 200 毫秒再试,超时才算失败。

超时给多长有讲究,给短了 CI 机器一忙就假失败,给长了真失败要干等。本篇的六个用例我统一给 5 秒,本地跑平均几百毫秒就过,超时只是兜底。轮询间隔 200 毫秒也是权衡过的,再密了对 broker 是无意义的压力,再疏了单次失败的等待体感明显。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
private static void untilAsserted(Duration timeout, Runnable assertion) {
    AssertionError last = null;
    long deadline = System.nanoTime() + timeout.toNanos();
    while (System.nanoTime() < deadline) {
        try {
            assertion.run();
            return;                        // 断言过了,收工
        } catch (AssertionError e) {
            last = e;                      // 记下,再等一轮
        }
        Thread.sleep(200);
    }
    throw last;
}

六个用例按 @Order 串联,每个用例的 @BeforeEach 里先 RabbitAdmin.purgeQueue 清空两个队列、内存台账复位,保证计数从零开始可断言。**共享 broker 上跑测试,上一轮的残留消息会把精心设计的计数搅得一塌糊涂。**这个习惯是缓存篇的脏缓存教训带过来的,学费只交一次。

顺便交代一下用 @Order 串联的原因:六个用例共享一个 broker 和一套计数,跑的顺序会影响计数基线,串起来才可复现。并行跑用例在这个场景里是给自己挖坑,我试过,两个用例同时 purge 对方刚发布的数据,断言结果完全不可信。

转换器的断言我单独说一下

1
2
assertThat(rabbitTemplate.getMessageConverter())
        .isInstanceOf(JacksonJsonMessageConverter.class);

一行,钉死「自动挂载机制已生效」。机制类的断言要钉在机制的产物上,这比只验证业务结果多一层保险,哪天升级 Boot 之后挂载行为变了,这条断言第一个知道。

避坑指南

坑一,@RabbitListener 收到的是字节,不是对象。 容器里没有唯一的 MessageConverter Bean,或者自定义容器工厂时忘了接转换器,监听方法的参数就反序列化不出来。现象是消息被反复 nack,或者参数变成 Message 原始类型。先查容器里有几个转换器 Bean,超过一个时自动配置会放弃挂载,监听失效九成是这个原因。

坑二,max-attempts 的语义要按重试次数算。 重试引擎是 Framework 7 的 RetryTemplate,max-attempts: 3 实测是最多重试 3 次、总共投递 4 次。按「总共 3 次」去写断言,计数永远差一次,我测试就栽在这里。重试间隔、总投递次数,这两个数建议像本篇一样写进测试断言。

坑三,RabbitTemplateCustomizer 有两个同名类。 Boot 自动配置的在 org.springframework.boot.amqp.autoconfigure;org.springframework.amqp.rabbit.core 里还有一个同名类,import 到那个,编译报找不到符号。IDE 弹提示的时候看清包名再回车。

坑四,只开重试不配死信,毒丸消息无限循环。 重试耗尽后消息默认回队重投,一条永远消费失败的消息会把消费者线程钉死在原地,队列深度不涨,日志疯狂刷错。业务队列声明时必须挂 x-dead-letter-exchange,给失败的消息留一条出路。

坑五,TypeId 裸奔。 不传信任包前缀的话,消费端对消息头声明的类型是放行的,公网环境里这是一类真实的反序列化攻击面。构造参数里把信任范围收窄到本工程的包前缀,一行代码的事。

坑六,消费端不做幂等。 at-least-once 是常态。网络抖动丢一次 ack,监听方法就会看到同一条事件两次。按 eventId 建台账加唯一索引,处理前先查,重复的直接跳过。这事上线前做是十分钟,上线后做是事故报告。

坑七,测试连共享 broker 不清队列。 上一轮残留的消息会让计数断言全面失守,症状是「本地全绿,CI 全红」或者反过来。每个用例开始前 purge 一次队列,成本几毫秒,换来计数断言的确定性。

小结

什么场景该上消息队列,我的判断标准就一条:下游挂了你等不等得起。等不起的(通知、积分、审计这类)异步化,等得起的留在主链路里。定了走 RabbitMQ 这条路,我建议的次序是:发送端把 confirm 和 return 都打开,转换器选 Jackson 3 那个并把信任包收窄到自己的包前缀,消费端重试参数从本篇那份 yaml 起步,队列声明时死信参数一次配齐,幂等设计放在上队列之前想。三道防线各管一段,补齐的成本都很低,事后补的代价都很高。

没解决的事也有两件。一是消费端的并发与顺序:多消费者并行时,同一订单的事件可能乱序到达,本篇场景无所谓,要做状态机类消费就得另外设计,我还没在工程里落地方案,不敢瞎给建议。二是毒丸场景的日志核对目前仍要人眼过一遍,验证脚本的断言还没覆盖到日志层,等我把这块补齐再来交代。

参考链接