Java多线程系列--“JUC线程池”02之 线程池原理(一)

ThreadPoolExecutor简介

ThreadPoolExecutor是线程池类。对于线程池，可以通俗的将它理解为"存放一定数量线程的一个线程集合。

Executor框架最核心的类是ThreadPoolExecutor，它是线程池的实现类，主要由下列4个组件构成。

• ·corePoolSize：核心线程池的大小。
• ·maximumPool：最大线程池的大小。
• keepAliveTime: 当线程数量大于核心线程池大小的时候，线程（超出核心线程池大小的这部分）可以空闲的最长时间，超过这个时间线程将自动销毁。
• ·workQueue：用来暂时保存任务的工作队列。
• ·RejectedExecutionHandler：当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和时（达到了最大线程池大小且工作队列已满），execute()方法将要调用的Handler。

ThreadPoolExecutor数据结构

ThreadPoolExecutor的数据结构如下图所示：

各个数据在ThreadPoolExecutor.java中的定义如下：

// 阻塞队列。
private final BlockingQueue<Runnable> workQueue;
// 互斥锁
private final ReentrantLock mainLock = new ReentrantLock();
// 线程集合。一个Worker对应一个线程。
private final HashSet<Worker> workers = new HashSet<Worker>();
// “终止条件”，与“mainLock”绑定。
private final Condition termination = mainLock.newCondition();
// 线程池中线程数量曾经达到过的最大值。
private int largestPoolSize;
// 已完成任务数量
private long completedTaskCount;
// ThreadFactory对象，用于创建线程。
private volatile ThreadFactory threadFactory;
// 拒绝策略的处理句柄。
private volatile RejectedExecutionHandler handler;
// 保持线程存活时间。
private volatile long keepAliveTime;

private volatile boolean allowCoreThreadTimeOut;
// 核心池大小
private volatile int corePoolSize;
// 最大池大小
private volatile int maximumPoolSize;

1. workers
    workers是HashSet<Work>类型，即它是一个Worker集合。而一个Worker对应一个线程，也就是说线程池通过workers包含了"一个线程集合"。当Worker对应的线程池启动时，它会执行线程池中的任务；当执行完一个任务后，它会从线程池的阻塞队列中取出一个阻塞的任务来继续运行。
    wokers的作用是，线程池通过它实现了"允许多个线程同时运行"。

2. workQueue
    workQueue是BlockingQueue类型，即它是一个阻塞队列。当线程池中的线程数超过它的容量的时候，线程会进入阻塞队列进行阻塞等待。
    通过workQueue，线程池实现了阻塞功能。

3. mainLock
    mainLock是互斥锁，通过mainLock实现了对线程池的互斥访问。

4. corePoolSize和maximumPoolSize
    corePoolSize是"核心池大小"，maximumPoolSize是"最大池大小"。它们的作用是调整"线程池中实际运行的线程的数量"。
    例如，当新任务提交给线程池时(通过execute方法)。
          -- 如果此时，线程池中运行的线程数量< corePoolSize，则创建新线程来处理请求。
          -- 如果此时，线程池中运行的线程数量> corePoolSize，但是却< maximumPoolSize；则仅当阻塞队列满时才创建新线程。
          如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心池大小和最大池大小的值是在创建线程池设置的；但是，也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。

5. poolSize
    poolSize是当前线程池的实际大小，即线程池中任务的数量。

6. allowCoreThreadTimeOut和keepAliveTime
    allowCoreThreadTimeOut表示是否允许"线程在空闲状态时，仍然能够存活"；而keepAliveTime是当线程池处于空闲状态的时候，超过keepAliveTime时间之后，空闲的线程会被终止。

7. threadFactory
    threadFactory是ThreadFactory对象。它是一个线程工厂类，"线程池通过ThreadFactory创建线程"。

