(十七)java多線程之ThreadPoolExecutor

本人郵箱: <kco1989@qq.com>
歡迎轉載,轉載請註明網址 http://blog.csdn.net/tianshi_kco
github: https://github.com/kco1989/kco
代碼已經所有託管github有須要的同窗自行下載

引言

在以前的例子,咱們要建立多個線程處理一批任務的時候.我是經過建立線程數組,或者使用線程集合來管理的.可是這樣作不太好,由於這些線程沒有被重複利用.因此這裏要引入線程池,今天咱們就講線程池執行器ThreadPoolExecutor.

理論

首先咱們來看一下它的構造器:
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) 這個是ThreadPoolExecutor完整的構造器,其餘的構造器其實也是在內部調用這個.

• corePoolSize 核心線程數,線程池保留線程的數量,即便這些線程是空閒.除非設置了allowCoreThreadTimeOut

• maximumPoolSize 線程池最大容許的線程數.

• keepAliveTime 噹噹前的線程數大於核心線程數,那麼這些多餘的空閒的線程在被終止以前能等待新任務的時間.

• unit keepAliveTime時間的單位

• workQueue 這個是用來保留將要執行的工做隊列.

• threadFactory 用於建立新線程的工廠

• handler 若是工做隊列(workQueue)滿了,那麼這個handler是將會被執行.

ThreadPoolExecutor還有幾個可不帶threadFactory或handler慘的構造器,說明java提供了一些默認的配置,讓咱們看一下.

若是構造不帶threadFactory,那麼默認使用java.util.concurrent.Executors.DefaultThreadFactory建立出一個新的工廠對象.經過閱讀源代碼,主要是在建立新的線程的時候修改了線程名爲pool-全局線程池遞增數編號-thread-當前線程池線程遞增編號,讓線程改成非守護線程,並設置線程的優先級爲NORM_PRIORITY.

ok,再看一下handler有什麼默認值.

• java.util.concurrent.ThreadPoolExecutor.AbortPolicy 這個是默認使用的拒絕策略,若是有要執行的任務隊列已滿,且還有任務提交,則直接拋出異常信息

• java.util.concurrent.ThreadPoolExecutor.DiscardPolicy 這個是忽略策略,若是有要執行的任務隊列已滿,且還有任務提交,則直接忽略掉這個任務,即不拋出異常也不作任何處理.

• java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy 忽略最先提交的任務.若是有要執行的任務隊列已滿,此時若還有任務提交且線程池尚未中止,則把隊列中最先提交的任務拋棄掉,而後把當前任務加入隊列中.

• java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy 這個是來着不拒策略.若是有要執行的任務隊列已滿,此時若還有任務提交且線程池尚未中止,則直接運行任務的run方法.

例子 使用默認的拒絕策略AbortPolicy

