为什么阿里不允许用Executors创建线程池

怼烎@ 2021-09-24 16:50 274阅读 0赞

> 欢迎大家关注公众号「JAVA前线」查看更多精彩分享文章，主要包括源码分析、实际应用、架构思维、职场分享、产品思考等等，同时欢迎大家加我微信「java\_front」一起交流学习

# 1 文章概述 #

《阿里巴巴JAVA开发手册》有这样一条强制规定：线程池不允许使用Executors去创建，而应该通过ThreadPoolExecutor方式，这样处理方式更加明确线程池运行规则，规避资源耗尽风险。本文我们从资源和排查问题两个角度进行分析，同时参考DUBBO线程池声明方式创建一个符合规范的线程池。

# 2 资源角度 #

《阿里巴巴JAVA开发手册》从资源角度对这个问题进行了分析

> FixedThreadPool SingleThreadPool  
> 允许请求队列长度为Integer.MAX\_VALUE可能会堆积大量请求从而导致OOM  
>   
> 
> 
> CachedThreadPool ScheduledThreadPool  
> 允许创建线程数量为Integer.MAX\_VALUE可能会创建大量线程从而导致OOM

FixedThreadPool、SingleThreadPool线程池使用链表实现的阻塞队列，不设大小理论上队列容量无上限，所以可能会堆积大量请求从而导致OOM

public static ExecutorService newFixedThreadPool(int nThreads) { 
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    public static ExecutorService newSingleThreadExecutor() { 
        return new FinalizableDelegatedExecutorService
               (new ThreadPoolExecutor(1, 1,
                                       0L, TimeUnit.MILLISECONDS,
                                       new LinkedBlockingQueue<Runnable>()));
    }

CachedThreadPool ScheduledThreadPool线程池maxSize使用Integer最大值，所以可能会创建大量线程从而导致OOM

public static ExecutorService newCachedThreadPool() { 
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    
    public static ScheduledExecutorService newSingleThreadScheduledExecutor() { 
        return new DelegatedScheduledExecutorService
               (new ScheduledThreadPoolExecutor(1));
    }
    
    public ScheduledThreadPoolExecutor(int corePoolSize) { 
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

# 3 排查问题角度 #

如果使用Executors创建线程池，大家应该最常使用如下语句

public void testThread() throws Exception { 
        ExecutorService fixedExecutor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 1000; i++) { 
            fixedExecutor.execute(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("公众号互联网公园");
                }
            });
        }
    }

上述语句在功能层面是没有问题的，但是在生产环境中有可能遇到CPU飙高，线程数持续增加，内存溢出等问题，我们时常需要通过线程快照进行观察。我们通过jstack命令观察上述代码线程快照

"pool-1-thread-2" #525 prio=5 os_prio=0 tid=0x00006f6561039100 nid=0xdaa waiting on condition [0x00006f64e646d000]
    java.lang.Thread.State: WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    parking to wait for <0x00000006e6f3e230> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:165)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
    at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1066)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1126)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:616)
    at java.lang.Thread.run(Thread.java:645)

我们发现从线程快照看不出任何业务信息，只有类似pool-1-thread-2这种编号信息，不利于排查问题，我们需要给线程命名。

# 4 为线程进行命名 #

在并发编程中我们一定要为线程命名，这样有助于排查问题，关于如何命名我们可以参考DUBBO源码，分析FixedThreadPool线程池会发现其使用命名工厂为生产者和消费者线程进行命名

public class FixedThreadPool implements ThreadPool { 
    
        @Override
        public Executor getExecutor(URL url) { 
    
            // 线程名称
            String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
    
            // 线程个数默认200
            int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
    
            // 队列容量默认0
            int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
    
            // 队列容量等于0使用阻塞队列SynchronousQueue
            // 队列容量小于0使用无界阻塞队列LinkedBlockingQueue
            // 队列容量大于0使用有界阻塞队列LinkedBlockingQueue
            // NamedInternalThreadFactory为线程命名
            return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
                                          queues == 0 ? new SynchronousQueue<Runnable>()
                                          : (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                             : new LinkedBlockingQueue<Runnable>(queues)),
                                          new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
        }
    }

生产者默认线程名DubboServerHandler

public abstract class AbstractServer extends AbstractEndpoint implements Server { 
        protected static final String SERVER_THREAD_POOL_NAME = "DubboServerHandler";
    }

生产者线程快照信息如下

"DubboServerHandler-1.1.1.1:20881-thread-20" #511 daemon prio = 5 os_prio = 0 tid = 0x00001f153121f200 nid = 0xd1a waiting on condition [0x00001f14edcdf000]
    java.lang.Thread.State: WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for <0x00000001e1f3abc0> (a java.util.concurrent.SynchronousQueue$TransferStack)
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java : 115)
    at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java : 452)
    at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java : 312)
    at java.util.concurrent.SynchronousQueue.take(SynchronousQueue.java : 924)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java : 1011)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java : 1121)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java : 111)

消费者默认线程名DubboClientHandler

public abstract class AbstractClient extends AbstractEndpoint implements Client { 
        protected static final String CLIENT_THREAD_POOL_NAME = "DubboClientHandler";
    }

