Spring Cloud Stream 4.0.1(Kafka)如何获取消息发布后的输出通道名称？

背景说明 #

Spring Cloud Stream 4.0.1（基于Spring Boot 3.x）的Kafka Binder已移除旧版的record-metadata-channel配置，原配置spring.cloud.stream.kafka.bindings.myBinding.producer.record-metadata-channel不再生效。

替代方案 #

1. 通过ProducerListener监听发送结果并获取通道信息 #

自定义ProducerListener捕获消息发送状态，同时借助框架内置的头信息获取输出通道名称：

import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

@Component
public class ChannelAwareProducerListener implements ProducerListener<Object, Object> {

    @Override
    public void onSuccess(ProducerRecordMetadata metadata, Message<?> message) {
        // 从消息头中获取框架内置的输出通道标识
        String outputChannel = message.getHeaders().get(KafkaMessageChannelBinder.OUTPUT_CHANNEL_HEADER, String.class);
        System.out.println("消息发送成功，输出通道：" + outputChannel);
        // 同时可获取Kafka元数据
        System.out.println("关联Topic：" + metadata.topic() + "，Partition：" + metadata.partition());
    }

    @Override
    public void onError(ProducerRecordMetadata metadata, Message<?> message, Exception exception) {
        String outputChannel = message.getHeaders().get(KafkaMessageChannelBinder.OUTPUT_CHANNEL_HEADER, String.class);
        System.err.println("消息发送失败，输出通道：" + outputChannel + "，错误：" + exception.getMessage());
    }
}

通过ProducerCustomizer将监听器注册到KafkaTemplate：

import org.springframework.cloud.stream.binder.kafka.producer.KafkaProducerProperties;
import org.springframework.cloud.stream.binder.kafka.producer.ProducerCustomizer;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducerConfigurer implements ProducerCustomizer<Object, Object> {

    private final ChannelAwareProducerListener listener;

    public KafkaProducerConfigurer(ChannelAwareProducerListener listener) {
        this.listener = listener;
    }

    @Override
    public void customize(KafkaTemplate<Object, Object> kafkaTemplate, String destination, KafkaProducerProperties properties) {
        kafkaTemplate.setProducerListener(listener);
    }
}

2. 使用StreamBridge发送时直接关联通道标识 #

如果用StreamBridge发送消息，可手动在消息头中标记通道名称，后续通过异步回调获取：

import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Service
public class MessageSender {

    private final StreamBridge streamBridge;

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

    public void sendToChannel(String outputChannel, Object payload) {
        Message<Object> message = MessageBuilder.withPayload(payload)
                .setHeader("target-channel", outputChannel)
                .build();
        // 异步处理发送结果
        streamBridge.send(outputChannel, message)
                .thenAccept(success -> {
                    if (success) {
                        System.out.println("通道 " + outputChannel + " 消息发送成功");
                    } else {
                        System.err.println("通道 " + outputChannel + " 消息发送失败");
                    }
                })
                .exceptionally(ex -> {
                    System.err.println("通道 " + outputChannel + " 发送异常：" + ex.getMessage());
                    return null;
                });
    }
}

关键提示 #

• KafkaMessageChannelBinder.OUTPUT_CHANNEL_HEADER是框架内置的头字段，无需手动添加，可直接从发送的消息中提取通道名称。
• 两种方案都可同时获取Kafka消息的元数据（如Topic、Partition、Offset），满足原record-metadata-channel的核心需求。

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