Spring Cloud StreamBridge对接Kafka：方案优化与投递确认问询

我的Spring Cloud Stream有以下需求：

• 从一个Kafka集群的单个主题获取KStream，发送消息至另一个集群的多个主题；
• 部分场景下需根据单条接收消息生成多条消息发送；
• 所有消息需保证至少一次投递。

我曾尝试使用Function，但未解决单消息生成多消息发送的问题；也考虑过Consumer+Supplier的组合，但适配性不佳。目前我采用Consumer结合StreamBridge通过副作用发送消息，实现代码如下：

@Bean
@SuppressWarnings("unchecked")
public Consumer<KStream<String, String>> generateMessage() {
    return messages -> {
        final Map<String, KStream<String, String>> splitMessages =
                branchOutput(filterMessages(messages));

        KStream<String, MessageData>[] ksArray = splitMessages
                .values()
                .stream()
                .map(message ->
                        message.mapValues((key, jsonMessage) -> {
                            try {
                                return new MessageData(dataTransformService
                                        .transformMessage(key, jsonMessage, extractTopic(jsonMessage)),
                                        removeTopic(jsonMessage), "");
                            } catch (ClassNotFoundException e) {
                                return new MessageData(Collections.singletonList(CLASS_NOT_FOUND_EXCEPTION),
                                        removeTopic(jsonMessage), e.getMessage());
                            }
                        }))
                .toArray(KStream[]::new);

        ksArray[0].peek((key, value) -> sendMessage(key, value.getTransformedMessages(),
                OUTPUT_BINDING_1, value.getOriginalMessage(), value.getError()));
        ksArray[1].peek((key, value) -> sendMessage(key, value.getTransformedMessages(),
                OUTPUT_BINDING_2, value.getOriginalMessage(), value.getError()));
        ksArray[2].peek((key, value) -> sendMessage(key, value.getTransformedMessages(),
                OUTPUT_BINDING_3, value.getOriginalMessage(), value.getError()));
        ksArray[3].peek((key, value) -> sendMessage(key, value.getTransformedMessages(),
                OUTPUT_BINDING_4, value.getOriginalMessage(), value.getError()));
    };
}

// send message(s) to topic or forward to dlq if there is a message handling exception
private void sendMessage(String key, List<String> transformedMessages, String binding, String originalMessage, String error) {
    try {
        for (String transformedMessage : transformedMessages) {
            if (!transformedMessage.equals(CLASS_NOT_FOUND_EXCEPTION)) {
                boolean sendTest = streamBridge.send(binding,
                        new GenericMessage<>(transformedMessage, Collections.singletonMap(
                                KafkaHeaders.KEY, (extractMessageId(transformedMessage)).getBytes())));

                log.debug(String.format("message sent = %s", sendTest));

            } else {
                log.warn(String.format("message transform error: %s", error));
                streamBridge.send(DLQ_OUTPUT_BINDING,
                        new GenericMessage<>(originalMessage, Collections.singletonMap(KafkaHeaders.KEY,
                                key.getBytes())));
            }
        }

    } catch (MessageHandlingException e) {
        log.warn(String.format("message send error: %s", e));
        streamBridge.send(DLQ_OUTPUT_BINDING,
                new GenericMessage<>(originalMessage, Collections.singletonMap(KafkaHeaders.KEY,
                        key.getBytes())));

    }
}

我想了解：

1. 是否有更优的方案满足上述需求？
2. 若没有，能否对接外部Kafka集群（非我方管理）的投递确认机制，在消息未被接收时进行重试？

一、更优方案：基于KStream原生API实现 #

你当前用StreamBridge在peek中发送消息的方式，绕过了KStream流处理的生命周期管理，会丢失Kafka Streams自带的偏移量自动提交、端到端投递保障能力。更推荐用KStream原生API实现需求：

核心逻辑 #

1. 分支拆分：用branch方法按业务规则将原流拆分为多个子流，对应不同输出主题；
2. 一对多转换：用flatMapValues将单条输入消息转换为多条输出消息，原生支持一对多场景；
3. 多集群绑定：通过Spring Cloud Stream的多binder配置，直接将子流绑定到目标Kafka集群的主题，无需手动发送。

示例代码 #