消费者线程快照信息如下

"DubboClientHandler-1.1.1.1:20881-thread-10" #688 daemon prio=1 os_prio=0 tid=0x00001f6114004800 nid=0x14d8 waiting on condition [0x00001f63e131a000]
    java.lang.Thread.State: TIMED_WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for <0x00000006e21df0d0> (a java.util.concurrent.SynchronousQueue$TransferStack)
    at java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:111)
    at java.util.concurrent.SynchronousQueue$TransferStack.awaitFulfill(SynchronousQueue.java:460)
    at java.util.concurrent.SynchronousQueue$TransferStack.transfer(SynchronousQueue.java:361)
    at java.util.concurrent.SynchronousQueue.poll(SynchronousQueue.java:141)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1066)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1111)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:611)
    at java.lang.Thread.run(Thread.java:141)

# 5 创建一个规范线程池 #

我们参考DUBBO线程池定义命名工厂

public class NamedInternalThreadFactory extends NamedThreadFactory { 
        public NamedInternalThreadFactory() { 
            super();
        }
    
        public NamedInternalThreadFactory(String prefix) { 
            super(prefix, false);
        }
    
        public NamedInternalThreadFactory(String prefix, boolean daemon) { 
            super(prefix, daemon);
        }
    
        @Override
        public Thread newThread(Runnable runnable) { 
            String name = mPrefix + mThreadNum.getAndIncrement();
            InternalThread ret = new InternalThread(mGroup, runnable, name, 0);
            ret.setDaemon(mDaemon);
            return ret;
        }
    }
    
    public class NamedThreadFactory implements ThreadFactory { 
        protected static final AtomicInteger POOL_SEQ = new AtomicInteger(1);
        protected final AtomicInteger mThreadNum = new AtomicInteger(1);
        protected final String mPrefix;
        protected final boolean mDaemon;
        protected final ThreadGroup mGroup;
    
        public NamedThreadFactory() { 
            this("pool-" + POOL_SEQ.getAndIncrement(), false);
        }
    
        public NamedThreadFactory(String prefix) { 
            this(prefix, false);
        }
    
        public NamedThreadFactory(String prefix, boolean daemon) { 
            mPrefix = prefix + "-thread-";
            mDaemon = daemon;
            SecurityManager s = System.getSecurityManager();
            mGroup = (s == null) ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
        }
    
        @Override
        public Thread newThread(Runnable runnable) { 
            String name = mPrefix + mThreadNum.getAndIncrement();
            Thread ret = new Thread(mGroup, runnable, name, 0);
            ret.setDaemon(mDaemon);
            return ret;
        }
    
        public ThreadGroup getThreadGroup() { 
            return mGroup;
        }
    }

再定义一个线程池，在线程池执行方法开放一个业务名称参数供调用方设置

public class ThreadPoolStarter { 
          public static ThreadPoolExecutor getExecutor(String threadName) { 
            if (executor == null) { 
              synchronized (ThreadPoolStarter.class) { 
                if (executor == null) { 
                  int coreSize = Runtime.getRuntime().availableProcessors();
                  BlockingQueue<Runnable> queueToUse = new LinkedBlockingQueue<Runnable>(QUEUE_SIZE);
                  executor = new ThreadPoolExecutor(coreSize, POOL_CORE_SIZE, MAX_SIZE, TimeUnit.SECONDS, queueToUse, new NamedInternalThreadFactory(threadName, true), new AbortPolicyDoReport(threadName));
                }
          }
        }
        return executor;
      }
    }
    
    public class ThreadExecutor { 
      public static void execute(String bizName, Runnable job) { 
        ThreadPoolStarter.getExecutor(bizName).execute(job);
      }
    
      public static Future<?> sumbit(String bizName, Runnable job) { 
        return ThreadPoolStarter.getExecutor(bizName).submit(job);
      }
    }

编写一个实例进行测试

public void testThread() throws Exception { 
        for (int i = 0; i < 10000; i++) { 
          ThreadExecutor.execute("BizName", new Runnable() { 
            @Override
            public void run() { 
              System.out.println("公众号互联网公园");
            }
          });
          Thread.sleep(1000L);
        }
      }
    }

再观察线程快照可以清晰查看业务名

"BizName-thread-8" #262 daemon prio=5 os_prio=0 tid=0x0000000023b5c000 nid=0x31d4 waiting on condition [0x000000003c0be000]
    java.lang.Thread.State: WAITING (parking)
    at sun.misc.Unsafe.park(Native Method)
    - parking to wait for <0x00000006c35781f0> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
    at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
    at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
    at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)

# 6 文章总结 #

本文首先介绍了《阿里巴巴JAVA开发手册》不允许使用Executors创建线程池这个规定，然后从资源和排查问题两个角度分析了为什么这么规定，最后我们参考DUBBO线程池声明方式创建了一个规范线程池，这样使用线程池有助于快速定位和排查问题。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvc2hpeHV5ZQ_size_16_color_FFFFFF_t_70_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvc2hpeHV5ZQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20210923/c6ccf405fdec49db857a3037ba3e6be6.png