爲何阿里巴巴不容許用Executors建立線程池

時間  2021-08-14
標籤 java 編程 微信 markdown 架構 併發 ide 源碼分析 學習 測試 欄目 阿里巴巴 简体版
原文   原文鏈接

歡迎你們關注公衆號「JAVA前線」查看更多精彩分享文章，主要包括源碼分析、實際應用、架構思惟、職場分享、產品思考等等，同時歡迎你們加我微信「java_front」一塊兒交流學習java


1 文章概述

阿里開發手冊有這樣一條強制規定：線程池不容許使用Executors去建立，而應該經過ThreadPoolExecutor方式，這樣處理方式更加明確線程池運行規則，規避資源耗盡風險。本文咱們從資源和排查問題兩個角度進行分析，同時參考DUBBO線程池聲明方式建立一個符合規範的線程池。編程


2 資源角度

阿里巴巴JAVA開發手冊主要是從資源角度對這個問題進行了分析：微信

FixedThreadPool、SingleThreadPool
容許請求隊列長度爲Integer.MAX_VALUE可能會堆積大量請求從而致使OOM

CachedThreadPool、ScheduledThreadPool
容許建立線程數量爲Integer.MAX_VALUE可能會建立大量線程從而致使OOM
複製代碼

FixedThreadPool、SingleThreadPool線程池使用鏈表實現的阻塞隊列，不設大小理論上隊列容量無上限，因此可能會堆積大量請求從而致使OOM：markdown

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("公衆號「JAVA前線」");
            }
        });
    }
}
複製代碼

上述語句在功能層面是沒有問題的，可是在生產環境中有可能遇到CPU飆高，線程數持續增長，內存溢出等問題，咱們時常須要經過線程快照進行觀察。咱們經過jstack命令觀察上述代碼線程快照：ide

"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：

"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：

"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("公衆號「JAVA前線」");
            }
        });
        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線程池聲明方式建立了一個規範線程池，這樣使用線程池有助於快速定位和排查問題。

歡迎你們關注公衆號「JAVA前線」查看更多精彩分享文章，主要包括源碼分析、實際應用、架構思惟、職場分享、產品思考等等，同時歡迎你們加我微信「java_front」一塊兒交流學習

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程