為什么阿里不允許用Executors創(chuàng)建線程池
[導讀]1 文章概述 《阿里巴巴JAVA開發(fā)手冊》有這樣一條強制規(guī)定：線程池不允許使用Executors去創(chuàng)建，而應該通過ThreadPoolExecutor方式，這樣處理方式更加明確線程池運行規(guī)則，規(guī)避資源耗盡風險。本文我們從資源和排查問題兩個角度進行分析，同時參考DUBBO線程池聲
1 文章概述
《阿里巴巴JAVA開發(fā)手冊》有這樣一條強制規(guī)定：線程池不允許使用Executors去創(chuàng)建，而應該通過ThreadPoolExecutor方式，這樣處理方式更加明確線程池運行規(guī)則，規(guī)避資源耗盡風險。本文我們從資源和排查問題兩個角度進行分析，同時參考DUBBO線程池聲明方式創(chuàng)建一個符合規(guī)范的線程池。
2 資源角度
《阿里巴巴JAVA開發(fā)手冊》從資源角度對這個問題進行了分析
FixedThreadPool SingleThreadPool允許請求隊列長度為Integer.MAX_VALUE可能會堆積大量請求從而導致OOMCachedThreadPool ScheduledThreadPool允許創(chuàng)建線程數量為Integer.MAX_VALUE可能會創(chuàng)建大量線程從而導致OOM
以下兩個線程池使用鏈表實現(xiàn)的阻塞隊列，不設大小理論上隊列容量無上限，所以可能會堆積大量請求從而導致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>()));}
以下兩個線程池maxSize使用Integer最大值，所以可能會創(chuàng)建大量線程從而導致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創(chuàng)建線程池，大家應該最常使用如下語句
public void testThread() throws Exception {ExecutorService fixedExecutor = Executors.newFixedThreadPool(10);for (int i = 0; i < 1000; i++) {fixedExecutor.execute(new Runnable() {public void run() {System.out.println("公眾號互聯(lián)網公園");}});}}
上述語句在功能層面是沒有問題的，但是在生產環(huán)境中有可能遇到CPU飆高，線程數持續(xù)增加，內存溢出等問題，我們時常需要通過線程快照進行觀察。我們通過jstack命令觀察上述代碼線程快照
#525 prio=5 os_prio=0 tid=0x00006f6561039100 nid=0xdaa waiting on condition [0x00006f64e646d000]: 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)
我們發(fā)現(xiàn)從線程快照看不出任何業(yè)務信息，只有類似pool-1-thread-2這種編號信息，不利于排查問題，我們需要給線程命名。
4 為線程進行命名
在并發(fā)編程中我們一定要為線程命名，這樣有助于排查問題，關于如何命名我們可以參考DUBBO源碼，分析FixedThreadPool線程池會發(fā)現(xiàn)其使用命名工廠為生產者和消費者線程進行命名
public class FixedThreadPool implements ThreadPool {public Executor getExecutor(URL url) {// 線程名稱String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);// 線程個數默認200int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);// 隊列容量默認0int 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";}
消費者線程快照信息如下
:20881-thread-10" #688 daemon prio=1 os_prio=0 tid=0x00001f6114004800 nid=0x14d8 waiting on condition [0x00001f63e131a000]: 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 創(chuàng)建一個規(guī)范線程池
我們參考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);}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();}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;}}
再定義一個線程池，在線程池執(zhí)行方法開放一個業(yè)務名稱參數供調用方設置
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() {public void run() {System.out.println("公眾號互聯(lián)網公園");}});Thread.sleep(1000L);}}}
再觀察線程快照可以清晰查看業(yè)務名
#262 daemon prio=5 os_prio=0 tid=0x0000000023b5c000 nid=0x31d4 waiting on condition [0x000000003c0be000]: 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開發(fā)手冊》不允許使用Executors創(chuàng)建線程池這個規(guī)定，然后從資源和排查問題兩個角度分析了為什么這么規(guī)定，最后我們參考DUBBO線程池聲明方式創(chuàng)建了一個規(guī)范線程池，這樣使用線程池有助于快速定位和排查問題。
特別推薦一個分享架構+算法的優(yōu)質內容，還沒關注的小伙伴，可以長按關注一下：
長按訂閱更多精彩▼
如有收獲，點個在看，誠摯感謝
免責聲明：本文內容由21ic獲得授權后發(fā)布，版權歸原作者所有，本平臺僅提供信息存儲服務。文章僅代表作者個人觀點，不代表本平臺立場，如有問題，請聯(lián)系我們，謝謝！
下載文檔
