Spring Boot 3分布式系统中如何将ThreadLocal的Correlation ID传入Kafka Headers？

在Spring Boot生态中，要实现不侵入业务逻辑的Kafka消息Correlation-ID自动注入，Spring Kafka提供的ProducerPostProcessor是标准的扩展方案，它能在生产者发送消息前统一处理消息头，无需在业务代码中手动添加Header。

方案1：实现ProducerPostProcessor自动注入Header #

ProducerPostProcessor是Spring Kafka提供的扩展接口，允许在消息发送前拦截并修改ProducerRecord。我们可以实现这个接口，从ThreadLocal中获取Correlation-ID并自动添加到Kafka Headers中。

代码实现 #

import org.springframework.kafka.core.ProducerPostProcessor;
import org.springframework.stereotype.Component;
import org.apache.kafka.clients.producer.ProducerRecord;

@Component
public class CorrelationIdProducerInterceptor implements ProducerPostProcessor<String, String> {

    @Override
    public ProducerRecord<String, String> postProcessBeforeSend(ProducerRecord<String, String> record, org.springframework.messaging.Message<?> message) {
        // 从自定义上下文Holder中获取Correlation-ID
        String correlationId = MyContextHolder.get();
        if (correlationId != null) {
            record.headers().add("correlation-id", correlationId.getBytes());
        }
        return record;
    }
}

效果说明 #

Spring会自动发现并注册这个组件，所有通过KafkaTemplate发送的消息都会经过这个处理器。业务代码只需正常调用kafkaTemplate.send()即可，完全不需要关心Correlation-ID的注入逻辑：

// 业务代码无需手动处理Header，直接发送消息
public void sendMessage(String payload) {
    kafkaTemplate.send("topic", payload);
}

方案2：解决异步线程的上下文传递问题 #

如果你的Kafka消息发送操作是在异步线程中执行的（比如使用@Async），ThreadLocal中的上下文默认不会自动传递到异步线程。此时需要配置线程池的TaskDecorator来复制上下文。

线程池配置 #

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;

@Configuration
public class AsyncThreadPoolConfig {

    @Bean(name = "kafkaAsyncExecutor")
    public Executor kafkaAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(4);
        executor.setMaxPoolSize(8);
        executor.setQueueCapacity(50);
        executor.setThreadNamePrefix("Kafka-Async-Sender-");
        // 配置上下文复制装饰器
        executor.setTaskDecorator(new CorrelationIdContextDecorator());
        executor.initialize();
        return executor;
    }

    private static class CorrelationIdContextDecorator implements TaskDecorator {

        @Override
        public Runnable decorate(Runnable runnable) {
            // 捕获当前线程的Correlation-ID
            String currentCorrelationId = MyContextHolder.get();
            return () -> {
                try {
                    // 将Correlation-ID设置到异步线程的ThreadLocal中
                    MyContextHolder.set(currentCorrelationId);
                    runnable.run();
                } finally {
                    // 执行完毕后清理上下文，避免内存泄漏
                    MyContextHolder.clear();
                }
            };
        }
    }
}

异步方法使用 #

在异步发送消息的方法上指定该线程池：

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

@Service
public class KafkaMessageService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaMessageService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Async("kafkaAsyncExecutor")
    public void sendAsyncMessage(String payload) {
        kafkaTemplate.send("topic", payload);
    }
}

核心优势 #

• 完全解耦：业务逻辑无需感知Correlation-ID的传播细节，专注于业务功能实现。
• 统一管理：所有消息的Header注入逻辑集中在ProducerPostProcessor中，便于后续修改和维护。
• 符合Spring生态规范：基于Spring Kafka官方扩展点实现，稳定性和兼容性有保障。

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