源码角度了解Skywalking之Trace可以跨线程吗

Trace信息是一个重要的信息，那么Skywalking的trace可以跨线程传播吗？

我们先给出答案，它是是可以的跨线程传播的，今天就带大家看一下Trace跨线程的使用和实现原理

使用

private static final ExecutorService SERVICE = Executors.newSingleThreadExecutor(r -> {
    Thread thread = new Thread(r);
    thread.setDaemon(true);
    return thread;
});

public void asyncRunnable(Runnable runnable) {
    SERVICE.submit(RunnableWrapper.of(runnable));
}

public void asyncCallable(Callable<Boolean> callable) {
    SERVICE.submit(CallableWrapper.of(callable));
}
public void asyncSupplier(Supplier<Boolean> supplier) {
    CompletableFuture.supplyAsync(SupplierWrapper.of(supplier));
}

定义单个线程池

RunnableWrapper包装Runnable对象实现Trace的跨线程

CallableWrapper包装Callable接口对象来实现Trace的跨线程

SupplierWrapper包装Supplier对象实现Trace的跨线程

这样Skywalking就可以捕捉到线程的trace信息了，那么它是怎么实现的呢？

实现原理

通过看Skywalking的apm-application-toolkit模块，可以看到RunnableWrapper、CallableWrapper类和SupplierWrapper类都有一个特点，那就是类上都有一个注解@TraceCrossThread，这个注解显然是Trace跨线程的意思

谁会扫描@TraceCrossThread注解呢？

apm-toolkit-trace-activation模块下的CallableOrRunnableActivation会对标注@TraceCrossThread注解的的类和call()方法、run()方法或者get()的方法进行拦截增强，对应拦截器为CallableOrRunnableInvokeInterceptor

CallableOrRunnableConstructInterceptor

CallableOrRunnableActivation还定义了构造方法的切入点，对标注@TraceCrossThread注解的类的任何构造方法进行拦截，拦截器为CallableOrRunnableConstructInterceptor，看一下这个拦截器的增强逻辑

CallableOrRunnableConstructInterceptor类：

public class CallableOrRunnableConstructInterceptor implements InstanceConstructorInterceptor {
    @Override
    public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
        if (ContextManager.isActive()) {
            objInst.setSkyWalkingDynamicField(ContextManager.capture());
        }
    }

}

代码比较简答，也就是将当前 TracingContext 的核心信息填充到 ContextSnapshot 中，并添加到_$EnhancedClassField_ws 字段中，接下来实例方法的增强会用到这个信息。

CallableOrRunnableInvokeInterceptor

beforeMethod()方法

CallableOrRunnableInvokeInterceptor的beforeMethod()方法：

public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    MethodInterceptResult result) throws Throwable {
    ContextManager.createLocalSpan("Thread/" + objInst.getClass().getName() + "/" + method.getName());
    ContextSnapshot cachedObjects = (ContextSnapshot)objInst.getSkyWalkingDynamicField();
    if (cachedObjects != null) {
        ContextManager.continued(cachedObjects);
    }
}

1. 创建新的TracingContext，创建LocalSpan
2. 判断ContextSnapshot对象是否为空，如果不为空，调用ContextManager.continued()方法在此段和跨线程段之间建立引用，填充TracingContext信息。

afterMethod()方法

CallableOrRunnableInvokeInterceptor的afterMethod()方法：

@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
    Object ret) throws Throwable {
    ContextManager.stopSpan();
    // clear ContextSnapshot
    objInst.setSkyWalkingDynamicField(null);
    return ret;
}

关闭span，清空ContextSnapshot信息，防止内存泄露的情况。

总结

这篇文章我们介绍了Skywalking的Trace信息是怎么实现跨线程的，核心逻辑是CallableOrRunnableActivation对标注@TraceCrossThread注解的类的构造方法和get()方法、run()方法和call()方法进行拦截增强，构造方法拦截增强的逻辑由CallableOrRunnableConstructInterceptor实现，主要的是向_EnhancedClassField_ws 字段中添加当前Trace的信息，在run()方法等方法执行前拦截器CallableOrRunnableInvokeInterceptor进行拦截，创建新的TracingContext和LocalSpan，判断_EnhancedClassField_ws字段中是否有相关trace信息，如果有就添加到新建的TracingContext中。