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

java1234

共 10275字,需浏览 21分钟

 · 2020-08-25

1 文章概述

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


2 资源角度

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

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

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

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

以下两个线程池maxSize使用Integer最大值,所以可能会创建大量线程从而导致OOM

# CachedThreadPool ScheduledThreadPoolpublic static ExecutorService newCachedThreadPool() {    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,                                  60L, TimeUnit.SECONDS,                                  new SynchronousQueue());}
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() : (queues < 0 ? new LinkedBlockingQueue() : new LinkedBlockingQueue(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 queueToUse = new LinkedBlockingQueue(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线程池声明方式创建了一个规范线程池,这样使用线程池有助于快速定位和排查问题。


长按二维码关注更多精彩文章

浏览 9
点赞
评论
收藏
分享

手机扫一扫分享

举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

举报