Java ThreadPoolExecutor 線程池

CollinPeng 發(fā)布于2019-08-16 16:21 / 3414人閱讀

摘要：是一個(gè)中的工具類(lèi)提供工廠方法來(lái)創(chuàng)建不同類(lèi)型的線程池從上圖中也可以看出的創(chuàng)建線程池的方法創(chuàng)建出來(lái)的線程池都實(shí)現(xiàn)了接口常用方法有以下幾個(gè)創(chuàng)建固定數(shù)目線程的線程池超出的線程會(huì)在隊(duì)列中等待創(chuàng)建一個(gè)可緩存線程池如果線程池長(zhǎng)度超過(guò)處理需要可靈活回收空閑

Executors

Executors 是一個(gè)Java中的工具類(lèi). 提供工廠方法來(lái)創(chuàng)建不同類(lèi)型的線程池.

從上圖中也可以看出, Executors的創(chuàng)建線程池的方法, 創(chuàng)建出來(lái)的線程池都實(shí)現(xiàn)了 ExecutorService接口. 常用方法有以下幾個(gè):

newFixedThreadPool(int Threads): 創(chuàng)建固定數(shù)目線程的線程池, 超出的線程會(huì)在隊(duì)列中等待.

newCachedThreadPool(): 創(chuàng)建一個(gè)可緩存線程池, 如果線程池長(zhǎng)度超過(guò)處理需要, 可靈活回收空閑線程(60秒), 若無(wú)可回收,則新建線程.

newSingleThreadExecutor(): 創(chuàng)建一個(gè)單線程化的線程池, 它只會(huì)用唯一的工作線程來(lái)執(zhí)行任務(wù), 保證所有任務(wù)按照指定順序(FIFO, LIFO, 優(yōu)先級(jí))執(zhí)行. 如果某一個(gè)任務(wù)執(zhí)行出錯(cuò), 將有另一個(gè)線程來(lái)繼續(xù)執(zhí)行.

newScheduledThreadPool(int corePoolSize): 創(chuàng)建一個(gè)支持定時(shí)及周期性的任務(wù)執(zhí)行的線程池, 多數(shù)情況下可用來(lái)替代Timer類(lèi).

Executors 例子 newCachedThreadPool

線程最大數(shù)為 Integer.MAX_VALUE, 當(dāng)我們往線程池添加了 n 個(gè)任務(wù), 這 n 個(gè)任務(wù)都是一起執(zhí)行的.

        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

newFixedThreadPool

        ExecutorService cachedThreadPool = Executors.newFixedThreadPool(1);
        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        cachedThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(1000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

newScheduledThreadPool

三秒執(zhí)行一次, 只有執(zhí)行完這一次后, 才會(huì)執(zhí)行.

        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);
        scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        Thread.currentThread().sleep(2000);
                        System.out.println(Thread.currentThread().getName());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }, 3, TimeUnit.SECONDS);

newSingleThreadExecutor

順序執(zhí)行各個(gè)任務(wù), 第一個(gè)任務(wù)執(zhí)行完, 才會(huì)執(zhí)行下一個(gè).

        ExecutorService executorService = Executors.newSingleThreadExecutor();
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        System.out.println(Thread.currentThread().getName());
                        Thread.currentThread().sleep(10000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                for (;;) {
                    try {
                        System.out.println(Thread.currentThread().getName());
                        Thread.currentThread().sleep(2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

Executors存在什么問(wèn)題

在阿里巴巴Java開(kāi)發(fā)手冊(cè)中提到，使用Executors創(chuàng)建線程池可能會(huì)導(dǎo)致OOM(OutOfMemory ,內(nèi)存溢出)，但是并沒(méi)有說(shuō)明為什么，那么接下來(lái)我們就來(lái)看一下到底為什么不允許使用Executors？

我們先來(lái)一個(gè)簡(jiǎn)單的例子，模擬一下使用Executors導(dǎo)致OOM的情況.

/**
 * @author Hollis
 */
public class ExecutorsDemo {
    private static ExecutorService executor = Executors.newFixedThreadPool(15);
    public static void main(String[] args) {
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            executor.execute(new SubThread());
        }
    }
}

class SubThread implements Runnable {
    @Override
    public void run() {
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            //do nothing
        }
    }
}

通過(guò)指定JVM參數(shù)：-Xmx8m -Xms8m 運(yùn)行以上代碼，會(huì)拋出OOM:

Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded
    at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371)
    at com.hollis.ExecutorsDemo.main(ExecutorsDemo.java:16)

以上代碼指出，ExecutorsDemo.java 的第16行，就是代碼中的 executor.execute(new SubThread());

Java中的 BlockingQueue 主要有兩種實(shí)現(xiàn), 分別是 ArrayBlockingQueue 和 LinkedBlockingQueue.

ArrayBlockingQueue 是一個(gè)用數(shù)組實(shí)現(xiàn)的有界阻塞隊(duì)列, 必須設(shè)置容量.

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

LinkedBlockingQueue 是一個(gè)用鏈表實(shí)現(xiàn)的有界阻塞隊(duì)列, 容量可以選擇進(jìn)行設(shè)置, 不設(shè)置的話, 將是一個(gè)無(wú)邊界的阻塞隊(duì)列, 最大長(zhǎng)度為 Integer.MAX_VALUE.

public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}

