Java項(xiàng)目中實(shí)現(xiàn)使用traceId跟蹤請(qǐng)求全流程日志

更新時(shí)間：2023年08月28日 08:52:54 作者：小楊Vita

這篇文章主要介紹了Java項(xiàng)目中實(shí)現(xiàn)使用traceId跟蹤請(qǐng)求全流程日志方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教

最近在項(xiàng)目開(kāi)發(fā)中遇到了一些問(wèn)題，項(xiàng)目為多機(jī)部署，使用kibana收集日志，但并發(fā)大時(shí)使用日志定位比較麻煩，大量日志輸出導(dǎo)致很難篩出指定請(qǐng)求的全部相關(guān)日志，以及下游服務(wù)調(diào)用對(duì)應(yīng)的日志。

因此計(jì)劃對(duì)項(xiàng)目日志打印進(jìn)行一些小改造，使用一個(gè)traceId跟蹤請(qǐng)求的全部路徑，前提是不修改原有的打印方式。

簡(jiǎn)單的解決思路

想要跟蹤請(qǐng)求，第一個(gè)想到的就是當(dāng)請(qǐng)求來(lái)時(shí)生成一個(gè)traceId放在ThreadLocal里，然后打印時(shí)去取就行了。

但在不改動(dòng)原有輸出語(yǔ)句的前提下自然需要日志框架的支持了，搜索的一番發(fā)現(xiàn)主流日志框架都提供了MDC功能。

MDC

MDC介紹

MDC（Mapped Diagnostic Context，映射調(diào)試上下文）是 log4j 和 logback 提供的一種方便在多線程條件下記錄日志的功能。
MDC 可以看成是一個(gè)與當(dāng)前線程綁定的Map，可以往其中添加鍵值對(duì)。MDC 中包含的內(nèi)容可以被同一線程中執(zhí)行的代碼所訪問(wèn)。
當(dāng)前線程的子線程會(huì)繼承其父線程中的 MDC 的內(nèi)容。當(dāng)需要記錄日志時(shí)，只需要從 MDC 中獲取所需的信息即可。
MDC 的內(nèi)容則由程序在適當(dāng)?shù)臅r(shí)候保存進(jìn)去。
對(duì)于一個(gè) Web 應(yīng)用來(lái)說(shuō)，通常是在請(qǐng)求被處理的最開(kāi)始保存這些數(shù)據(jù)。

簡(jiǎn)而言之，MDC就是日志框架提供的一個(gè) InheritableThreadLocal ，項(xiàng)目代碼中可以將鍵值對(duì)放入其中，然后使用指定方式取出打印即可。

在 log4j 和 logback 的取值方式為：

%X{traceid}

初步實(shí)現(xiàn)

首先創(chuàng)建攔截器，加入攔截列表中，在請(qǐng)求到達(dá)時(shí)生成traceId。

當(dāng)然你還可以根據(jù)需求在此處后或后續(xù)流程中放入spanId、訂單流水號(hào)等需要打印的信息。

public class Constants {
    /**
     * 日志跟蹤id名。
     */
    public static final String LOG_TRACE_ID = "traceid";
    /**
     * 請(qǐng)求頭跟蹤id名。
     */
    public static final String HTTP_HEADER_TRACE_ID = "app_trace_id";
}

import org.slf4j.MDC;
public class TraceInterceptor extends HandlerInterceptorAdapter {
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        // "traceId"
        MDC.put(Constants.LOG_TRACE_ID, TraceLogUtils.getTraceId());
        return true;
    }
}

然后在日志配置xml文件中添加traceId打印：

<property name="normal-pattern" value="[%p][%d{yyyy-MM-dd'T'HH:mm:ss.SSSZ,Asia/Shanghai}][%X{traceid}][%15.15t][%c:%L] %msg%n"/>

初步改造完成！是不是感覺(jué)還挺簡(jiǎn)單的？且慢，僅僅這樣的改造在實(shí)際使用過(guò)程中會(huì)遇到以下問(wèn)題：

線程池中的線程會(huì)打印錯(cuò)誤的traceId
調(diào)用下游服務(wù)后會(huì)生成新的traceId，無(wú)法繼續(xù)跟蹤

下面來(lái)一一解決這些問(wèn)題。

支持線程池跟蹤

MDC使用的 InheritableThreadLocal 只是在線程被創(chuàng)建時(shí)繼承，但是線程池中的線程是復(fù)用的，后續(xù)請(qǐng)求使用已有的線程將打印出之前請(qǐng)求的traceId。

這時(shí)候就需要對(duì)線程池進(jìn)行一定的包裝，在線程在執(zhí)行時(shí)讀取之前保存的MDC內(nèi)容。

不僅自身業(yè)務(wù)會(huì)用到線程池，spring項(xiàng)目也使用到了很多線程池，比如 @Async 異步調(diào)用，zookeeper線程池、kafka線程池等。

不管是哪種線程池都大都支持傳入指定的線程池實(shí)現(xiàn)，拿 @Async 舉例：

