Java併發篇-全面解析Executor框架

時間 2020-04-22

標籤 java 併發全面解析 executor 框架欄目 Java 简体版

原文原文鏈接

Executor你們族成員

成員分爲四個部分：任務、任務執行、任務執行結果以及任務執行工具類java

任務：實現Callable接口或Runnable接口面試

任務執行部分：ThreadPoolExecutor以及ScheduledThreadPoolExecutor多線程

任務執行結果：Future接口以及FutureTask實現類框架

任務執行工廠類：Executors異步

任務執行框架

ThreadPoolExecutor（核心）ide

ThreadPoolExecutor一般使用Executors進行建立，其內部含有三種線程池：函數

FixedThreadPool：含有固定線程數的線程池。工具
SingleThreadExecutor：單線程的線程池，須要保證任務順序執行時採用。post
CachedThreadPool：大小無界的線程池，只要須要線程就能夠一直建立線程。this

ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor一般使用Executors進行建立，其內部含有兩種線程池：

ScheduledThreadPoolExecutor：含有固定線程數的定時任務線程池
SingleThreadScheduledExecutor：只包含一個線程數的定時任務線程池，須要保證任務順序執行時採用。

任務執行結果

提交任務給任務執行框架執行後，任務執行框架會返回一個Future接口類型的對象，該對象內部包含了任務執行結果信息。

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<> submit(Runnable task)
複製代碼

在JDK1.8返回的是FutureTask對象，但不必定返回的就是FutureTask對象，只保證返回Future接口。

任務

實現Runnable或Callable都可被任務執行框架執行，它們之間的區別是實現Runnable的任務執行完成後不會返回結果，而實現Callable接口的任務能夠返回結果。

返回結果爲null的封裝（沒有什麼意義）

public static Callable<Object> callable(Runnable task) 複製代碼

擁有返回結果的封裝

public static <T> Callable<T> callable(Runnable task, T result) 複製代碼

ThreadPoolExecutor

上一篇講過ThreadPoolExecutor構造函數中的參數含義，這裏再也不贅述，若是忘記了或者還沒閱讀過，能夠先去了解一下，兩篇文章是銜接的：juejin.im/post/5e9c45…

主要參數有4個：

corePolSize：核心線程池大小
maximumPoolSize：最大線程池大小
BlockingQueue：存儲未執行任務的阻塞隊列
RejectedExecutionHandler：任務沒法被執行時的拒絕策略

FixedThreadPool詳解

fixedThreadPool稱爲可重用固定線程數的線程池，下面是構造該線程池的方法源碼：

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

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}
複製代碼

參數解讀

corePoolSize和maximumPoolSize都設置爲用戶傳入的nThreads。
keepAliveTime設置爲0L，表明只要空閒線程閒下來就立刻被終止，若是設置爲SECONDS > 0，則表明線程空閒下來後等待新任務的最長時間SECONDS。
TimeUnit表明keepAliveTime的單位
workQueue設置阻塞隊列

execute()執行流程

curThread表明核心線程池的當前線程數

若是 curThread < corPoolSize，那麼建立新線程執行任務
若是 curThread = corPoolSize，那麼將任務加入阻塞隊列當中
線程池中的線程執行完任務後會空閒下來，而後會一直從阻塞隊列中獲取任務執行

使用無界阻塞隊列LinkedBlockingQueue的後果：

curThread = corPoolSize後，任務會一直加入到阻塞隊列，嚴重時阻塞隊列的任務數過多會形成GC內存溢出。
使用無界隊列，那麼參數maximumPoolSize是一個無效參數。只有當任務沒法加入到阻塞隊列時知足curThread < maximumPoolSiize纔會在線程池中建立新的線程執行該任務。
因爲第1點和第2點致使keepAliveTime參數無效
使用無界隊列後，任務均可以加入到阻塞隊列中，因此不會出現curThread > maximumPoolSize，也就不會出現任務拒絕執行的狀況，不會調用任務拒絕方法。

GC內存溢出示例（我的理解，若是有錯，歡迎指出！）：

public class Task implements Runnable {    
    private String name;   
    public Task(String name) {
        this.name = name;
    }
    @Override
    public void run() {
        System.out.println(name);
    }
}
public class FixedThreadPoolTest {   
    private static final ExecutorService fixThreadPool = Executors.newFixedThreadPool(10);
    private static int i = 0;
    
    public static void main(String[] args) {
        // 不停地往線程池中提交任務
        for (;i < 100000;i++) {
            fixThreadPool.execute(new Task("任務" + i));
        }
        fixThreadPool.shutdown();
    }
}
複製代碼

設置堆內存參數：-Xms2m -Xmx2m

拋出OOM異常，緣由在於阻塞隊列中的任務數過多。

SingleThreadPool詳解

使用單個Worker線程的Executor。本質上是fixedThreadPool的corePoolSize和maximumPoolSize都設置爲1，此外無其它區別。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
複製代碼

參數解釋

corePoolSize和MaximumPoolSize的參數都設置爲1，其它參數含義和默認值都和fixedThreadPool相同。
無界阻塞隊列對線程池的影響和fixedThreadPool相同。

execute()執行流程

若是線程池尚未預熱，第一個任務來的時候就建立Worker線程來執行任務
沒法建立新線程時放入阻塞隊列中
惟一的工做者線程不斷地執行阻塞隊列中的任務

CachedThreadPool詳解

根據須要建立新線程的線程池

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
複製代碼

參數解釋