這里的問(wèn)題就出在如果我們不設(shè)置 LinkedBlockingQueue 的容量的話, 其默認(rèn)容量將會(huì)是 Integer.MAX_VALUE.

而 newFixedThreadPool 中創(chuàng)建 LinkedBlockingQueue 時(shí), 并未指定容量. 此時(shí), LinkedBlockingQueue 就是一個(gè)無(wú)邊界隊(duì)列, 對(duì)于一個(gè)無(wú)邊界隊(duì)列來(lái)說(shuō), 是可以不斷的向隊(duì)列中加入任務(wù)的, 這種情況下就有可能因?yàn)槿蝿?wù)過(guò)多而導(dǎo)致內(nèi)存溢出問(wèn)題.

newCachedThreadPool 和 newScheduledThreadPool 這兩種方式創(chuàng)建的最大線程數(shù)可能是Integer.MAX_VALUE, 而創(chuàng)建這么多線程, 必然就有可能導(dǎo)致OOM.

ThreadPoolExecutor 創(chuàng)建線程池

避免使用 Executors 創(chuàng)建線程池, 主要是避免使用其中的默認(rèn)實(shí)現(xiàn), 那么我們可以自己直接調(diào)用 ThreadPoolExecutor 的構(gòu)造函數(shù)來(lái)自己創(chuàng)建線程池. 在創(chuàng)建的同時(shí), 給 BlockQueue 指定容量就可以了.

ExecutorService executor = new ThreadPoolExecutor(10, 10,
        60L, TimeUnit.SECONDS,
        new ArrayBlockingQueue(10));

這種情況下, 一旦提交的線程數(shù)超過(guò)當(dāng)前可用線程數(shù)時(shí), 就會(huì)拋出 java.util.concurrent.RejectedExecutionException, 這是因?yàn)楫?dāng)前線程池使用的隊(duì)列是有邊界隊(duì)列, 隊(duì)列已經(jīng)滿了便無(wú)法繼續(xù)處理新的請(qǐng)求.

除了自己定義 ThreadPoolExecutor 外. 還有其他方法. 如apache和guava等.

四個(gè)構(gòu)造函數(shù)

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue)
             
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory)
             
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              RejectedExecutionHandler handler)
             
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

int corePoolSize => 該線程池中核心線程數(shù)最大值
線程池新建線程的時(shí)候，如果當(dāng)前線程總數(shù)小于corePoolSize, 則新建的是核心線程, 如果超過(guò)corePoolSize, 則新建的是非核心線程

核心線程默認(rèn)情況下會(huì)一直存活在線程池中, 即使這個(gè)核心線程啥也不干(閑置狀態(tài)).

如果指定 ThreadPoolExecutor 的 allowCoreThreadTimeOut 這個(gè)屬性為 true, 那么核心線程如果不干活(閑置狀態(tài))的話, 超過(guò)一定時(shí)間(時(shí)長(zhǎng)下面參數(shù)決定), 就會(huì)被銷(xiāo)毀掉

很好理解吧, 正常情況下你不干活我也養(yǎng)你, 因?yàn)槲铱傆杏玫侥愕臅r(shí)候, 但有時(shí)候特殊情況(比如我自己都養(yǎng)不起了), 那你不干活我就要把你干掉了

int maximumPoolSize
該線程池中線程總數(shù)最大值

線程總數(shù) = 核心線程數(shù) + 非核心線程數(shù).

long keepAliveTime
該線程池中非核心線程閑置超時(shí)時(shí)長(zhǎng)

一個(gè)非核心線程, 如果不干活(閑置狀態(tài))的時(shí)長(zhǎng)超過(guò)這個(gè)參數(shù)所設(shè)定的時(shí)長(zhǎng), 就會(huì)被銷(xiāo)毀掉

如果設(shè)置 allowCoreThreadTimeOut = true, 則會(huì)作用于核心線程

TimeUnit unit

keepAliveTime的單位, TimeUnit是一個(gè)枚舉類(lèi)型, 其包括:

