Spring Cloud Stream Rabbit Binder批量消息发DLQ报错及事务处理问题

问题1：批量消费触发DLQ的类型转换报错 #

这是Spring Cloud Stream 3.1.x版本Rabbit Binder的已知问题，批量消费模式下默认错误处理器生成的ErrorMessage payload格式与Binder期望格式不匹配，解决方案如下：

• 方案1：关闭自动DLQ，手动处理异常
  移除配置中自动绑定DLQ的相关配置，在消费逻辑最外层增加try-catch包裹整个处理链路，捕获异常后手动将批量消息封装为Message<List<?>>格式，自行投递到死信队列。
• 方案2：自定义错误处理器
  重写ConditionalRejectingErrorHandler，在错误处理逻辑中将payload为List<Message<?>>的ErrorMessage重新封装为Message<List<?>>类型后再传递给后续的DLQ发送逻辑，即可避免类转换异常。
• 方案3：版本升级
  升级到Spring Cloud Stream 3.2及以上版本，该版本官方已修复批量消费场景下的DLQ格式兼容问题。

问题2：函数组合链路的事务一致性保证 #

要保证listen|process|send整条链路的事务原子性，同时实现StreamBridge发送消息的回滚，按以下步骤配置即可：

1. 配置统一事务管理器
   定义RabbitTransactionManager作为全局事务管理器，若有数据库事务需求可使用ChainedTransactionManager将Rabbit事务与数据源事务串联，保证多资源事务的一致性。
2. 消费端绑定事务
   在消费端绑定配置中增加事务管理器关联，让整个消费链路都运行在同一个事务上下文内：

   spring:
     cloud:
       stream:
         bindings:
           listen-in-0:
             consumer:
               transaction-manager: rabbitTransactionManager
   

3. 保证StreamBridge事务同步
   确保StreamBridge使用的RabbitTemplate开启了channelTransacted=true，且与消费端共用同一个事务上下文，不要单独初始化独立的RabbitTemplate实例。此时整条链路任意节点抛出异常，事务会整体回滚：process中通过StreamBridge发送的消息不会提交到Rabbit Broker，原消费的批次消息也会重新入队。

注意事项 #

• 当前配置中maxAttempts=1表示异常后不会重试，若开启重试需注意批量消费的重试是整个批次统一重试，不会单条重试。
• 手动处理异常时不要部分确认批次内的消息，避免出现消息丢失或重复消费的不一致问题。

内容的提问来源于stack exchange，提问作者Sébastien