日志追踪对于接口故障排查非常重要，可以有效、快捷的定位故障点，但在多线程环境中，若没有相关框架的支持，想要实现日志追踪，就需要编码实现将主线程的日志参数传递给子线程，本文就在线程池场景下借助MDC实现了traceId参数的透传

1 MDC

1.1 简介

MDC（Mapped Diagnostic Context，映射调试上下文）是 log4j 和 logback 提供的一种方便在多线程条件下记录日志的功能。某些应用程序采用多线程的方式来处理多个用户的请求。在一个用户的使用过程中，可能有多个不同的线程来进行处理。典型的例子是Web 应用服务器。当用户访问某个页面时，应用服务器可能会创建一个新的线程来处理该请求，也可能从线程池中复用已有的线程。在一个用户的会话存续期间，可能有多个线程处理过该用户的请求。这使得比较难以区分不同用户所对应的日志。当需要追踪某个用户在系统中的相关日志记录时，就会变得很麻烦。

一种解决的办法是采用自定义的日志格式，把用户的信息采用某种方式编码在日志记录中。这种方式的问题在于要求在每个使用日志记录器的类中，都可以访问到用户相关的信息。这样才可能在记录日志时使用。这样的条件通常是比较难以满足的。MDC 的作用是解决这个问题。MDC 可以看成是一个与当前线程绑定的哈希表，可以往其中添加键值对。MDC 中包含的内容可以被同一线程中执行的代码所访问。当前线程的子线程会继承其父线程中的 MDC 的内容。当需要记录日志时，只需要从 MDC 中获取所需的信息即可。MDC 的内容则由程序在适当的时候保存进去。对于一个 Web 应用来说，通常是在请求被处理的最开始保存这些数据

1.2 MDC坐标和使用

<dependency>
    <groupId>log4j</groupId>
    <artifactId>log4j</artifactId>
    <version>1.2.17</version>
  </dependency>
  <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>1.7.21</version>
  </dependency>

log4j.xml配置样例，追踪日志自定义格式主要在name="traceId"的layout里面进行设置，我们使用%X{traceId}来定义此处会打印MDC里面key为traceId的value，如果所定义的字段在MDC不存在对应的key，那么将不会打印，会留一个占位符
点击了解Loback.xml文件解释

1.3 主要方法

API 说明：

• clear()：移除所有MDC
• get (String key)：获取当前线程 MDC 中指定 key 的值
• getCopyOfContextMap()：将MDC从内存获取出来，再传给线程
• put(String key, Object o)：往当前线程的 MDC 中存入指定的键值对
• remove(String key)：删除当前线程 MDC 中指定的键值对
• setContextMap()：将父线程的MDC内容传给子线程

MDC异步线程间传递：
用MDC的put时，子线程在创建的时候会把父线程中的inheritableThreadLocals变量设置到子线程的inheritableThreadLocals中，而MDC内部是用InheritableThreadLocal实现的，所以会把父线程中的上下文带到子线程中
但在线程池中，由于线程会被重用，但是线程本身只会初始化一次，所以之后重用线程的时候，就不会进行初始化操作了，也就不会有父线程inheritableThreadLocals拷贝到子线程中的过程了，这个时候如果还想传递父线程的上下文的话，就要使用getCopyOfContextMap方法

2 多线程间使用

2.1 MDC工具类

定义MDC工具类，支持Runnable和Callable两种，目的就是为了把父线程的traceId设置给子线程

import org.slf4j.MDC;
import org.springframework.util.CollectionUtils;

import java.util.Map;
import java.util.concurrent.Callable;

/**
 * @Description 封装MDC用于向线程池传递
 */
public class MDCUtil {
    // 设置MDC中的traceId值，不存在则新生成，针对不是子线程的情况，
    // 如果是子线程，MDC中traceId不为null
    public static void setTraceIdIfAbsent() {
        if (MDC.get(Constants.TRACE_ID) == null) {
            MDC.put(Constants.TRACE_ID, TraceIdUtil.getTraceId());
        }
    }
    public static <T> Callable<T> wrap(final Callable<T> callable, final Map<String, String> context) {
        return () -> {
            if (CollectionUtils.isEmpty(context)) {
                MDC.clear();
            } else {
                MDC.setContextMap(context);
            }
            setTraceIdIfAbsent();
            try {
                return callable.call();
            } finally {//清除子线程的，避免内存溢出，就和ThreadLocal.remove()一个原因
                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();
            }
        };
    }

