Spring boot整合Skywalking实现线程池无侵入Tid传递及简单源码分析

这里是weihubeats,觉得文章不错可以关注公众号小奏技术，文章首发。拒绝营销号，拒绝标题党

skywalking版本

• 9.4.0

demo源码

• github:github.com/weihubeats/…

背景

继之前skywalking 9.x入门(二) skywalking全链路tid追踪
,发现了一个问题。就是在使用线程池打印log的时候tid会丢失
比如这样

        ExecutorService executorService = Executors.newFixedThreadPool(3);
        executorService.submit(() -> {
            log.info("test log executorService traceId:{}", TraceContext.traceId());
        });

如何实现线程池的tid传递

实际在通过agent对类进行增强的时候我们会发现，线程池中提交任务的常用三个类:Callable、Runnable、Supplier。
但是这三个类是非常核心和基础的类。

一方面这三个类的加载时机比较早，需要在 Runnable 接口及其实现类加载之前启动 Java Agent。总得来说直接增强Callable、Runnable、Supplier这个三个类是非常困难的，所以skywalking选择了新增三个包装类，而不是增强所有的Callable、Runnable、Supplier实现类

这三个类分别是

• CallableWrapper
• RunnableWrapper
• SupplierWrapper

所以我们在线程池的增强的时候一般是如下使用方式

• Case 1

    @TraceCrossThread
    public static class MyCallable implements Callable {
        @Override
        public String call() throws Exception {
            return null;
        }
    }
...
    ExecutorService executorService = Executors.newFixedThreadPool(1);
    executorService.submit(new MyCallable());

• Case 2.

    ExecutorService executorService = Executors.newFixedThreadPool(1);
    executorService.submit(CallableWrapper.of(new Callable() {
        @Override public String call() throws Exception {
            return null;
        }
    }));

or

    ExecutorService executorService = Executors.newFixedThreadPool(1);
    executorService.execute(RunnableWrapper.of(new Runnable() {
        @Override public void run() {
            //your code
        }
    }));

• Case 3

    @TraceCrossThread
    public class MySupplier implements Supplier {
        @Override
        public String get() {
            return null;
        }
    }
...
    CompletableFuture.supplyAsync(new MySupplier());

or

    CompletableFuture.supplyAsync(SupplierWrapper.of(()->{
            return "SupplierWrapper";
    })).thenAccept(System.out::println);

• Case 4

    CompletableFuture.supplyAsync(SupplierWrapper.of(() -> {
        return "SupplierWrapper";
    })).thenAcceptAsync(ConsumerWrapper.of(c -> {
        // your code visit(url)
        System.out.println("ConsumerWrapper");
    }));

or

    CompletableFuture.supplyAsync(SupplierWrapper.of(() -> {
        return "SupplierWrapper";
    })).thenApplyAsync(FunctionWrapper.of(f -> {
        // your code visit(url)
        return "FunctionWrapper";
    }));

官方文档地址:skywalking.apache.org/docs/skywal…

线程池插件 apm-jdk-threadpool-plugin

可以看到上面的使用方式是业务需要改动代码，必须使用RunnableWrapper或CallableWrapper对象

所以基于这个业务场景在Skywalking出现了

issues 8743:github.com/apache/skyw…

旨在增加一个线程池插件apm-jdk-threadpool-plugin，以增强线程池在使用的时候无需使用任何包装对象，业务无需改动任何代码

值得注意的是这个插件不是一个默认插件。所以我们使用该插件需要在打包后手动将该agent.jar移动到skywalking-agent目录

使用apm-jdk-threadpool-plugin插件

我们只需将bootstrap-plugins插件下面的apm-jdk-threadpool-plugin-8.16.0-SNAPSHOT.jar插件copy到plugins目录即可

试试效果

启动脚本

-javaagent:/Users/xiaozoujishu/Downloads/skywalking-agent/skywalking-agent.jar
-DSW_AGENT_NAME=order-service
-DSW_AGENT_COLLECTOR_BACKEND_SERVICES=127.0.0.1:11800

实际效果

可以看到线程池原生使用方式是增强成功了，我们不用改动任何代码

源码分析

我们可以简单分析下apm-jdk-threadpool-plugin插件的源码

首先核心增强的类就是

private static final String ENHANCE_CLASS = "java.util.concurrent.ThreadPoolExecutor";

增强的方法就是

    private static final String INTERCEPT_EXECUTE_METHOD = "execute";

    private static final String INTERCEPT_SUBMIT_METHOD = "submit";

实现增强类的逻辑就是

    private static final String INTERCEPT_EXECUTE_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolExecuteMethodInterceptor";

    private static final String INTERCEPT_SUBMIT_METHOD_HANDLE = "org.apache.skywalking.apm.plugin.ThreadPoolSubmitMethodInterceptor";

主要是这两个类

我们随便看一个类ThreadPoolExecuteMethodInterceptor吧，另一个也是类似的

核心就是将Runnable替换为SwRunnableWrapper，而SwRunnableWrapper里面就进行了trace传递获取

核心代码

总结

总的来说如果线程池我们要支持无缝tid传递，需要将bootstrap-plugins中的apm-jdk-threadpool-plugin-8.16.0-SNAPSHOT.jar插件移动到我们的agent中，才会生效。

但是仅仅支持线程池，即ThreadPoolExecutor