TimeUnit.DAYS;               //天
TimeUnit.HOURS;             //小時(shí)
TimeUnit.MINUTES;           //分鐘
TimeUnit.SECONDS;           //秒
TimeUnit.MILLISECONDS;      //毫秒
TimeUnit.MICROSECONDS;      //微妙
TimeUnit.NANOSECONDS;       //納秒

BlockingQueue workQueue

一個(gè)阻塞隊(duì)列, 用來(lái)存儲(chǔ)等待執(zhí)行的任務(wù). 也就是說(shuō)現(xiàn)在有10個(gè)任務(wù), 核心線程 有四個(gè), 非核心線程有六個(gè), 那么這六個(gè)線程會(huì)被添加到 workQueue 中, 等待執(zhí)行.

這個(gè)參數(shù)的選擇也很重要, 會(huì)對(duì)線程池的運(yùn)行過(guò)程產(chǎn)生重大影響, 一般來(lái)說(shuō), 這里的阻塞隊(duì)列有以下幾種選擇:

SynchronousQueue: 這個(gè)隊(duì)列接收到任務(wù)的時(shí)候, 會(huì)直接提交給線程處理, 而不保留它, 如果所有線程都在工作怎么辦？那就*新建一個(gè)線程來(lái)處理這個(gè)任務(wù)！所以為了保證不出現(xiàn)<線程數(shù)達(dá)到了maximumPoolSize而不能新建線程>的錯(cuò)誤, 使用這個(gè)類(lèi)型隊(duì)列的時(shí)候, maximumPoolSize 一般指定成 Integer.MAX_VALUE, 即無(wú)限大.

LinkedBlockingQueue: 這個(gè)隊(duì)列接收到任務(wù)的時(shí)候, 如果當(dāng)前線程數(shù)小于核心線程數(shù), 則核心線程處理任務(wù); 如果當(dāng)前線程數(shù)等于核心線程數(shù), 則進(jìn)入隊(duì)列等待. 由于這個(gè)隊(duì)列最大值為 Integer.MAX_VALUE , 即所有超過(guò)核心線程數(shù)的任務(wù)都將被添加到隊(duì)列中，這也就導(dǎo)致了 maximumPoolSize 的設(shè)定失效, 因?yàn)榭偩€程數(shù)永遠(yuǎn)不會(huì)超過(guò) corePoolSize.

ArrayBlockingQueue: 可以限定隊(duì)列的長(zhǎng)度, 接收到任務(wù)的時(shí)候, 如果沒(méi)有達(dá)到 corePoolSize 的值, 則核心線程執(zhí)行任務(wù), 如果達(dá)到了, 則入隊(duì)等候, 如果隊(duì)列已滿, 則新建線程(非核心線程)執(zhí)行任務(wù), 又如果總線程數(shù)到了maximumPoolSize, 并且隊(duì)列也滿了, 則發(fā)生錯(cuò)誤.

DelayQueue: 隊(duì)列內(nèi)元素必須實(shí)現(xiàn) Delayed 接口, 這就意味著你傳進(jìn)去的任務(wù)必須先實(shí)現(xiàn)Delayed接口. 這個(gè)隊(duì)列接收到任務(wù)時(shí), 首先先入隊(duì), 只有達(dá)到了指定的延時(shí)時(shí)間, 才會(huì)執(zhí)行任務(wù).

ThreadFactory threadFactory

它是ThreadFactory類(lèi)型的變量, 用來(lái)創(chuàng)建新線程.

默認(rèn)使用 Executors.defaultThreadFactory() 來(lái)創(chuàng)建線程. 使用默認(rèn)的 ThreadFactory 來(lái)創(chuàng)建線程時(shí), 會(huì)使新創(chuàng)建的線程具有相同的 NORM_PRIORITY 優(yōu)先級(jí)并且是非守護(hù)線程, 同時(shí)也設(shè)置了線程的名稱(chēng).

RejectedExecutionHandler handler

表示當(dāng)拒絕處理任務(wù)時(shí)的策略, 有以下四種取值:

ThreadPoolExecutor.AbortPolicy:丟棄任務(wù)并拋出RejectedExecutionException異常(默認(rèn)).
ThreadPoolExecutor.DiscardPolicy：直接丟棄任務(wù), 但是不拋出異常.
ThreadPoolExecutor.DiscardOldestPolicy：丟棄隊(duì)列最前面的任務(wù), 然后重新嘗試執(zhí)行任務(wù)(重復(fù)此過(guò)程)
ThreadPoolExecutor.CallerRunsPolicy：用調(diào)用者所在的線程來(lái)執(zhí)行任務(wù).