    public static void setMDCContextMap(final Map<String, String> context) {
        if (CollectionUtils.isEmpty(context)) {
            MDC.clear();
        } else {
            MDC.setContextMap(context);
        }
    }

}

2.2 拦截器定义和配置

package demo;

import org.slf4j.MDC;
import org.springframework.lang.Nullable;
import org.springframework.util.StringUtils;
import org.springframework.web.servlet.handler.HandlerInterceptorAdapter;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.Arrays;
import java.util.List;

public class RequestInterceptor extends HandlerInterceptorAdapter {

    private static final List<String> paramSet = Arrays.asList("traceId");

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        this.setParam(request);
        return super.preHandle(request, response, handler);
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, @Nullable Exception ex) throws Exception {
        MDC.clear();
    }

    private void setParam(HttpServletRequest request) {
        // 设置要放到MDC中的参数
        for (String key : paramSet) {
            String val = request.getHeader(key);
            if (!StringUtils.isEmpty(val)) {
                MDC.put(key, val);
            }
        }
    }

}

拦截器配置

import demo.RequestInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

/**
 * 拦截WEB请求
 */
@Configuration
public class InterceptorConfig implements WebMvcConfigurer {

    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new RequestInterceptor());
    }

}

2.3 Java线程池中使用

2.3.1 配置线程池

@Configuration
public class ThreadPoolService {
    @Bean
    public ThreadPoolExecutor threadPoolExecutor() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor
                (4, 8, 10,
                TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(5536),
                new ScheduledThreadFactory("demo-"), new ThreadPoolExecutor.CallerRunsPolicy());
        return threadPoolExecutor;
    }   
}

点击了解线程池相关信息

2.3.2 使用ExecutorCompletionService方式

使用ExecutorCompletionService实现多线程调用
点击了解更多关于ExecutorCompletionService信息

/**
 * 使用MDC传递traceId
 */
public class Demo {

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    public void demo() {
        ExecutorCompletionService ecs = new ExecutorCompletionService(threadPoolExecutor);
        ecs.submit(MDCUtil.wrap(new TestMDC(), MDC.getCopyOfContextMap()));
    }
    
    class TestMDC implements Callable {
        @Override
        public Object call() throws Exception {
            // TODO 代码逻辑
            return null;
        }
    }
}

2.3.3 使用CompletableFuture方式

使用CompletableFuture实现多线程调用，其中收集CompletableFuture运行结果，
点击了解更多关于CompletableFuture信息

public class Result {}
/**
 * 使用MDC传递traceId
 */
public class Demo {

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    private CompletableFuture<Result> test() {
    
        Map<String, String> copyOfContextMap = MDC.getCopyOfContextMap();
        
        return CompletableFuture.supplyAsync(() -> {
        
            // 必须在打印日志前设置
            MDCUtil.setMDCContextMap(copyOfContextMap);
            //MDC.put("subTraceId",''); //如果需要对子线程进行加线程跟踪号，可在此处设定
            // TODO 业务逻辑
            return new Result();
            
        }, threadPoolExecutor).exceptionally(new Function<Throwable, Result>() {
            /**捕捉异常,不会导致整个流程中断**/
            @Override
            public Result apply(Throwable throwable) {
                log.error("线程[{}]发生了异常[{}], 继续执行其他线程", Thread.currentThread().getName(), throwable.getMessage());
                return null;
            }
        });
    }
}

2.4 Spring线程池中使用

2.4.1 继承ThreadPoolTaskExecutor

public class ThreadPoolMdcWrapper extends ThreadPoolTaskExecutor {

    public ThreadPoolMdcWrapper() {

    }

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

    @Override
    public void execute(Runnable task, long startTimeout) {
        super.execute(MDCUtil.wrap(task, MDC.getCopyOfContextMap()), startTimeout);
    }

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

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

    @Override
    public ListenableFuture<?> submitListenable(Runnable task) {
        return super.submitListenable(MDCUtil.wrap(task, MDC.getCopyOfContextMap()));
    }

    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
        return super.submitListenable(MDCUtil.wrap(task, MDC.getCopyOfContextMap()));
    }
}

2.4.2 配置线程池

