Spring Boot 3中Spring Cloud Stream发送Kafka后如何关联TraceId记录元数据

我使用Spring Boot 3，通过Spring Cloud Stream向Kafka发送消息后，希望用发送消息线程的TraceId来记录已发送消息的offset和分区。

相关配置与代码如下：

• build.gradle依赖：

implementation "io.micrometer:micrometer-registry-prometheus"
implementation 'io.micrometer:micrometer-tracing'
implementation 'io.micrometer:micrometer-tracing-bridge-brave'

• 生产者application.yml配置：

recordMetadataChannel: meta
spring.cloud.stream.kafka.binder.enable-observation: true

• 配置类代码：

@Bean
public MessageChannel meta() {
    return new DirectChannel();
}

@Bean
public IntegrationFlow logMeta() {
    return f -> f.channel("meta")
            .handle((payload, headers) -> {
                log.info("Log metadata from headers");
                return payload;
            });
}

目前能正常获取消息元数据，但在IntegrationFlow的处理器中，Kafka头里没有traceparent头（不过后续发送的消息本身能看到这个头），希望能在日志中记录对应的TraceId和SpanId，请问有什么办法可以获取到发送线程的TraceId？

方法一：直接从Tracing上下文获取 #

由于元数据回调线程继承了发送消息线程的Trace上下文，无需从消息头提取，可直接通过Micrometer Tracing的API获取当前TraceId和SpanId：

修改IntegrationFlow的处理器代码：

import io.micrometer.tracing.Tracer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.MDC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlow;

public class KafkaMetaConfig {
    private static final Logger log = LoggerFactory.getLogger(KafkaMetaConfig.class);

    @Bean
    public MessageChannel meta() {
        return new DirectChannel();
    }

    @Bean
    public IntegrationFlow logMeta(Tracer tracer) {
        return f -> f.channel("meta")
                .handle((payload, headers) -> {
                    RecordMetadata meta = (RecordMetadata) payload;
                    // 获取当前Trace上下文的ID
                    String traceId = tracer.currentSpan().context().traceId();
                    String spanId = tracer.currentSpan().context().spanId();
                    // 放入MDC，日志框架可自动输出
                    MDC.put("traceId", traceId);
                    MDC.put("spanId", spanId);
                    try {
                        log.info("Kafka message sent - offset: {}, partition: {}", meta.offset(), meta.partition());
                    } finally {
                        // 清理MDC，防止线程池复用导致上下文污染
                        MDC.remove("traceId");
                        MDC.remove("spanId");
                    }
                    return payload;
                });
    }
}

方法二：通过通道拦截器传递上下文 #

如果元数据回调线程未继承发送线程的上下文，可给meta通道添加拦截器，将发送线程的Trace上下文存入消息头：

1. 定义Tracing通道拦截器：

import io.micrometer.tracing.Tracer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.MessageHeaderAccessor;

public class TracingChannelInterceptor implements ChannelInterceptor {
    private final Tracer tracer;

    public TracingChannelInterceptor(Tracer tracer) {
        this.tracer = tracer;
    }

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        if (tracer.currentSpan() != null) {
            MessageHeaderAccessor accessor = MessageHeaderAccessor.getMutableAccessor(message);
            accessor.setHeader("traceId", tracer.currentSpan().context().traceId());
            accessor.setHeader("spanId", tracer.currentSpan().context().spanId());
            return accessor.getMessage();
        }
        return message;
    }
}

2. 给meta通道添加拦截器：

@Bean
public MessageChannel meta(Tracer tracer) {
    DirectChannel channel = new DirectChannel();
    channel.addInterceptor(new TracingChannelInterceptor(tracer));
    return channel;
}

3. 在处理器中从消息头提取：

@Bean
public IntegrationFlow logMeta() {
    return f -> f.channel("meta")
            .handle((payload, headers) -> {
                RecordMetadata meta = (RecordMetadata) payload;
                String traceId = (String) headers.get("traceId");
                String spanId = (String) headers.get("spanId");
                if (traceId != null && spanId != null) {
                    MDC.put("traceId", traceId);
                    MDC.put("spanId", spanId);
                }
                try {
                    log.info("Kafka message sent - offset: {}, partition: {}", meta.offset(), meta.partition());
                } finally {
                    MDC.remove("traceId");
                    MDC.remove("spanId");
                }
                return payload;
            });
}

补充说明 #

• 由于已开启spring.cloud.stream.kafka.binder.enable-observation: true，发送消息的线程会自动创建并持有Trace上下文，方法一通常是最优解，无需额外配置。
• 若使用SLF4J日志框架，可在日志配置文件中添加%X{traceId}和%X{spanId}到日志格式，自动输出TraceId和SpanId，无需手动拼接日志内容。

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