8. handler
    handler是RejectedExecutionHandler类型。它是"线程池拒绝策略"的句柄，也就是说"当某任务添加到线程池中，而线程池拒绝该任务时，线程池会通过handler进行相应的处理"。

newFixedThreadPool

对图10-4的说明如下。

1，如果当前运行的线程数少于corePoolSize， 会立刻创建新线程执行任务。
2，当线程数到达corePoolSize后，将任务加入到LinkedBlockingQueue中。
3，当线程执行完任务后，会循环从LinkedBlockingQueue中获取任务来执行。

FixedThreadPool使用了LinkedBlockingQueue， 也就是无界队列（队列最大可容纳Integer.MAX_VALUE）， 因此会造成以下影响：
a, 线程池线程数到达corePoolSize后，任务会被存放在LinkedBlockingQueue中
b, 因为无界队列，运行中(未调用shutdown()或者shutdownNow()方法)的不会拒绝任务（队列无界，可以放入"无限"任务）

 使用举例：

public class FixedThreadPoolTest {
    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(4*20);
        for (int i = 0; i < 10; i++) {
            final int index = i;
            fixedThreadPool.execute(new Runnable() {
                public void run() {
                    try {
                        System.out.println(Thread.currentThread().getName()+">>"+index);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        
//        // 此方法不阻塞, 调用后不再接收新的任务,所有线程池内任务执行完毕则关闭线程池
//        fixedThreadPool.shutdown();
//        System.out.println("..........分割线.............");
//        fixedThreadPool.execute(new Runnable() {
//            @Override
//            public void run() {
//                System.out.println("run after shutdown.......");
//            }
//        });
        
    }
}

运行结果为：（会阻塞，因为线程池还没关闭）

 如果将上面的注释代码取消注释，运行结果为：

pool-1-thread-2>>1
pool-1-thread-5>>4
pool-1-thread-4>>3
pool-1-thread-1>>0
pool-1-thread-3>>2
pool-1-thread-6>>5
pool-1-thread-7>>6
pool-1-thread-8>>7
pool-1-thread-9>>8
pool-1-thread-10>>9
..........分割线.............
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.test.lesson01.FixedThreadPoolTest$2@4b67cf4d rejected from java.util.concurrent.ThreadPoolExecutor@7ea987ac[Shutting down, pool size = 10, active threads = 10, queued tasks = 0, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
    at com.test.lesson01.FixedThreadPoolTest.main(FixedThreadPoolTest.java:26)

Process finished with exit code 1

不会阻塞

newSingleThreadExecutor

SingleThreadExecutor是使用单个worker线程的Executor

SingleThreadExecutor的corePoolSize和maximumPoolSize被设置为1。其他参数与 FixedThreadPool相同。SingleThreadExecutor使用无界队列LinkedBlockingQueue作为线程池的工 作队列（队列的容量为Integer.MAX_VALUE）。SingleThreadExecutor使用无界队列作为工作队列 对线程池带来的影响与FixedThreadPool相同，这里就不赘述了。

对图10-5的说明如下。
1）如果当前运行的线程数少于corePoolSize（即线程池中无运行的线程），则创建一个新线
程来执行任务。
2）在线程池完成预热之后（当前线程池中有一个运行的线程），将任务加入LinkedBlockingQueue。
3）线程执行完1中的任务后，会在一个无限循环中反复从LinkedBlockingQueue获取任务来
执行。

举例：

public class SingleThreadExecutorTest {

    public static void main(String[] args) {
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            Runnable task = new Runnable() {
                public void run() {
                    try {
                        System.out.println(Thread.currentThread().getName()+">>"+index);
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            singleThreadExecutor.execute(task);
        }
        
        singleThreadExecutor.shutdown();
    }

}

结果为：（每隔1秒就会打印一个输出）

newCachedThreadPool

CachedThreadPool是一个会根据需要创建新线程的线程池。CachedThreadPool的corePoolSize被设置为0，即corePool为空；maximumPoolSize被设置为 Integer.MAX_VALUE，即maximumPool是无界的。这里把keepAliveTime设置为60L，意味着 CachedThreadPool中的空闲线程等待新任务的最长时间为60秒，空闲线程超过60秒后将会被终止。

FixedThreadPool和SingleThreadExecutor使用无界队列LinkedBlockingQueue作为线程池的 工作队列。CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列，但 CachedThreadPool的maximumPool是无界的。这意味着，如果主线程提交任务的速度高于 maximumPool中线程处理任务的速度时，CachedThreadPool会不断创建新线程。极端情况下， CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。

对图10-6的说明如下。
1）首先执行SynchronousQueue.offer（Runnable task）。如果当前maximumPool中有空闲线程 正在执行SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS），那么主线程执行 offer操作与空闲线程执行的poll操作配对成功，主线程把任务交给空闲线程执行，execute()方 法执行完成；否则执行下面的步骤2）。
2）当初始maximumPool为空，或者maximumPool中当前没有空闲线程时，将没有线程执行 SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS）。这种情况下，步骤1）将失 败。此时CachedThreadPool会创建一个新线程执行任务，execute()方法执行完成。
3）在步骤2）中新创建的线程将任务执行完后，会执行 SynchronousQueue.poll（keepAliveTime，TimeUnit.NANOSECONDS）。这个poll操作会让空闲线
程最多在SynchronousQueue中等待60秒钟。如果60秒钟内主线程提交了一个新任务（主线程执 行步骤1）），那么这个空闲线程将执行主线程提交的新任务；否则，这个空闲线程将终止。由于 空闲60秒的空闲线程会被终止，因此长时间保持空闲的CachedThreadPool不会使用任何资源。
前面提到过，SynchronousQueue是一个没有容量的阻塞队列。每个插入操作必须等待另一 个线程的对应移除操作，反之亦然。CachedThreadPool使用SynchronousQueue，把主线程提交的 任务传递给空闲线程执行。CachedThreadPool中任务传递的示意图如图10-7所示。

使用举例：

package com.test.lesson01;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolTest {

    public static void main(String[] args) {
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            /*try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/
            Runnable task = new Runnable() {
                public void run() {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName()+">>"+index+","+Thread.currentThread().hashCode());
                }
            };
            cachedThreadPool.execute(task);
        }
    }

}

结果为：

pool-1-thread-9>>8,1119372326
pool-1-thread-5>>4,2118164867
pool-1-thread-8>>7,1955604205
pool-1-thread-6>>5,1242746377
pool-1-thread-1>>0,1296698618
pool-1-thread-4>>3,188351009
pool-1-thread-10>>9,1881928750
pool-1-thread-7>>6,282966402
pool-1-thread-3>>2,1447026184
pool-1-thread-2>>1,278180759

并且，上述运行在1分钟后就会终止，因为CachedThreadPool在没有任务继续提交的时候就会关闭线程池。上面的线程的名字都是一样的，是因为任务执行时间很短，任务已经执行完了，线程池都还没有关闭，下一个任务来了直接可以采用刚刚的缓存的线程来执行任务。

如果将上面的休眠时间取消注释，将会出现如下结果：

pool-1-thread-1>>0,1986922183
pool-1-thread-2>>1,903833002
pool-1-thread-1>>2,1986922183
pool-1-thread-2>>3,903833002
pool-1-thread-1>>4,1986922183
pool-1-thread-2>>5,903833002
pool-1-thread-1>>6,1986922183
pool-1-thread-2>>7,903833002
pool-1-thread-1>>8,1986922183
pool-1-thread-2>>9,903833002

出现这种情况是因为当下次任务来的时候，上次的任务还没执行结束，线程池中只好再新建立一个线程来执行这个新到的任务。当没有任务需要执行的时候，程序运行将在1分钟后自动结束。

参考文献：

《Java并发编程艺术》

https://www.cnblogs.com/skywang12345/p/3509941.html