继承ThreadPoolTaskExecutor ，重写线程执行的方法。到这我们就做完了大部分的准备工作，还剩下最关键的就是让程序用到我们封装后的线程池。我们可以在声明线程池的时候，直接使用我们封装好的线程池（因为继承了ThreadPoolTaskExecutor）
点击了解Spring线程池配置

@Bean
public ThreadPoolTaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolMdcWrapper();
    //核心线程数，默认为1
    taskExecutor.setCorePoolSize(1);
    //最大线程数，默认为Integer.MAX_VALUE
    taskExecutor.setMaxPoolSize(200);
    //队列最大长度，一般需要设置值>=notifyScheduledMainExecutor.maxNum；默认为Integer.MAX_VALUE
    taskExecutor.setQueueCapacity(2000);
    //线程池维护线程所允许的空闲时间，默认为60s
    taskExecutor.setKeepAliveSeconds(60);
    //线程池对拒绝任务（无线程可用）的处理策略
    taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    // 初始化线程池
    taskExecutor.initialize();
    return  taskExecutor;
}

到这我们所做的准备工作，改造工作也就结束了，剩下的就是使用了。只要在程序异步调用时，利用声明好的taskExecutor线程池进行调用，就可以在线程上下文正确传递Traceid了

2.5 异步线程 AsyncConfigurer

继承了Spring的AsyncConfigurer，并重写了getAsyncExecutor方法,这样在Spring中使用@Async注解开启异步线程,会自动传递MDC信息给子线程,

另外关于异步线程的异常捕获,先列举一下一般开启异步的方式:

• 使用Spring的@Async注解开启异步
• 通过executor.execute开启异步
• 通过executor.submit开启异步
• 通过CompletableFuture开启异步

下面针对异步子线程的异常捕获提供几种解决方案：

• 重写AsyncConfigurer的getAsyncUncaughtExceptionHandler方法,这种方式只能捕获方式A开启的异步
• 使用Future.get(),可以捕获方式C开启的异步
• 使用Completable.join()或者Completable.get(),可以捕获方式D开启的异步
• 重写getAsyncExecutor方法时,在runnable.run()代码块上使用try/catch,可以捕获方式A,B,C开启的异步
• 使用try/catch包裹整个runnable函数式接口,这样可以捕获A,B,C,D开启的异步

    executor.execute(() -> {
            try {
                //需要开启异步的业务逻辑方法或者代码块
                xxx();
            } catch (Throwable e) {
                log.error("异常", e);
            }
        });

下面给出完整的代码

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
 
@Slf4j
@EnableAsync
@Configuration
@RequiredArgsConstructor
public class ThreadPoolTaskConfig implements AsyncConfigurer {
 
    @Bean("AsyncExecutor")
    @Override
    public ThreadPoolTaskExecutor getAsyncExecutor() {
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor() {
            /**
             * 所有线程都会委托给这个execute方法，在这个方法中我们把父线程的MDC内容赋值给子线程
             * https://logback.qos.ch/manual/mdc.html#managedThreads
             *
             * @param runnable runnable
             */
            @Override
            public void execute(Runnable runnable) {
                // 获取父线程MDC中的内容，必须在run方法之前，否则等异步线程执行的时候有可能MDC里面的值已经被清空了，这个时候就会返回null
                Map<String, String> context = MDC.getCopyOfContextMap();
                super.execute(() -> {
                    // 将父线程的MDC内容传给子线程
                    if (context != null) {
                        MDC.setContextMap(context);
                    }
                    try {
                        // 执行异步操作
                        runnable.run();
                    } catch (Throwable e) {
                        log.info("异步线程执行异常:{}", e.getMessage(), e);
                        //替换成业务异常
                        throw new RuntimeException("异步线程执行异常");
                    } finally {
                        // 清空MDC内容
                        MDC.clear();
                    }
                });
            }
 
            @Override
            public <T> Future<T> submit(Callable<T> task) {
                // 获取父线程MDC中的内容，必须在run方法之前，否则等异步线程执行的时候有可能MDC里面的值已经被清空了，这个时候就会返回null
                Map<String, String> context = MDC.getCopyOfContextMap();
                return super.submit(() -> {
                    // 将父线程的MDC内容传给子线程
                    if (context != null) {
                        MDC.setContextMap(context);
                    }
                    try {
                        // 执行异步操作
                        return task.call();
                    } catch (Throwable e) {
                        log.info("异步线程执行异常:{}", e.getMessage(), e);
                        //替换成业务异常
                        throw new RuntimeException("异步线程执行异常");
                    } finally {
                        // 清空MDC内容
                        MDC.clear();
                    }
                });
            }
        };
        ;
        // 设置核心线程数
        threadPoolTaskExecutor.setCorePoolSize(30);
        // 设置最大线程数
        threadPoolTaskExecutor.setMaxPoolSize(50);
        // 设置队列容量
        threadPoolTaskExecutor.setQueueCapacity(1000);
        // 设置线程活跃时间（秒）
        threadPoolTaskExecutor.setKeepAliveSeconds(60);
        // 设置拒绝策略
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 设置线程池终止等待时间
        threadPoolTaskExecutor.setAwaitTerminationSeconds(10);
 
        threadPoolTaskExecutor.initialize();
        return threadPoolTaskExecutor;
    }
 
    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return (Throwable throwable, Method method, Object... objects) -> {
            log.error("AsyncUncaughtExceptionHandler: ", throwable);
            log.info("method: {}", method.getName());
            log.info("objects: {}", objects);
        };
    }
}

