Spring Cloud Stream中如何获取消息的Correlation ID？

当你使用Supplier<Flux<Integer>>这种简洁方式发送Kafka消息时，由于Spring会自动订阅Flux并处理发送流程，默认确实无法直接拿到类似ReactiveKafkaSender返回的SenderResult（包含Correlation ID等元数据）。不过可以通过以下几种方式实现需求：

1. 手动构建消息+ProducerListener监听结果 #

首先在生产者中为每个消息生成并绑定Correlation ID到消息头：

@Bean
Supplier<Flux<Message<Integer>>> someProducer() {
    return () -> Flux.range(1, 10)
            .map(num -> {
                // 生成唯一Correlation ID
                String correlationId = UUID.randomUUID().toString();
                return MessageBuilder.withPayload(num)
                        .setHeader(KafkaHeaders.CORRELATION_ID, correlationId)
                        .build();
            });
}

然后自定义ProducerListener捕获发送结果并提取Correlation ID：

@Component
public class KafkaProducerListener implements ProducerListener<Integer, Integer> {
    @Override
    public void onSuccess(ProducerRecord<Integer, Integer> record, RecordMetadata metadata) {
        Header correlationIdHeader = record.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
        if (correlationIdHeader != null) {
            String correlationId = new String(correlationIdHeader.value());
            // 这里做Correlation ID相关处理，比如日志、状态更新
            System.out.printf("消息发送成功，Correlation ID: %s，偏移量: %d%n", 
                             correlationId, metadata.offset());
        }
    }

    @Override
    public void onError(ProducerRecord<Integer, Integer> record, Exception exception) {
        Header correlationIdHeader = record.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
        if (correlationIdHeader != null) {
            String correlationId = new String(correlationIdHeader.value());
            System.err.printf("消息发送失败，Correlation ID: %s，异常: %s%n", 
                             correlationId, exception.getMessage());
        }
    }
}

2. 用ReactiveKafkaProducerTemplate直接控制发送流程 #

如果需要更直接地获取每个消息的发送结果，可以在Supplier中手动使用ReactiveKafkaProducerTemplate发送，返回包含SenderResult的Flux：

@Bean
Supplier<Flux<SenderResult<Void>>> someProducer(ReactiveKafkaProducerTemplate<Integer, Integer> producerTemplate) {
    return () -> Flux.range(1, 10)
            .flatMap(num -> {
                String correlationId = UUID.randomUUID().toString();
                // 构建ProducerRecord并添加Correlation ID头
                ProducerRecord<Integer, Integer> record = new ProducerRecord<>("your-topic-name", num);
                record.headers().add(KafkaHeaders.CORRELATION_ID, correlationId.getBytes());
                
                // 发送并处理结果
                return producerTemplate.send(record)
                        .map(senderResult -> {
                            // 直接获取Correlation ID做后续处理
                            System.out.printf("消息已发送，Correlation ID: %s，分区: %d%n", 
                                             correlationId, senderResult.recordMetadata().partition());
                            return senderResult;
                        })
                        .onErrorResume(e -> {
                            System.err.printf("消息发送失败，Correlation ID: %s，原因: %s%n", 
                                             correlationId, e.getMessage());
                            return Mono.empty();
                        });
            });
}

这种方式下Spring依然会自动订阅Flux，但你可以在map或onErrorResume中直接处理每个消息的Correlation ID和发送状态。

关键说明 #

• 无论哪种方式，都需要手动生成并绑定Correlation ID到消息头，Spring默认不会为自动发送的消息添加该元数据。
• 若使用Spring Cloud Stream，需确保Kafka绑定器配置正确，ProducerListener会被自动识别注册。

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