@Bean
public Function<KStream<String, String>, KStream<String, String>[]> processMessages() {
    return input -> {
        // 1. 过滤输入消息
        KStream<String, String> filteredStream = input.filter((key, value) -> /* 你的过滤逻辑 */);
        
        // 2. 按业务规则分支为4个子流
        KStream<String, String>[] branches = filteredStream.branch(
            (key, value) -> /* 匹配OUTPUT_BINDING_1的条件 */,
            (key, value) -> /* 匹配OUTPUT_BINDING_2的条件 */,
            (key, value) -> /* 匹配OUTPUT_BINDING_3的条件 */,
            (key, value) -> /* 匹配OUTPUT_BINDING_4的条件 */
        );

        // 3. 对每个分支做一对多转换+异常处理
        return Arrays.stream(branches)
            .map(stream -> stream.flatMapValues((key, jsonMessage) -> {
                try {
                    // 单消息转多条，返回List<String>
                    return dataTransformService.transformMessage(key, jsonMessage, extractTopic(jsonMessage));
                } catch (ClassNotFoundException e) {
                    // 转换异常时返回DLQ标记
                    return Collections.singletonList(CLASS_NOT_FOUND_EXCEPTION);
                }
            })
            // 拆分异常流与正常流，异常消息转发到DLQ
            .split()
            .branch((key, value) -> value.equals(CLASS_NOT_FOUND_EXCEPTION),
                subStream -> subStream.to(DLQ_OUTPUT_BINDING))
            .defaultBranch(Function.identity()))
            .toArray(KStream[]::new);
    };
}

配置示例（application.yml） #

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            brokers: 源Kafka集群地址
      bindings:
        processMessages-in-0:
          destination: 源主题名称
        processMessages-out-0:
          destination: 目标主题1
          binder: target-kafka-binder
        processMessages-out-1:
          destination: 目标主题2
          binder: target-kafka-binder
        processMessages-out-2:
          destination: 目标主题3
          binder: target-kafka-binder
        processMessages-out-3:
          destination: 目标主题4
          binder: target-kafka-binder
        dlq-output:
          destination: dlq主题名称
          binder: target-kafka-binder
      binders:
        target-kafka-binder:
          type: kafka
          environment:
            spring:
              kafka:
                bootstrap-servers: 目标Kafka集群地址

方案优势 #

• 完全基于Kafka Streams原生能力，自动管理偏移量，天然满足至少一次投递要求；
• 一对多转换逻辑更简洁，无副作用；
• 多集群通过配置实现，无需手动处理跨集群连接。

二、投递确认与重试机制（适配外部集群） #

如果必须保留StreamBridge实现，或外部集群有特殊确认要求，可通过以下方式对接投递确认与重试：

1. 启用生产者原生确认与重试 #

在外部集群的binder配置中开启生产者同步确认与内置重试：

spring:
  cloud:
    stream:
      binders:
        target-kafka-binder:
          environment:
            spring:
              kafka:
                producer:
                  acks: all # 等待所有副本确认消息写入
                  retries: 3 # 内置重试次数
                  retry-backoff-ms: 1000 # 重试间隔

2. 手动监听投递结果并自定义重试 #

StreamBridge的send方法返回的boolean仅表示消息进入生产者缓冲区，并非最终投递成功。需用重载方法获取ListenableFuture监听结果，实现精确重试：

private void sendMessage(String key, List<String> transformedMessages, String binding, String originalMessage, String error) {
    for (String transformedMessage : transformedMessages) {
        if (!transformedMessage.equals(CLASS_NOT_FOUND_EXCEPTION)) {
            Message<String> message = MessageBuilder
                .withPayload(transformedMessage)
                .setHeader(KafkaHeaders.KEY, extractMessageId(transformedMessage).getBytes())
                .build();
            
            // 异步监听投递结果
            ListenableFuture<SendResult<String, String>> future = streamBridge.send(binding, message);
            future.addCallback(
                result -> log.debug("消息投递成功: {}", result.getRecordMetadata()),
                ex -> {
                    log.warn("消息投递失败，触发重试: {}", ex.getMessage());
                    // 调用自定义重试方法
                    retrySend(message, binding, originalMessage, key);
                }
            );
        } else {
            // 转发到DLQ
            streamBridge.send(DLQ_OUTPUT_BINDING, MessageBuilder
                .withPayload(originalMessage)
                .setHeader(KafkaHeaders.KEY, key.getBytes())
                .build());
        }
    }
}

// 结合Spring Retry实现自定义重试
@Retryable(value = {MessageHandlingException.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000))
private void retrySend(Message<String> message, String binding, String originalMessage, String key) {
    try {
        streamBridge.send(binding, message);
    } catch (MessageHandlingException e) {
        log.error("重试3次仍失败，转发到DLQ");
        streamBridge.send(DLQ_OUTPUT_BINDING, MessageBuilder
            .withPayload(originalMessage)
            .setHeader(KafkaHeaders.KEY, key.getBytes())
            .build());
        throw e; // 终止重试
    }
}

注意事项 #

• 若外部集群禁用生产者重试，需完全依赖自定义重试逻辑；
• 重试可能导致消息重复，建议给每条消息设置唯一ID，目标消费者实现幂等处理；
• 若外部集群支持事务，可开启Spring Kafka事务管理器，进一步保障投递一致性。

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