public class Demo1 {
    public static void main(String[] args) {
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10);
        RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
        ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 3, 0, TimeUnit.SECONDS, queue, handler);
        for (int i = 0; i < 20; i ++){
            final int temp = i;
            pool.execute(() -> {
                System.out.println("客戶" + temp + "來了.......");
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        pool.shutdown();
    }
}

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.kco.test17.demo1.Demo1$$Lambda$1/15497079@ca494b rejected from java.util.concurrent.ThreadPoolExecutor@1a4f24f[Running, pool size = 5, active threads = 5, queued tasks = 10, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
at com.kco.test17.demo1.Demo1.main(Demo1.java:16)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)
pool-1-thread-1客戶0來了.......
pool-1-thread-2客戶1來了.......
pool-1-thread-3客戶2來了.......
pool-1-thread-5客戶14來了.......
pool-1-thread-4客戶13來了.......
pool-1-thread-2客戶3來了.......
pool-1-thread-1客戶4來了.......
pool-1-thread-5客戶5來了.......
pool-1-thread-3客戶6來了.......
pool-1-thread-4客戶7來了.......
pool-1-thread-2客戶9來了.......
pool-1-thread-1客戶8來了.......
pool-1-thread-3客戶10來了.......
pool-1-thread-5客戶11來了.......
pool-1-thread-4客戶12來了.......

從結果看出來,能夠看出線程是重複被使用的,並且當執行的任務超過工做隊列的容量時,線程確實拋出了異常.

例子2 使用忽略策略 DiscardPolicy

將RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();改成 RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();
運行結果以下:

pool-1-thread-1客戶0來了.......
pool-1-thread-3客戶2來了.......
pool-1-thread-4客戶13來了.......
pool-1-thread-5客戶14來了.......
pool-1-thread-3客戶3來了.......
pool-1-thread-4客戶4來了.......
pool-1-thread-1客戶5來了.......
pool-1-thread-5客戶6來了.......
pool-1-thread-2客戶1來了.......
pool-1-thread-3客戶7來了.......
pool-1-thread-4客戶8來了.......
pool-1-thread-5客戶9來了.......
pool-1-thread-1客戶10來了.......
pool-1-thread-2客戶11來了.......
pool-1-thread-4客戶12來了.......

如今線程池正確退出了,並且也不拋出異常了,可是超過工做隊列容量的任務所有被忽略了.

例子3 使用忽略最先任務策略 DiscardOldestPolicy

RejectedExecutionHandler改成RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardOldestPolicy();

pool-1-thread-1客戶0來了.......
pool-1-thread-2客戶1來了.......
pool-1-thread-3客戶2來了.......
pool-1-thread-5客戶14來了.......
pool-1-thread-4客戶13來了.......
pool-1-thread-4客戶8來了.......
pool-1-thread-1客戶11來了.......
pool-1-thread-5客戶10來了.......
pool-1-thread-3客戶9來了.......
pool-1-thread-2客戶12來了.......
pool-1-thread-1客戶15來了.......
pool-1-thread-4客戶16來了.......
pool-1-thread-5客戶17來了.......
pool-1-thread-2客戶19來了.......
pool-1-thread-3客戶18來了.......

從以上結果,咱們能夠看出除了客戶0到客戶2恰好是3個核心線程被執行後,客戶3到客戶7直接被忽略掉了.

例子4 使用來着不拒策略 CallerRunsPolicy

一樣講拒絕策略改成RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
運行程序,結果以下:

pool-1-thread-1客戶0來了.......
pool-1-thread-2客戶1來了.......
pool-1-thread-3客戶2來了.......
pool-1-thread-4客戶13來了.......
main客戶15來了.......
pool-1-thread-5客戶14來了.......
pool-1-thread-2客戶3來了.......
pool-1-thread-1客戶4來了.......
main客戶18來了.......
pool-1-thread-3客戶5來了.......
pool-1-thread-4客戶7來了.......
pool-1-thread-5客戶6來了.......
pool-1-thread-5客戶8來了.......
pool-1-thread-1客戶9來了.......
pool-1-thread-4客戶10來了.......
pool-1-thread-3客戶12來了.......
pool-1-thread-2客戶11來了.......
pool-1-thread-1客戶16來了.......
pool-1-thread-5客戶19來了.......
pool-1-thread-3客戶17來了.......

結果,咱們能夠發現全部的任務都被執行,並且居然還有兩個是在主線程執行的.如今明白我以前說的則直接運行任務的run方法的意思了吧,沒錯是直接調用run方法,而不是開啓線程去執行任務.

例子5 使用自定義的拒絕策略

如今咱們本身寫一個拒絕策略,要求全部的任務都必須被線程池執行,並且都要在線程池中執行.

public class Demo5 {
    public static void main(String[] args) {
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(10);
        RejectedExecutionHandler handler = new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                if (!executor.isShutdown()){
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 5, 0, TimeUnit.SECONDS, queue, handler);
        for (int i = 0; i < 20; i ++){
            final int temp = i;
            pool.execute(() -> {
                String name = Thread.currentThread().getName();
                System.out.println(name + "客戶" + temp + "來了.......");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        pool.shutdown();
    }
}

運行結果:

pool-1-thread-1客戶0來了.......
pool-1-thread-3客戶2來了.......
pool-1-thread-5客戶14來了.......
pool-1-thread-4客戶13來了.......
pool-1-thread-2客戶1來了.......
pool-1-thread-1客戶3來了.......
pool-1-thread-3客戶4來了.......
pool-1-thread-5客戶5來了.......
pool-1-thread-2客戶6來了.......
pool-1-thread-4客戶7來了.......
pool-1-thread-1客戶8來了.......
pool-1-thread-3客戶9來了.......
pool-1-thread-5客戶10來了.......
pool-1-thread-4客戶11來了.......
pool-1-thread-2客戶12來了.......
pool-1-thread-1客戶15來了.......
pool-1-thread-3客戶16來了.......
pool-1-thread-5客戶17來了.......
pool-1-thread-4客戶19來了.......
pool-1-thread-2客戶18來了.......

ok.全部任務都被線程池執行了.並且咱們自定義的拒絕策略也很簡單,就是讓工做隊列調用put讓其一直等待,直到有可用的容量存聽任務.

打賞

若是以爲個人文章寫的還過得去的話,有錢就捧個錢場,沒錢給我捧我的場(幫我點贊或推薦一下)