调用测试

@Slf4j
@RestController
public class UserController {
    @Autowired
    @Qualifier("asyncExe")
    private Executor executor;
    @Autowired
    private AsyncServiceImpl ayncService;
    @GetMapping("/t1")
    public void test1(){
        log.info("开始....");
        CompletableFuture.runAsync(() ->{
            log.info("异步中....");
        });
        executor.execute(() ->{
            log.info("线程池中....");
        });
        ayncService.test();
        log.info("结束....");
    }
}   

@Slf4j
@Service
public class AsyncServiceImpl {
    @Async("asyncExe")
    public void test(){
        //...具体业务逻辑
        log.info("异步async中....");
    }
}

2.6 多线程间传递 TransmittableThreadLocal

2.6.1 引言

假如使用logback/log4j官网推荐的方案，显示调用 MDC.getCopyOfContextMap() 和 MDC.setContextMap() ，在向线程池提交任务的时候需要显示的去调用。这种方式很繁琐，而且侵入性很高，可维护性也很低。

如果使用阿里的TransmittableThreadLocal方案，是使用TransmittableThreadLocal的实现去增强ThreadPoolExecutor，不需要在任务提交运行的时候去显示的调用MDC，但是TransmittableThreadLocal的官网上没有明确的结合MDC的教程。
主要有2种，一种是自己实现一个MDCAdapter替换logback/log4j的MDCAdapter，内部将其ThreadLocal替换为TransmittableThreadLocal的实现，在通过其他方式注入到日志框架中。
另外一种方式是使用 logback-mdc-ttl 来更换项目中的logback框架，内部的思路和上面类似，也是替换了MDCAdapter的实现。
但是这2种方式都有很大的问题，第一种需要修改日志框架的注入实现，在后续升级日志框架有很大的风险。第二种方式是引入了一个三方的日志框架，不可维护。

2.6.2 解决方案

总结来看上述几种解决方案都不太理解，第二种方式虽然使用了TransmittableThreadLocal解决了包装类的问题，但是没有很好的适配MDC，修改了大量的实现代码，而且不利于后续的升级维护。
在搜索的相关的资料、源码以及TransmittableThreadLocal的issue里，发现了一种比较简洁的实现方式。

添加 HandlerInterceptor 拦截器，核心的实现思路是实现 TransmittableThreadLocal 的 initialValue,beforeExecute,afterExecute接口，在多线程数据传递的时候，将数据复制一份给MDC。

@Component
public class TraceIdInterceptor implements HandlerInterceptor {

    /**
     * 实现 TransmittableThreadLocal 的 initialValue,beforeExecute,afterExecute接口
     */
    static TransmittableThreadLocal<Map<String, String>> ttlMDC = new TransmittableThreadLocal<>() {
        /**
         * 在多线程数据传递的时候，将数据复制一份给MDC
         */
        @Override
        protected void beforeExecute() {
            final Map<String, String> mdc = get();
            mdc.forEach(MDC::put);
        }

        @Override
        protected void afterExecute() {
            MDC.clear();
        }

        @Override
        protected Map<String, String> initialValue() {
            return Maps.newHashMap();
        }
    };


    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        //MDC记录traceId
        String traceId = IdUtil.fastUUID();
        MDC.put("traceId", traceId);

