Spring Boot 3中Micrometer追踪上下文向Spring Batch异步任务传播咨询

核心问题原因 #

SimpleAsyncTaskExecutor默认不会自动传播Micrometer的Trace上下文，并行任务启动时会创建全新的线程上下文，导致Trace ID和主流程脱节。

1. 实现上下文传播的自定义TaskExecutor #

创建包装类，在提交任务时将当前线程的Trace上下文绑定到异步任务线程中：

import io.micrometer.tracing.TraceContextHolder;
import io.micrometer.tracing.context.ContextSnapshot;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.SimpleAsyncTaskExecutor;

public class TraceContextPropagatingTaskExecutor implements TaskExecutor {

    private final TaskExecutor delegate = new SimpleAsyncTaskExecutor();

    @Override
    public void execute(Runnable task) {
        // 捕获当前线程的Trace上下文
        ContextSnapshot snapshot = ContextSnapshot.captureAll();
        // 用上下文包装任务后提交
        delegate.execute(snapshot.wrap(task));
    }
}

2. 配置Spring Batch使用自定义TaskExecutor #

在Job配置类中，将并行步骤的TaskExecutor替换为自定义实现：

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

@Configuration
@EnableBatchProcessing
public class BatchJobConfig {

    @Bean
    public TaskExecutor tracePropagatingTaskExecutor() {
        return new TraceContextPropagatingTaskExecutor();
    }

    // 定义并行步骤
    @Bean
    public Step parallelStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("parallelStep", jobRepository)
                .<YourInput, YourOutput>chunk(100, transactionManager)
                .reader(yourItemReader())
                .processor(yourItemProcessor())
                .writer(yourItemWriter())
                // 指定自定义的TaskExecutor
                .taskExecutor(tracePropagatingTaskExecutor())
                .build();
    }

    // 主流程Job定义（省略其他步骤）
    @Bean
    public Job mainJob(JobRepository jobRepository, Step mainStep, Step parallelStep) {
        return new JobBuilder("mainJob", jobRepository)
                .start(mainStep)
                .next(parallelStep)
                .build();
    }

    // 省略ItemReader/Processor/Writer的定义
}

3. 关键依赖检查 #

确保项目已引入Micrometer Tracing相关依赖（Spring Boot 3默认可能已包含，需确认）：

<!-- Maven示例 -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-tracing</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-tracing-bridge-brave</artifactId> <!-- 或opentelemetry等其他追踪实现 -->
</dependency>

4. 验证效果 #

在并行步骤的处理器或写入器中添加日志，打印当前Trace ID：

import io.micrometer.tracing.TraceContextHolder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class YourItemProcessor implements ItemProcessor<YourInput, YourOutput> {
    private static final Logger log = LoggerFactory.getLogger(YourItemProcessor.class);

    @Override
    public YourOutput process(YourInput item) throws Exception {
        log.info("并行步骤Trace ID: {}", TraceContextHolder.get().traceId());
        // 业务逻辑处理
        return yourOutput;
    }
}

运行任务后，检查日志中主流程和并行步骤的Trace ID是否一致。

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