源碼解析Executors.newFixedThreadPool(int)

source 發(fā)布于2019-08-16 11:11 / 2150人閱讀

摘要：創(chuàng)建一個(gè)線(xiàn)程池，具有固定線(xiàn)程數(shù)，運(yùn)行在共享的無(wú)界隊(duì)列中。固定線(xiàn)程數(shù)源碼如下是的實(shí)現(xiàn)類(lèi)。線(xiàn)程池中允許最大的線(xiàn)程數(shù)。如果線(xiàn)程數(shù)超過(guò)了核心線(xiàn)程數(shù)，過(guò)量的線(xiàn)程在關(guān)閉前等待新任務(wù)的最大時(shí)間。處理因?yàn)榫€(xiàn)程邊界和隊(duì)列容量導(dǎo)致的堵塞。

1.Executors.newFixedThreadPool(int nThreads)：創(chuàng)建一個(gè)線(xiàn)程池，具有固定線(xiàn)程數(shù)，運(yùn)行在共享的無(wú)界隊(duì)列中。在大多數(shù)時(shí)候，線(xiàn)程會(huì)主動(dòng)執(zhí)行任務(wù)，當(dāng)所有的線(xiàn)程都在執(zhí)行任務(wù)時(shí)，有新的任務(wù)加入進(jìn)來(lái)，就會(huì)進(jìn)入等待隊(duì)列(可以有源源不斷的任務(wù)加入進(jìn)來(lái)，因?yàn)槭菬o(wú)界隊(duì)列)，當(dāng)有空閑的線(xiàn)程，等待隊(duì)列中的任務(wù)就會(huì)被執(zhí)行。如果有線(xiàn)程在執(zhí)行過(guò)程中因?yàn)閳?zhí)行失敗要關(guān)閉，新創(chuàng)建的線(xiàn)程會(huì)替失敗的線(xiàn)程執(zhí)行接下來(lái)的任務(wù)。如果想要關(guān)閉這個(gè)線(xiàn)程池，可以調(diào)用ExecutorService的shutDown方法。

nThreads 固定線(xiàn)程數(shù)

源碼如下：

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

2.ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue workQueue):是ExecutorService的實(shí)現(xiàn)類(lèi)。

corePoolSize 核心線(xiàn)程數(shù)，即使是空閑的時(shí)候，線(xiàn)程池中也會(huì)保留線(xiàn)程的數(shù)量。如果設(shè)置allowCoreThreadTimeOut為true，會(huì)使用keepAliveTime作為等待工作時(shí)間，默認(rèn)為false。

maximumPoolSize 線(xiàn)程池中允許最大的線(xiàn)程數(shù)。

keepAliveTime 如果線(xiàn)程數(shù)超過(guò)了核心線(xiàn)程數(shù)，過(guò)量的線(xiàn)程在關(guān)閉前等待新任務(wù)的最大時(shí)間。

unit keepAliveTime 參數(shù)的時(shí)間單元。

workQueue 用來(lái)存放等待執(zhí)行的任務(wù)，這些任務(wù)是通過(guò)execute方法提交的Runnable任務(wù)。

源碼如下：

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

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

threadFactory 用來(lái)創(chuàng)建新線(xiàn)程的工廠。

handler 處理因?yàn)榫€(xiàn)程邊界和隊(duì)列容量導(dǎo)致的堵塞。

源碼如下：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue 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.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

4.LinkedBlockingQueue:Executors.newFixedThreadPool()使用基于鏈表結(jié)點(diǎn)的無(wú)界隊(duì)列LinkedBlockingQueue存儲(chǔ)待執(zhí)行的任務(wù)。繼承了AbstractQueue類(lèi)，實(shí)現(xiàn)了BlockingQueue接口，采用先進(jìn)先出的排列方式，頭結(jié)點(diǎn)是入隊(duì)時(shí)間最長(zhǎng)的元素，尾結(jié)點(diǎn)是入隊(duì)時(shí)間最短的元素。新結(jié)點(diǎn)添加到隊(duì)尾，從隊(duì)頭彈出結(jié)點(diǎn)。鏈表隊(duì)列的特點(diǎn)是：跟基于數(shù)組的隊(duì)列相比有更大的吞吐量，但在大多并發(fā)應(yīng)用中性能會(huì)比較差。LinkedBlockingQueue可以在創(chuàng)建的時(shí)候傳遞一個(gè)容量參數(shù)，限制隊(duì)列的長(zhǎng)度，不設(shè)定的情況下，默認(rèn)是Integer.MAX_VALUE。在沒(méi)有超過(guò)隊(duì)列邊界的情況下，每次添加會(huì)自動(dòng)創(chuàng)建鏈表結(jié)點(diǎn)。

源碼如下：

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

public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node(null);
}

5.DefaultThreadFactory:返回默認(rèn)的線(xiàn)程工廠創(chuàng)建新線(xiàn)程。執(zhí)行器Executor在同一個(gè)線(xiàn)程組中創(chuàng)建所有新的線(xiàn)程。如果存在SecurityManager，就使用SecurityManager的線(xiàn)程組，否則使用當(dāng)前線(xiàn)程的線(xiàn)程組。每一個(gè)新創(chuàng)建的線(xiàn)程作為非守護(hù)線(xiàn)程，其優(yōu)先級(jí)設(shè)置為T(mén)hread.NORM_PRIORITY和線(xiàn)程組允許的最大優(yōu)先級(jí)的較小者?？梢酝ㄟ^(guò)Thread.getName()獲取線(xiàn)程的名稱(chēng)，形式如：pool-N-thread-M，N表示線(xiàn)程工廠的序列號(hào)，M表示線(xiàn)程工廠創(chuàng)建的線(xiàn)程的序列號(hào)。

源碼如下：

public static ThreadFactory defaultThreadFactory() {
    return new DefaultThreadFactory();
}

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    ......
}

6.AbortPolicy：默認(rèn)的拒絕執(zhí)行處理器。拋出一個(gè)RejectedExecutionException。

源碼如下：

private static final RejectedExecutionHandler defaultHandler =
    new AbortPolicy();

 public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        throw new RejectedExecutionException("Task " + r.toString() +
                                             " rejected from " +
                                             e.toString());
    }
}

7.例子

@Test
public void testFixedDemo01() {
    ExecutorService executorService = Executors.newFixedThreadPool(4);
    for(int i = 0;i < 10;i++) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + " "  +" 走啦");
            }
        });
    }
}

運(yùn)行結(jié)果：

核心線(xiàn)程數(shù)和最大線(xiàn)程數(shù)都是4，創(chuàng)建了一個(gè)線(xiàn)程工廠。