MDC实现线程池tranceId全链路传递

Posted 2023-02-15 百里东君~

1、首先定义一个自定义线程池类继承ThreadPoolTaskExecutor

public class ThreadPoolExecutorMdcWrapper extends ThreadPoolTaskExecutor 

    @Override
    public void execute(Runnable task) 
        super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
    


    @Override
    public <T> Future<T> submit(Callable<T> task) 
        return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
    

    @Override
    public Future<?> submit(Runnable task) 
        return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
    

2、编写tranceid包装工具类
判断当前线程对应MDC的上下文是否存在，存在则是子线程，设置MDC中的traceId值，不存在则生成新的tranceid，再执行run方法，执行结束之后清除线程tranceId

public class ThreadMdcUtil 
    public static void setTraceIdIfAbsent() 
        if (MDC.get(LtLogFormat.MDC_TRACEID) == null) 
            MDC.put(LtLogFormat.MDC_TRACEID, UUID.randomUUID().toString().replaceAll("-",""));
        
    

    public static <T> Callable<T> wrap(final Callable<T> callable, final Map<String, String> context) 
        return () -> 
            if (context == null) 
                MDC.clear();
             else 
                MDC.setContextMap(context);
            
            setTraceIdIfAbsent();
            try 
                return callable.call();
             finally 
                MDC.clear();
            
        ;
    

    public static Runnable wrap(final Runnable runnable, final Map<String, String> context) 
        return () -> 
            if (context == null) 
                MDC.clear();
             else 
                MDC.setContextMap(context);
            
            setTraceIdIfAbsent();
            try 
                runnable.run();
             finally 
                MDC.clear();
            
        ;
    

3、初始化自定义线程池

@Configuration
@Component
public class TaskThreadPoolConfig 

	@Bean(name = "optimizeTaskExecutor", destroyMethod = "shutdown")
	public Executor optimizeTaskExecutor() 
		ThreadPoolExecutorMdcWrapper executor = new ThreadPoolExecutorMdcWrapper();
		executor.setCorePoolSize(50);
		executor.setMaxPoolSize(500);
		executor.setQueueCapacity(500);
		executor.setKeepAliveSeconds(60);
		executor.setAllowCoreThreadTimeOut(true);
		executor.setThreadNamePrefix("Feh_Optimize_");
		// 设置等待所有任务执行结束再关闭线程池
		executor.setWaitForTasksToCompleteOnShutdown(true);
		// 该方法用来设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，
		// 以确保应用最后能够被关闭，而不是阻塞住
		executor.setAwaitTerminationSeconds(60);
		// rejection-policy：当pool已经达到max size的时候，如何处理新任务
		// CALLER_RUNS：不在新线程中执行任务，而是由调用者所在的线程来执行
		executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		executor.initialize();
		return executor;
	

4、接下来进入实战环节
编写一个测试类

    @Autowired
    private Executor optimizeTaskExecutor;

    @GetMapping("/testMdcTranceId")
    public Response testMdcTranceId() 
        //    public static final String MDC_TRACEID = "X-B3-TraceId";
        log.info("标记0,tranceId = ", MDC.get(LtLogFormat.MDC_TRACEID));
        CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> 
            test01();
        , optimizeTaskExecutor);
        log.info("标记2,tranceId = ", MDC.get(LtLogFormat.MDC_TRACEID));
        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> 
            test02();
        , optimizeTaskExecutor);
        CompletableFuture.allOf(future1,future2).join();
        log.info("标记4,tranceId = ", MDC.get(LtLogFormat.MDC_TRACEID));
        return Response.buildSuccessResponse();
    

    private void test01()
        log.info("标记1,tranceId = ", MDC.get(LtLogFormat.MDC_TRACEID));
    
    private void test02()
        log.info("标记3,tranceId = ", MDC.get(LtLogFormat.MDC_TRACEID));
    

打印结果tranceid均一致，快动手试试吧