@Bean("SpExecutor")
public Executor getAsyncExecutor() {
    // 對(duì)線程池進(jìn)行包裝，使之支持traceId透?jìng)?
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor() {
        @Override
        public <T> Future<T> submit(Callable<T> task) {
	        // 傳入線程池之前先復(fù)制當(dāng)前線程的MDC
            return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
        @Override
        public void execute(Runnable task) {
            super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
    };
    executor.setCorePoolSize(config.getPoolCoreSize());
    ... // 其他配置
    executor.initialize();
    return executor;
}
public static <T> Callable<T> wrap(final Callable<T> callable, final Map<String, String> context) {
    return new Callable<T>() {
        @Override
        public T call() throws Exception {
	        // 實(shí)際執(zhí)行前導(dǎo)入對(duì)應(yīng)請(qǐng)求的MDC副本
            if (context == null) {
                MDC.clear();
            } else {
                MDC.setContextMap(context);
            }
	        if (MDC.get(Constants.LOG_TRACE_ID) == null) {
	            MDC.put(Constants.LOG_TRACE_ID, TraceLogUtils.getTraceId());
	        }
            try {
                return callable.call();
            } finally {
                MDC.clear();
            }
        }
    };
}

ThreadPoolExecutor 的包裝也類似，注意為了嚴(yán)謹(jǐn)考慮，需要對(duì)連接池中的所有調(diào)用方法進(jìn)行封裝。

在 ThreadPoolExecutor 中有：

public void execute(Runnable command)
public <T> Future<T> submit(Callable<T> task)
public Future<?> submit(Runnable task)
public <T> Future<T> submit(Runnable task, T result)

在 ThreadPoolTaskExecutor 中有：

public void execute(Runnable command)
public void execute(Runnable task, long startTimeout)
public Future<?> submit(Runnable task)
public <T> Future<T> submit(Runnable task, T result)
public <T> ListenableFuture<T> submitListenable(Callable<T> task)
public ListenableFuture<?> submitListenable(Runnable task)

方式與上述的實(shí)現(xiàn)類似，不做贅述。

提供一下我的工具類：

public class ThreadMdcUtil {
    public static void setTraceIdIfAbsent() {
        if (MDC.get(Constants.LOG_TRACE_ID) == null) {
            MDC.put(Constants.LOG_TRACE_ID, TraceLogUtils.getTraceId());
        }
    }
    public static void setTraceId() {
        MDC.put(Constants.LOG_TRACE_ID, TraceLogUtils.getTraceId());
    }
    public static void setTraceId(String traceId) {
        MDC.put(Constants.LOG_TRACE_ID, traceId);
    }
    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();
            }
        };
    }
    public static class ThreadPoolTaskExecutorMdcWrapper extends ThreadPoolTaskExecutor {
        @Override
        public void execute(Runnable task) {
            super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
        @Override
        public void execute(Runnable task, long startTimeout) {
            super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()), startTimeout);
        }
        @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()));
        }
        @Override
        public ListenableFuture<?> submitListenable(Runnable task) {
            return super.submitListenable(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
        @Override
        public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
            return super.submitListenable(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
    }
    public static class ThreadPoolExecutorMdcWrapper extends ThreadPoolExecutor {
        public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                            BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }
        public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        }
        public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                            BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        }
        public ThreadPoolExecutorMdcWrapper(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                            RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        }
        @Override
        public void execute(Runnable task) {
            super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
        @Override
        public <T> Future<T> submit(Runnable task, T result) {
            return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()), result);
        }
        @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()));
        }
    }
    public static class ForkJoinPoolMdcWrapper extends ForkJoinPool {
        public ForkJoinPoolMdcWrapper() {
            super();
        }
        public ForkJoinPoolMdcWrapper(int parallelism) {
            super(parallelism);
        }
        public ForkJoinPoolMdcWrapper(int parallelism, ForkJoinWorkerThreadFactory factory,
                                      Thread.UncaughtExceptionHandler handler, boolean asyncMode) {
            super(parallelism, factory, handler, asyncMode);
        }
        @Override
        public void execute(Runnable task) {
            super.execute(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
        @Override
        public <T> ForkJoinTask<T> submit(Runnable task, T result) {
            return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()), result);
        }
        @Override
        public <T> ForkJoinTask<T> submit(Callable<T> task) {
            return super.submit(ThreadMdcUtil.wrap(task, MDC.getCopyOfContextMap()));
        }
    }
}

下游服務(wù)使用相同traceId

以上方式在多級(jí)服務(wù)調(diào)用中每個(gè)服務(wù)都會(huì)生成新的traceId，導(dǎo)致無(wú)法銜接跟蹤。

這時(shí)就需要對(duì)http調(diào)用工具進(jìn)行相應(yīng)的改造了，在發(fā)送http請(qǐng)求時(shí)自動(dòng)將traceId添加到header中，以RestTemplate為例，注冊(cè)攔截器：

// 以下省略其他相關(guān)配置
RestTemplate restTemplate = new RestTemplate();
// 使用攔截器包裝http header
restTemplate.setInterceptors(new ArrayList<ClientHttpRequestInterceptor>() {
    {
        add((request, body, execution) -> {
            String traceId = MDC.get(Constants.LOG_TRACE_ID);
            if (StringUtils.isNotEmpty(traceId)) {
                request.getHeaders().add(Constants.HTTP_HEADER_TRACE_ID, traceId);
            }
            return execution.execute(request, body);
        });
    }
});
HttpComponentsClientHttpRequestFactory factory = new HttpComponentsClientHttpRequestFactory();
// 注意此處需開(kāi)啟緩存，否則會(huì)報(bào)getBodyInternal方法“getBody not supported”錯(cuò)誤
factory.setBufferRequestBody(true);
restTemplate.setRequestFactory(factory);

下游服務(wù)的攔截器改為：

public class TraceInterceptor extends HandlerInterceptorAdapter {
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        String traceId = request.getHeader(Constants.HTTP_HEADER_TRACE_ID);
        if (StringUtils.isEmpty(traceId)) {
            traceId = TraceLogUtils.getTraceId();
        }
        MDC.put(Constants.LOG_TRACE_ID, traceId);
        return true;
    }
}

若使用自定義的http客戶端，則直接修改其工具類即可。

針對(duì)其他協(xié)議的調(diào)用暫無(wú)實(shí)踐經(jīng)驗(yàn)，可以借鑒上面的思路，通過(guò)攔截器插入特定字段，再在下游讀取指定字段加入MDC中。