Spring Cloud Stream绑定RabbitMQ反应式消费者闲置后失效问题

问题根因 #

• 违反Spring Cloud Stream反应式消费者的约定：你定义的Consumer<Flux<Message<RssModel>>>自行手动调用了subscribe()，而框架本身会负责整个流的生命周期管理、错误恢复、重试逻辑，你自行订阅后框架无法接管流的异常处理和自动重建。
• 流异常未捕获：当前的Flux处理链没有定义错误恢复策略，一旦运行中出现任何未捕获异常（比如反序列化异常、网络波动、业务处理报错），Flux流会直接终止，对应消息通道的订阅者也会被销毁，后续新消息就会抛出无订阅者的错误。
• 隐含风险：你发送的JSON字段为小驼峰格式（contentCategory、rssFeedUrl），但RssModel的字段为全小写（contentcategory、rssfeedurl），如果反序列化配置不匹配，也可能触发异常导致流终止。

解决方案 #

1. 修复反应式消费者写法，移除自行subscribe逻辑 #

优先使用符合框架约定的Function写法，不需要自行处理订阅，框架会自动管理流的生命周期，出现异常也会自动重建流：

@Bean
public Function<Flux<Message<RssModel>>, Flux<Void>> rssConsumer1(RssTaskHandler taskHandler) {
    return flux -> flux
        .map(Message::getPayload)
        .doOnNext(System.out::println)
        // 单条消息出错跳过，不终止整个流
        .onErrorContinue((throwable, obj) -> {
            System.err.println("处理消息出错，消息内容：" + obj + "，异常信息：" + throwable.getMessage());
        })
        .thenMany(Flux.empty());
}

如果要保留Consumer写法，需要在subscribe时增加错误回调，避免流异常终止：

@Bean
public Consumer<Flux<Message<RssModel>>> rssConsumer1(RssTaskHandler taskHandler) {
    return flux -> flux
        .map(Message::getPayload)
        .doOnNext(System.out::println)
        .onErrorContinue((throwable, obj) -> {
            System.err.println("消费异常：" + throwable.getMessage());
        })
        .subscribe(
            data -> {},
            error -> System.err.println("流全局异常：" + error.getMessage())
        );
}

2. 修复RssModel序列化匹配问题，避免反序列化异常 #

给字段加上@JsonProperty注解匹配传入的JSON字段名：

public class RssModel {
    @JsonProperty("contentCategory")
    private String contentcategory;
    @JsonProperty("rssFeedUrl")
    private String rssfeedurl;
    private String channel;
    @JsonProperty("contentOwner")
    private String contentowner;

    // 原有getter、setter、toString逻辑保持不变
}

3. 可选配置：增加重试和死信队列，避免消息丢失 #

在application.yml中补充消费者可靠性配置：

spring:
  main:
    allow-bean-definition-overriding: true
  cloud:
    stream:
      bindings:
        requester1:
          destination: rss-exchange
          group: requester
          consumer:
            max-attempts: 3 # 消费失败重试次数
            retryable-exceptions: java.lang.Exception # 全异常类型触发重试
      rabbit:
        bindings:
          requester1:
            consumer:
              auto-bind-dlq: true # 自动绑定死信队列，重试失败的消息转入死信
              dlq-ttl: 86400000 # 死信消息保存1天
      function:
        bindings:
          rssConsumer1-in-0: requester1        
        definition: rssConsumer1

内容的提问来源于stack exchange，提问作者MAY