Tomcat容器应用中使用CompletableFuture时，

作者: 我有一只喵喵 | 来源:发表于2020-12-23 21:34 被阅读0次

背景

有一个功能，这个功能里需要调用几个不同的RPC请求，一开始不以为然，没觉得什么，所以所有的RPC请求都是串行执行，后来发现部分RPC返回时间比较长导致此功能接口时间耗时较长，于是乎就使用了JDK8新特性CompletableFuture打算将这些不同的RPC请求异步执行，等所有的RPC请求结束后，再返回请求结果。

因为功能比较简单没什么特殊的，所以这里在使用CompletableFuture的时候，并没有自定义线程池，默认那么就是ForkJoinPool。下面看下伪代码：

        CompletableFuture task1 = CompletableFuture.runAsync(()->{
            /**
             * 这里会调用一个RPC请求，而这个RPC请求处理的过程中会通过SPL机制load指定接口的实现，这个接口所在jar存在于WEB-INFO/lib
             */
            System.out.println("任务1执行");
        });

        CompletableFuture task2 = CompletableFuture.runAsync(()->{
            System.out.println("任务2执行");
        });

        CompletableFuture task3 = CompletableFuture.runAsync(()->{
            System.out.println("任务3执行");
        });

        // 等待所以任务执行完成返回
        CompletableFuture.allOf(task1,task2,task3).join();

        return result;

其实初步上看，这段代码没什么特别的，每个任务都是调用一个RPC请求。初期测试这段代码的时候是通过IDEA启动项目，也就是用的是 SpringBoot 内嵌 Tomcat启动的，这段代码功能正常。然后呢，代码开始commit,merge。

到了第二天之后，同事测试发现这段代码抛出了异常，而且这个功能是主入口，那么就是说大大的阻塞啊，此时我心里心情是这样的

立马上后台看日志，但是却发现这个异常是RPC内部处理时抛出来的，第一反应那就是找上游服务提供方，问他们是不是改接口啦？准备开始甩锅！

然后结果就是没有！！！于是乎我又跑了下项目，测试了一下接口，没问题！确实没问题！卧槽？？？还有更奇怪的事情，那就是同时装了好几套环境，其他环境是没问题的，此时就没再去关注，后来发现只有在重启了服务器之后，这个问题就会作为必现问题，着实头疼。

问题定位

到这里只能老老实实去debug RPC调用过程的源码了。也就是代码示例中写的，RPC调用过程中，会使用ServiceLoader去找XX接口对应的实现类，而这个配置是在RPC框架的jar包中，这个jar包那自然肯定是在对应微服务的WEB-INFO/lib里了。

这段源码大概长这样吧：

       ArrayList list = new ArrayList<String>();
        ServiceLoader<T> serviceLoader = ServiceLoader.load(xxx interface);
        serviceLoader.forEach(xxx->{
            list.add(xxx)
        });

这步执行完后，如果list是空的，那就会抛个异常，这个异常就是前面所说RPC调用过程中的异常了。

到这里，加载不到，那就要怀疑ClassLoader了，先看下ClassLoader加载范围

Bootstrap ClassLoader
%JRE_HOME%\lib 下的 rt.jar、resources.jar、charsets.jar 和 class
ExtClassLoader
%JRE_HOME%\lib\ext 目录下的jar包和class
AppClassLoader
当前应用ClassPath指定的路径中的类
ParallelWebappClassLoader
这个就属于Tomcat自定义ClassLoader了，可以加载当前应用下WEB-INFO/lib

再看下ServiceLoader的实现：

    public static <S> ServiceLoader<S> load(Class<S> service) {
        ClassLoader cl = Thread.currentThread().getContextClassLoader();
        return ServiceLoader.load(service, cl);
    }

    public static <S> ServiceLoader<S> load(Class<S> service,
                                            ClassLoader loader)
    {
        return new ServiceLoader<>(service, loader);
    }

    private ServiceLoader(Class<S> svc, ClassLoader cl) {
        service = Objects.requireNonNull(svc, "Service interface cannot be null");
        loader = (cl == null) ? ClassLoader.getSystemClassLoader() : cl;
        acc = (System.getSecurityManager() != null) ? AccessController.getContext() : null;
        reload();
    }

调用load的时候，先获取当前线程的上下文ClassLoader，然后调用new，进入到ServiceLoader的私有构造方法中，这里重点有一句 loader = (cl == null) ? ClassLoader.getSystemClassLoader() : cl; ，如果传入的classLoader是null(null就代表是BootStrapClassLoader)，就使用ClassLoader.getSystemClassLoader()，其实就是AppClassLoader了。

然后就要确定下执行ServiceLoader.load方法时，最终ServiceLoader的loader到底是啥？

1.Debug 通过Sring Boot 内嵌Tomcat启动的应用
在这种情况下ClassLoader是org.springframework.boot.web.embedded.tomcat.TomcatEmbeddedWebappClassLoader
2.Debug 通过Tomcat启动的应用
在这种情况下ClassLoader是AppClassLoader，通过Thread.currentThread().getContextClassLoader()获取到的是null

真相已经快要接近，为啥同样的代码，Tomcat应用启动的获取到的线程当前上下文类加载器却是BootStrapClassLoader呢？

