使用Spring Cloud Function实现RabbitMQ下Spring Cloud Stream DLQ处理是否可行？

通过Spring Cloud Function实现DLQ处理的方案完全可行，无需依赖@RabbitListener注解，写法和普通函数式消费者完全一致。

需要完成的配置和注意事项如下：

1. 基础配置调整 #

首先需要在配置文件中完成两部分配置：业务消费者开启DLQ、DLQ消费者绑定死信队列：

spring:
  cloud:
    stream:
      # 全局序列化配置，保证业务消息和DLQ消息序列化规则一致
      content-type: application/json
      rabbit:
        bindings:
          # 业务消费者的Rabbit专属配置，开启自动绑定DLQ
          businessConsumer-in-0:
            consumer:
              auto-bind-dlq: true
              # 可选配置：DLQ消息过期时间、自定义死信交换器等
              # dlq-ttl: 86400000
      bindings:
        # 业务消费者配置示例
        businessConsumer-in-0:
          destination: order-event
          group: order-service-group
          consumer:
            max-attempts: 1 # 不需要本地重试则设为1，消费失败直接进入DLQ
        # DLQ处理消费者的绑定配置
        dlqprocess-in-0:
          # 死信队列默认命名规则：{业务topic名}.{业务消费组名}.dlq
          destination: order-event.order-service-group.dlq
          group: dlq-process-group

注意：函数式消费者的binding命名遵循{方法名}-in-{序号}规则，你定义的Consumer方法名为dlqprocess，单输入参数对应的binding名就是dlqprocess-in-0。

2. 代码优化建议 #

你提供的代码样例可以进一步扩展，获取死信队列自带的元数据用于问题排查：

@Bean
public Consumer<Message<OrderEvent>> dlqprocess(DLQProcess dlqprocess) {
    return message -> {
        // 从Header中获取死信相关原生信息
        String exceptionMsg = (String) message.getHeaders().get("x-exception-message");
        String originalQueue = (String) message.getHeaders().get("x-original-queue");
        // 执行自定义死信处理逻辑
        dlqprocess.process(message.getPayload(), exceptionMsg, originalQueue);
    };
}

3. 额外注意事项 #

• 若消息为自定义对象类型，需要保证DLQ消费者和业务生产者的序列化规则一致，避免反序列化失败
• 如果修改过默认的死信队列命名规则，dlqprocess-in-0绑定的destination需要替换为你自定义的DLQ队列名
• 若不需要保留DLQ消费的偏移量，可以将dlq-process-group设置为匿名消费组，正式环境建议指定固定消费组避免消息丢失

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