源码角度了解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);
}
}
- 创建新的TracingContext,创建LocalSpan
- 判断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中。
❤️ 感谢大家
如果你觉得这篇内容对你挺有有帮助的话:
- 欢迎关注我❤️,点赞 标签:objInst,Trace,ContextManager,源码,线程,Skywalking,public From: https://blog.51cto.com/u_15460453/5735179