异步线程中链路追踪方案

在我的博客阅读本文

• 1. 解决的问题
• 2. 环境
• 3. Brave的currentTraceContext
• 4. 用Brave提供的API实现
• • 4.1. Runnable使用
  • 4.2. Spring线程池ThreadPoolTaskExecutor中使用

当一个线程执行过程中开启了新的异步线程，会导致异步线程与当前线程的traceId不一致的问题。

在线程池中，traceId可能在线程创建那一刻就已经固定了，不会跟着使用场景上下文traceId变动，在后面的线程复用环节中一直都是这个traceId，会带来traceId混乱在一起的情况，同样也会带来异步线程与当前线程的traceId不一致的问题。

最终，导致异步线程在日志上无法准确追踪到整个调用链路。

总的来说基于 Zipkin server + Brave library，这一块基本使用需要参考一下文档：

https://zipkin.io/pages/instrumenting.html

https://github.com/openzipkin/brave

以MDCCurrentTraceContext作为上下文容器进行初始化方式为例：

var Tracing = Tracing.newBuilder().endpoint(endpoint)
   .spanReporter(spanReporter()).currentTraceContext(MDCCurrentTraceContext.create()).build();

当调用tracing#Tracer#nextSpan()来开启一个新的执行片段（Span）时：

  public Span nextSpan() {
    TraceContext parent = currentTraceContext.get();
    return parent != null ? newChild(parent) : newTrace();
  }

本质上是从currentTraceContext中取出TraceContext再进行后续操作,而TraceContext中又有我们需要的traceId：TraceContext#traceIdString

因此，认为只要将这个currentTraceContext从当前线程“传递”到异步线程中就可以满足需求。

我这里追了下代码，MDCCurrentTraceContext是通过ThreadLocal（准确的说是InheritableThreadLocal）绑定到线程中的，TraceContext有多种实现，也有可能有其他的方式。猜想作为框架的Brave其实应该提供一个方法来统一处理这种需求。

Brave提供了以下API来装饰线程（池），帮我们做这个“传递”的动作：

• brave.propagation.CurrentTraceContext#wrap(java.lang.Runnable)
• brave.propagation.CurrentTraceContext#wrap(java.util.concurrent.Callable)
• brave.propagation.CurrentTraceContext#executor
• brave.propagation.CurrentTraceContext#executorService

下文是几个Demo代码。

// Ignore DI Tracing
// Ignore DI THREAD_POOL
CompletableFuture smartFuture = CompletableFuture.runAsync(tracing.currentTraceContext().wrap(() -> {
    var tracer = tracing.tracer();
    var span = getNextSpan(tracer, "spanName");
    try (var ignored = tracer.withSpanInScope(span)) {
        // biz code
    } catch (Exception e) {
        span.error(e);
        throw e;
    } finally {
        span.finish();
    }
}), THREAD_POOL);

// Ignore DI Tracing
@Bean
public ThreadPoolTaskExecutor getThreadPoolTaskExecutor() {
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(20);
    threadPoolTaskExecutor.setMaxPoolSize(100);
    threadPoolTaskExecutor.setQueueCapacity(100);
    threadPoolTaskExecutor.setKeepAliveSeconds(60);
    threadPoolTaskExecutor.setThreadNamePrefix("thread-prefix");
    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    // decorate runnable used in thread pool
    threadPoolTaskExecutor.setTaskDecorator(tracing.currentTraceContext()::wrap);
    return threadPoolTaskExecutor;
}

Java自带的ThreadPoolExecutor也是一样的，重写即可，方法同Spring的ThreadPoolTaskExecutor类中的setTaskDecorator方法。