通过REST调用重路由DLQ消息至Spring Cloud Stream重新处理

1. 新增重试用的Function #

创建一个简单的Function，用于接收DLQ消息并直接转发到原输入队列，无需修改消息内容：

import java.util.function.Function;
import org.springframework.stereotype.Component;

@Component
public class DlqRetryFunction {

    // 定义重试函数，名称为retryDlq
    public Function<byte[], byte[]> retryDlq() {
        return message -> message;
    }
}

采用byte[]类型是为了匹配原配置的序列化规则，避免额外的序列化/反序列化操作导致消息格式异常。

2. 更新Spring Cloud Stream绑定配置 #

修改原配置文件，添加重试函数的定义与绑定规则：

spring:
  application:
    name: app
  cloud:
    stream:
      default:
        group: abs
      # 新增retryDlq函数，与原有函数用分号分隔
      function.definition: abc|def|ghi;retryDlq
      kafka:
        binder:
        bindings:
          # 原输入绑定配置保持不变
          abcdefghi-in-0:
            consumer:
              enable-dlq: true
              dlq-name: a.b.c.deadletter
              dlq-producer-properties:
                configuration:
                  schema.registry.url: ${kafka.schema.registry.url}
                  key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
                  value.serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
          # 重试函数输入绑定（从DLQ消费）
          retryDlq-in-0:
            consumer:
              group: abs-dlq-retry # 独立消费组，避免与其他消费者冲突
              properties:
                schema.registry.url: ${kafka.schema.registry.url}
                key.deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
                value.deserializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufDeserializer
                specific.protobuf.value.type: 你的Protobuf消息全类名 # 替换为实际Protobuf类路径
          # 重试函数输出绑定（发送到原输入队列）
          retryDlq-out-0:
            producer:
              properties:
                schema.registry.url: ${kafka.schema.registry.url}
                key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer
                value.serializer: io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
      bindings:
        # 原输入绑定保持不变
        abcdefghi-in-0:
          destination: a.b.c.in
          content-type: application/x-protobuf
        # 重试函数输入绑定到DLQ
        retryDlq-in-0:
          destination: a.b.c.deadletter
          content-type: application/x-protobuf
        # 重试函数输出绑定到原输入队列
        retryDlq-out-0:
          destination: a.b.c.in
          content-type: application/x-protobuf

3. 通过REST触发重试 #

提供两种触发方式，按需选择：

方式一：使用Spring Cloud Stream内置Actuator端点 #

先启用Actuator的Function端点：

management:
  endpoints:
    web:
      exposure:
        include: function

然后通过POST请求触发重试：

POST /actuator/function
Content-Type: application/json

{
  "name": "retryDlq",
  "args": []
}

方式二：自定义REST接口 #

如果需要更灵活的控制（比如过滤特定消息、批量重试），可以自定义Controller：

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class DlqRetryController {

    private final StreamBridge streamBridge;

    public DlqRetryController(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    @PostMapping("/dlq/retry")
    public String initiateDlqRetry() {
        // 触发重试函数执行，将DLQ消息转发到原输入队列
        streamBridge.send("retryDlq-out-0", null);
        return "DLQ重试已启动";
    }
}

4. 关键注意事项 #

• 消费组隔离：重试函数的DLQ消费组需与其他消费者组区分，避免重复消费。
• 序列化一致性：确保重试流程的序列化/反序列化配置与原流程完全一致，防止消息格式错误。
• 幂等性保障：若原业务逻辑未实现幂等，需补充幂等校验，避免重复处理消息导致业务异常。

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