问题就在于CompletableFuture.runAsync这里，这里并没有显示指定Executor，所以会使用ForkJoinPool线程池，而ForkJoinPool中的线程不会继承父线程的ClassLoader。enmm,很奇妙，为啥不继承，也不知道。。。

问题印证

下面通过例子来证实下，先从基本的看下，这里主要是看子线程会不会继承父线程的上下文ClassLoader，先自定义一个ClassLoader,更加直观：

class MyClassLoader extends ClassLoader{
    
}

测试一

    private static void test1(){
        MyClassLoader myClassLoader = new MyClassLoader();

        Thread.currentThread().setContextClassLoader(myClassLoader);

        // 创建一个新线程
       new Thread(()->{
           System.out.println( Thread.currentThread().getContextClassLoader());
       }).start();

    }

输出

classloader.MyClassLoader@4ff782ab

测试结论： 通过普通new Thread方法创建子线程，会继承父线程的上下文ClassLoader

*源码分析：
查看new Thread创建线程源码发现有如下代码

        if (security == null || isCCLOverridden(parent.getClass()))
            this.contextClassLoader = parent.getContextClassLoader();
        else
            this.contextClassLoader = parent.contextClassLoader;

所以子线程的上下文ClassLoader会继承父线程的上下文ClassLoader

测试二

在Tomcat容器环境下执行下述代码

        MyClassLoader myClassLoader = new MyClassLoader();

        Thread.currentThread().setContextClassLoader(myClassLoader);

        CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getContextClassLoader());
        });

输出

null

但是如果通过main函数执行上述代码，依然是会打印出自定义类加载器

为啥呢？查了一下资料，Tomcat 默认使用SafeForkJoinWorkerThreadFactory作为ForkJoinWorkerThreadFactory，然后看下SafeForkJoinWorkerThreadFactory源码

    private static class SafeForkJoinWorkerThread extends ForkJoinWorkerThread {
        protected SafeForkJoinWorkerThread(ForkJoinPool pool) {
            super(pool);
            this.setContextClassLoader(ForkJoinPool.class.getClassLoader());
        }
    }

这里发现，ForkJoinPool线程设置的ClassLoader是java.util.concurrent.ForkJoinPool的类加载器，而此类位于rt.jar包下，那它的类加载器自然就是BootStrapClassLoader了

问题解决

解决方式一：

        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();

        CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {
            Thread.currentThread().setContextClassLoader(contextClassLoader);
        });

那就是在ForkJoinPool线程中再重新设置一下上下文ClassLoader

解决方式二：

        CompletableFuture<Void> task1 = CompletableFuture.runAsync(() -> {
            Thread.currentThread().setContextClassLoader(contextClassLoader);
        },new MyExecutorService());

那就是不使用CompletableFuture的默认线程池ForkJoinPool，转而使用我们的自定义线程池

后续补充

为什么在Tomcat中CompletableFutrue的默认线程池ForkJoinPool的threadFactory却是SafeForkJoinWorkerThreadFactory？

1）首先查看ForkJoinPool设置ThreadFactory的地方源码

private static ForkJoinPool makeCommonPool() {
        int parallelism = -1;
        ForkJoinWorkerThreadFactory factory = null;
        UncaughtExceptionHandler handler = null;
        try {  // ignore exceptions in accessing/parsing properties
   、、、、、、省略不重要代码
            String fp = System.getProperty
                ("java.util.concurrent.ForkJoinPool.common.threadFactory");
            if (fp != null)
                factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
                           getSystemClassLoader().loadClass(fp).newInstance());
   、、、、、、省略不重要代码
        } catch (Exception ignore) {
        }
        if (factory == null) {
            if (System.getSecurityManager() == null)
                factory = defaultForkJoinWorkerThreadFactory;
            else // use security-managed default
                factory = new InnocuousForkJoinWorkerThreadFactory();
        }
          、、、、、、省略不重要代码

可以看到，如果可以从java.util.concurrent.ForkJoinPool.common.threadFactory中获取到值，那么就使用这个值作为ThreadFactory，相当于是一个扩展。

2）那么这个值是在哪里设置进去的呢？
设置的地方就是org.apache.catalina.core.JreMemoryLeakPreventionListener。这个类实现了LifecycleListener。其中有一行代码

                if (forkJoinCommonPoolProtection && !JreCompat.isJre9Available()) {
                    // Don't override any explicitly set property
                    if (System.getProperty(FORK_JOIN_POOL_THREAD_FACTORY_PROPERTY) == null) {
                        System.setProperty(FORK_JOIN_POOL_THREAD_FACTORY_PROPERTY,
                                SafeForkJoinWorkerThreadFactory.class.getName());
                    }
                }

可以看到就是这里的锅。

参考

https://segmentfault.com/a/1190000022798514
https://www.jianshu.com/p/8fcce16ae507
http://blog.itpub.net/69912579/viewspace-2658278/
https://tomcat.apache.org/tomcat-8.5-doc/config/listeners.html

https://bugs.java.com/bugdatabase/view_bug.do?bug_id=JDK-8172726
http://tomcat.apache.org/tomcat-8.5-doc/changelog.html

Tomcat容器应用中使用CompletableFuture时，

背景

问题定位

问题印证

测试一

测试二

问题解决

解决方式一：

解决方式二：

后续补充

为什么在Tomcat中CompletableFutrue的默认线程池ForkJoinPool的threadFactory却是SafeForkJoinWorkerThreadFactory？

参考

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读