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

Posted 2021-01-19 Hermioner

 

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()方法)的不会拒绝任务（队列无界，可以放入"无限"任务）

 使用举例：

 1 public class FixedThreadPoolTest {
 2     public static void main(String[] args) {
 3         ExecutorService fixedThreadPool = Executors.newFixedThreadPool(4*20);
 4         for (int i = 0; i < 10; i++) {
 5             final int index = i;
 6             fixedThreadPool.execute(new Runnable() {
 7                 public void run() {
 8                     try {
 9                         System.out.println(Thread.currentThread().getName()+">>"+index);
10                         Thread.sleep(2000);
11                     } catch (InterruptedException e) {
12                         e.printStackTrace();
13                     }
14                 }
15             });
16         }
17         
18 //        // 此方法不阻塞, 调用后不再接收新的任务,所有线程池内任务执行完毕则关闭线程池
19 //        fixedThreadPool.shutdown();
20 //        System.out.println("..........分割线.............");
21 //        fixedThreadPool.execute(new Runnable() {
22 //            @Override
23 //            public void run() {
24 //                System.out.println("run after shutdown.......");
25 //            }
26 //        });
27         
28     }
29 }

View Code

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

 

 

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

 1 pool-1-thread-2>>1
 2 pool-1-thread-5>>4
 3 pool-1-thread-4>>3
 4 pool-1-thread-1>>0
 5 pool-1-thread-3>>2
 6 pool-1-thread-6>>5
 7 pool-1-thread-7>>6
 8 pool-1-thread-8>>7
 9 pool-1-thread-9>>8
10 pool-1-thread-10>>9
11 ..........分割线.............
12 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]
13     at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
14     at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
15     at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
16     at com.test.lesson01.FixedThreadPoolTest.main(FixedThreadPoolTest.java:26)
17 
18 Process finished with exit code 1

View Code

不会阻塞

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获取任务来
执行。

举例：

 1 public class SingleThreadExecutorTest {
 2 
 3     public static void main(String[] args) {
 4         ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
 5         for (int i = 0; i < 10; i++) {
 6             final int index = i;
 7             Runnable task = new Runnable() {
 8                 public void run() {
 9                     try {
10                         System.out.println(Thread.currentThread().getName()+">>"+index);
11                         Thread.sleep(1000);
12                     } catch (InterruptedException e) {
13                         e.printStackTrace();
14                     }
15                 }
16             };
17             singleThreadExecutor.execute(task);
18         }
19         
20         singleThreadExecutor.shutdown();
21     }
22 
23 }

View Code

结果为：（每隔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所示。

 

 

使用举例：

 1 package com.test.lesson01;
 2 
 3 import java.util.concurrent.ExecutorService;
 4 import java.util.concurrent.Executors;
 5 
 6 public class CachedThreadPoolTest {
 7 
 8     public static void main(String[] args) {
 9         ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
10         for (int i = 0; i < 10; i++) {
11             final int index = i;
12             /*try {
13                 Thread.sleep(1000);
14             } catch (InterruptedException e) {
15                 e.printStackTrace();
16             }*/
17             Runnable task = new Runnable() {
18                 public void run() {
19                     try {
20                         Thread.sleep(1000);
21                     } catch (InterruptedException e) {
22                         e.printStackTrace();
23                     }
24                     System.out.println(Thread.currentThread().getName()+">>"+index+","+Thread.currentThread().hashCode());
25                 }
26             };
27             cachedThreadPool.execute(task);
28         }
29     }
30 
31 }

View Code

结果为：

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

View Code

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

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

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

View Code

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

 

 

参考文献：

《Java并发编程艺术》

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