線程池使用與機制

之前創(chuàng)建線程的時候都是用的 newCachedThreadPoo,newFixedThreadPool,newScheduledThreadPool,newSingleThreadExecutor 這四個方法禾酱。
當然 Executors 也是用不同的參數(shù)去 new ThreadPoolExecutor 實現(xiàn)的，本文先分析前四種線程創(chuàng)建方式酗宋，后在分析 new ThreadPoolExecutor 創(chuàng)建方式

使用 Executors 創(chuàng)建線程池
1.newFixedThreadPool()
由于使用了LinkedBlockingQueue所以maximumPoolSize沒用业崖，當corePoolSize滿了之后就加入到LinkedBlockingQueue隊列中炭玫。
每當某個線程執(zhí)行完成之后就從LinkedBlockingQueue隊列中取一個弄屡。
所以這個是創(chuàng)建固定大小的線程池昼浦。

源碼分析

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

2.newSingleThreadPool()
創(chuàng)建線程數(shù)為1的線程池，由于使用了LinkedBlockingQueue所以maximumPoolSize 沒用竭沫，corePoolSize為1表示線程數(shù)大小為1,滿了就放入隊列中燥翅，執(zhí)行完了就從隊列取一個。

源碼分析

public static ExecutorService newSingleThreadExecutor() {
    return new Executors.FinalizableDelegatedExecutorService
            (
                    new ThreadPoolExecutor(
                            1,
                            1,
                            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);
}

3.newCachedThreadPool()
創(chuàng)建可緩沖的線程池蜕提。沒有大小限制森书。由于corePoolSize為0所以任務(wù)會放入SynchronousQueue隊列中，SynchronousQueue只能存放大小為1谎势，所以會立刻新起線程凛膏，由于maxumumPoolSize為Integer.MAX_VALUE所以可以認為大小為2147483647。受內(nèi)存大小限制脏榆。

源碼分析

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

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

使用 ThreadPoolExecutor 創(chuàng)建線程池
源碼分析 ,ThreadPoolExecutor 的構(gòu)造函數(shù)

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

構(gòu)造函數(shù)參數(shù)
1猖毫、corePoolSize 核心線程數(shù)大小，當線程數(shù) < corePoolSize 须喂，會創(chuàng)建線程執(zhí)行 runnable

2吁断、maximumPoolSize 最大線程數(shù)趁蕊，當線程數(shù) >= corePoolSize的時候，會把 runnable 放入 workQueue中

3仔役、keepAliveTime 保持存活時間掷伙，當線程數(shù)大于corePoolSize的空閑線程能保持的最大時間。

4骂因、unit 時間單位

5炎咖、workQueue 保存任務(wù)的阻塞隊列

6、threadFactory 創(chuàng)建線程的工廠

7寒波、handler 拒絕策略

任務(wù)執(zhí)行順序
1乘盼、當線程數(shù)小于 corePoolSize時，創(chuàng)建線程執(zhí)行任務(wù)俄烁。

2绸栅、當線程數(shù)大于等于 corePoolSize并且 workQueue 沒有滿時页屠，放入workQueue中

3、線程數(shù)大于等于 corePoolSize并且當 workQueue 滿時辰企，新任務(wù)新建線程運行牢贸，線程總數(shù)要小于 maximumPoolSize

4、當線程總數(shù)等于 maximumPoolSize 并且 workQueue 滿了的時候執(zhí)行 handler 的 rejectedExecution臭增。也就是拒絕策略竹习。

四個拒絕策略
ThreadPoolExecutor默認有四個拒絕策略：

1、ThreadPoolExecutor.AbortPolicy() 直接拋出異常RejectedExecutionException

2拗窃、ThreadPoolExecutor.CallerRunsPolicy() 直接調(diào)用run方法并且阻塞執(zhí)行

3泌辫、ThreadPoolExecutor.DiscardPolicy() 直接丟棄后來的任務(wù)

4甥郑、ThreadPoolExecutor.DiscardOldestPolicy() 丟棄在隊列中隊首的任務(wù)

當然可以自己繼承RejectedExecutionHandler來寫拒絕策略.

TestThreadPoolExecutor 示例
TestThreadPoolExecutor.java
package io.ymq.thread.TestThreadPoolExecutor;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 描述:
 *
 * @author yanpenglei
 * @create 2017-10-12 15:39
 **/
public class TestThreadPoolExecutor {
    public static void main(String[] args) {

        long currentTimeMillis = System.currentTimeMillis();

        // 構(gòu)造一個線程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 6, 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3)
        );

        for (int i = 1; i <= 10; i++) {
            try {
                String task = "task=" + i;
                System.out.println("創(chuàng)建任務(wù)并提交到線程池中：" + task);
                threadPool.execute(new ThreadPoolTask(task));

                Thread.sleep(100);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        try {
            //等待所有線程執(zhí)行完畢當前任務(wù)澜搅。
            threadPool.shutdown();

            boolean loop = true;
            do {
                //等待所有線程執(zhí)行完畢當前任務(wù)結(jié)束
                loop = !threadPool.awaitTermination(2, TimeUnit.SECONDS);//等待2秒
            } while (loop);

            if (loop != true) {
                System.out.println("所有線程執(zhí)行完畢");
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            System.out.println("耗時：" + (System.currentTimeMillis() - currentTimeMillis));
        }


    }
}

ThreadPoolTask.java

package io.ymq.thread.TestThreadPoolExecutor;

import java.io.Serializable;

/**
 * 描述:
 *
 * @author yanpenglei
 * @create 2017-10-12 15:40
 **/
public class ThreadPoolTask implements Runnable, Serializable {

    private Object attachData;

    ThreadPoolTask(Object tasks) {
        this.attachData = tasks;
    }

    public void run() {

        try {

            System.out.println("開始執(zhí)行任務(wù)：" + attachData + "任務(wù)勉躺，使用的線程池，線程名稱：" + Thread.currentThread().getName());

            System.out.println();

        } catch (Exception e) {
            e.printStackTrace();
        }
        attachData = null;
    }

}

遇到j(luò)ava.util.concurrent.RejectedExecutionException

第一

你的線程池 ThreadPoolExecutor 顯示的 shutdown() 之后饵溅，再向線程池提交任務(wù)的時候。如果你配置的拒絕策略是 AbortPolicy 的話咬荷，這個異常就會拋出來轻掩。

第二

當你設(shè)置的任務(wù)緩存隊列過小的時候，或者說唇牧，你的線程池里面所有的線程都在干活（線程數(shù)== maxPoolSize),并且你的任務(wù)緩存隊列也已經(jīng)充滿了等待的隊列，這個時候腔召，你再向它提交任務(wù)，則會拋出這個異常臀蛛。

響應(yīng)

可以看到線程 pool-1-thread-1 到5 循環(huán)使用

創(chuàng)建任務(wù)并提交到線程池中：task=1
開始執(zhí)行任務(wù)：task=1任務(wù)掺栅，使用的線程池纳猪，線程名稱：pool-1-thread-1

創(chuàng)建任務(wù)并提交到線程池中：task=2
開始執(zhí)行任務(wù)：task=2任務(wù)氏堤，使用的線程池，線程名稱：pool-1-thread-2

創(chuàng)建任務(wù)并提交到線程池中：task=3
開始執(zhí)行任務(wù)：task=3任務(wù)闪檬，使用的線程池购笆，線程名稱：pool-1-thread-3

創(chuàng)建任務(wù)并提交到線程池中：task=4
開始執(zhí)行任務(wù)：task=4任務(wù)同欠，使用的線程池样傍，線程名稱：pool-1-thread-4

創(chuàng)建任務(wù)并提交到線程池中：task=5
開始執(zhí)行任務(wù)：task=5任務(wù)衫哥，使用的線程池茎刚，線程名稱：pool-1-thread-5

創(chuàng)建任務(wù)并提交到線程池中：task=6
開始執(zhí)行任務(wù)：task=6任務(wù)膛锭，使用的線程池蚊荣，線程名稱：pool-1-thread-1

創(chuàng)建任務(wù)并提交到線程池中：task=7
開始執(zhí)行任務(wù)：task=7任務(wù)妇押，使用的線程池，線程名稱：pool-1-thread-2

創(chuàng)建任務(wù)并提交到線程池中：task=8
開始執(zhí)行任務(wù)：task=8任務(wù)敲霍，使用的線程池俊马，線程名稱：pool-1-thread-3

創(chuàng)建任務(wù)并提交到線程池中：task=9
開始執(zhí)行任務(wù)：task=9任務(wù)肩杈，使用的線程池柴我，線程名稱：pool-1-thread-4

創(chuàng)建任務(wù)并提交到線程池中：task=10
開始執(zhí)行任務(wù)：task=10任務(wù)，使用的線程池扩然，線程名稱：pool-1-thread-5

所有線程執(zhí)行完畢
耗時：1015