        //同时给TransmittableThreadLocal记录traceId
        ttlMDC.get().put("traceId", traceId);

        return true;
    }

    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler,
        @Nullable Exception ex) {

        //清除数据
        MDC.clear();
        ttlMDC.get().clear();
        ttlMDC.remove();
    }
}

使用 TransmittableThreadLocal 提供的包装池，

@Bean
public Executor asyncExecutor() {
    log.info("start asyncExecutor");
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    //配置核心线程数
    executor.setCorePoolSize(10);
    //配置最大线程数
    executor.setMaxPoolSize(50);
    //配置队列大小
    executor.setQueueCapacity(0);
    //配置线程池中的线程的名称前缀
    executor.setThreadNamePrefix("async-service-");

    // rejection-policy：当pool已经达到max size的时候，如何处理新任务
    // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    //执行初始化
    executor.initialize();
    //使用TransmittableThreadLocal提供的包装池
    return TtlExecutors.getTtlExecutor(executor);
}

2.7 HTTP调用丢失traceId

在使用 HTTP 调用第三方服务接口时traceId将丢失，需要对HTTP调用工具进行改造，在发送时在request header中添加traceId，在下层被调用方添加拦截器获取header中的traceId添加到MDC中

HTTP调用有多种方式，比较常见的有HttpClient、OKHttp、RestTemplate，所以只给出这几种HTTP调用的解决方式

2.7.1 HttpClient

实现HttpClient拦截器

public class HttpClientTraceIdInterceptor implements HttpRequestInterceptor {
    @Override
    public void process(HttpRequest httpRequest, HttpContext httpContext) throws HttpException, IOException {
        String traceId = MDC.get(Constants.TRACE_ID);
        //当前线程调用中有traceId，则将该traceId进行透传
        if (traceId != null) {
            //添加请求体
            httpRequest.addHeader(Constants.TRACE_ID, traceId);
        }
    }
}

实现 HttpRequestInterceptor接口并重写process方法
如果调用线程中含有traceId，则需要将获取到的traceId通过request中的header向下透传下去

为HttpClient添加拦截器

private static CloseableHttpClient httpClient = HttpClientBuilder.create()
            .addInterceptorFirst(new HttpClientTraceIdInterceptor())
            .build();

通过addInterceptorFirst方法为HttpClient添加拦截器

2.7.2 OKHttp

实现OKHttp拦截器

public class OkHttpTraceIdInterceptor implements Interceptor {
    @Override
    public Response intercept(Chain chain) throws IOException {
        String traceId = MDC.get(Constants.TRACE_ID);
        Request request = null;
        if (traceId != null) {
            //添加请求体
            request = chain.request().newBuilder().addHeader(Constants.TRACE_ID, traceId).build();
        }
        Response originResponse = chain.proceed(request);

        return originResponse;
    }
}

实现Interceptor拦截器，重写interceptor方法，实现逻辑和HttpClient差不多，如果能够获取到当前线程的traceId则向下透传

为OkHttp添加拦截器，调用addNetworkInterceptor方法添加拦截器

private static OkHttpClient client = new OkHttpClient.Builder()
            .addNetworkInterceptor(new OkHttpTraceIdInterceptor())
            .build();

2.7.3 RestTemplate

实现RestTemplate拦截器

public class RestTemplateTraceIdInterceptor implements ClientHttpRequestInterceptor {
    @Override
    public ClientHttpResponse intercept(HttpRequest httpRequest, byte[] bytes, ClientHttpRequestExecution clientHttpRequestExecution) throws IOException {
        String traceId = MDC.get(Constants.TRACE_ID);
        if (traceId != null) {
            httpRequest.getHeaders().add(Constants.TRACE_ID, traceId);
        }

        return clientHttpRequestExecution.execute(httpRequest, bytes);
    }
}

实现ClientHttpRequestInterceptor接口，并重写intercept方法，其余逻辑都是一样的不重复说明

为RestTemplate添加拦截器，调用setInterceptors方法添加拦截器

restTemplate.setInterceptors(Arrays.asList(new RestTemplateTraceIdInterceptor()));