Java线程池初探

电玩女神 2021-12-12 22:09 271阅读 0赞

### 1. 池化技术 ###

池化技术就是提前保存大量的资源，以备不时之需。在机器资源有限的情况下，使用池化技术可以大大的提高资源的利用率，提升性能等。

在编程领域，比较典型的池化技术有：**线程池**、**连接池**、**内存池**、**对象池**等。

池化技术通过预先创建多个资源（如多个线程，多个数据库连接），放在池中，这样可以在需要使用线程的时候直接获取，避免多次重复创建、销毁带来的开销。

### 2. 线程池 ###

线程池顾名思义，就是由很多线程构成的池子，来一个任务，就从池子中取一个线程，处理这个任务。

但是实际上线程池会比这个复杂。例如线程池肯定不会无限扩大的，否则资源会耗尽；当线程数到达一个阶段，提交的任务会被暂时存储在一个队列中，如果队列内容可以不断扩大，极端下也会耗尽资源，那选择什么类型的队列，当队列满如何处理任务，都有涉及很多内容。

线程池总体的工作过程如下图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3phaV94aWE_size_16_color_FFFFFF_t_70]

线程池内的线程数的大小相关的概念有两个，一个是核心池大小，还有最大池大小。  
如果当前的线程个数比核心池个数小，当任务到来，会优先创建一个新的线程并执行任务。  
当已经到达核心池大小，则把任务放入队列，为了资源不被耗尽，队列的最大容量可能也是有上限的，如果达到队列上限则考虑继续创建新线程执行任务，如果此刻线程的个数已经到达最大池上限，则考虑把任务丢弃。

新任务进入线程池的执行策略如下：

*  如果运行的线程少于`corePoolSize`，则`Executor`始终首选添加新的线程直接运行新任务。
 *  如果运行的线程大于等于`corePoolSize`，则`Executor`始终首选将新任务加入队列，不会添加新的线程。
 *  如果无法将新任务加入队列（如队列已满），则创建新的线程，如果创建线程时超过了`maximumPoolSize`，这时拒绝策略会被执行，新任务会被拒绝。

### 3. Java中的实现 ###

#### 3.1 创建线程池 ####

在`java.util.concurrent`包中，提供了ThreadPoolExecutor的实现。

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler) { 
    }

参数含义如下：

**corePoolSize**：核心池大小。在初建线程池时不会立即启动，直到有任务提交过去才会开始启动线程并逐渐将线程数提升到`corePoolSize`，若一开始就创建所有核心线程需调用`prestartAllCoreThreads`方法。

**maximumPoolSize**：池中允许的最大线程数。当核心线程满且阻塞队列也满时才会判断当前线程数是否小于最大线程数，并决定是否创建新线程。

**keepAliveTime**：当线程大于核心时，多余的空闲线程最多存活时间。

**unit**：keepAliveTime参数的时间单位，如`TimeUnit.SECONDS`

**workQueue**： 当线程数目超过核心线程数时用于保存任务的队列。主要有3种类型的BlockingQueue可供选择：无界队列，有界队列和同步移交。参数中可以看到，此队列仅保存实现Runnable接口的任务。

**threadFactory**：执行程序创建新线程时使用的工厂。

**handler**：阻塞队列已满且线程数达到最大值时所采取的饱和策略。java默认提供了4种饱和策略的实现方式：中止、抛弃、抛弃最旧的、调用者运行。

#### 3.2 可供选择的阻塞队列（BlockingQueue） ####

##### 3.2.1 无界队列 #####

队列大小无限制，常用的为无界的`LinkedBlockingQueue`，因为该队列没有大小限制，所以当任务耗时较长时，可能会导致大量新任务在队列中堆积最终导致OOM。

而`Executors.newFixedThreadPool`采用的就是`LinkedBlockingQueue`，可能会导致生产问题！

##### 3.2.2 有界队列 #####

*  遵循FIFO原则的队列如`ArrayBlockingQueue`与有界的`LinkedBlockingQueue`
 *  优先级队列如`PriorityBlockingQueue`。该队列中优先级由任务的`Comparator`决定

使用有界队列时，队列的大小需要与线程池大小配合，线程池较小，有界队列较大时可减少内存消耗，降低CPU使用频率与上下文切换，但可能会限制系统吞吐量。

##### 3.2.3 同步移交队列 #####

如果不希望任务在队列中等待而是希望将任务直接移交给工作线程，可使用`SynchronousQueue`作为等待队列。SynchronousQueue不是一个真正的队列，而是一种线程之间移交的机制。要将一个元素放入`SynchronousQueue`中，必须有另一个线程正在等待接收这个元素。只有在使用无界线程池或者有饱和策略时才建议使用该队列。

#### 3.3 可选择的饱和策略RejectedExecutionHandler ####

##### 3.3.1 AbortPolicy中止策略 #####

该策略是默认饱和拒绝策略

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

使用该策略时在线程池饱和时会抛出RejectedExecutionException（继承自RuntimeException），调用者可自行处理该异常。

##### 3.3.2 DiscardPolicy抛弃策略 #####

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
    }

如代码所示，该策略将直接抛弃该任务

##### 3.3.3 DiscardOldestPolicy抛弃旧任务策略 #####

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
                if (!e.isShutdown()) { 
                    e.getQueue().poll();
                    e.execute(r);
                }
    }

该策略先将阻塞队列中的头元素出队抛弃，再尝试提交任务。如果阻塞队列使用了`PriorityBlockingQueue`优先级队列，那么该策略将导致优先级最高的任务被抛弃，所以在使用优先级队列时，不建议使用该种策略。

##### 3.3.4 CallerRunsPolicy调用者运行策略 #####

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
                if (!e.isShutdown()) { 
                    r.run();
                }
    }

该策略将会将任务退回给调用者，调用者在自己的线程内运行该任务，调用者执行任务的这段时间里主线程无法再提交新任务，从而使线程池中工作线程有时间将正在处理的任务完成。

### 4. Java提供的四种常用线程池解析 ###

#### 4.1 newCachedThreadPool ####

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

在newCachedThreadPool中如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

虽然该线程池初始化时核心线程为0，但是要将一个元素放入SynchronousQueue中，必须有另一个线程正在等待接收这个元素。因此即便SynchronousQueue一开始为空且大小为1，第一个任务也无法放入其中，因为没有线程在等待从SynchronousQueue中取走元素。因此第一个任务到达时便会创建一个新线程执行该任务。

#### 4.2 newFixedThreadPool ####

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

该线程数量固定，使用无限大的队列，容易引发OOM问题。

#### 4.3 newScheduledThreadPool ####

创建一个定长线程池，支持定时及周期性任务执行。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { 
            return new ScheduledThreadPoolExecutor(corePoolSize);
    }

ScheduledThreadPoolExecutor()的构造函数如下：

public ScheduledThreadPoolExecutor(int corePoolSize) { 
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }

ScheduledThreadPoolExecutor的父类即ThreadPoolExecutor，因此这里各参数含义和上面一样。

值得关心的是DelayedWorkQueue这个阻塞对列，在上面没有介绍，它作为静态内部类就在ScheduledThreadPoolExecutor中进行了实现。简单的说，DelayedWorkQueue是一个无界队列，它能按一定的顺序对工作队列中的元素进行排列。

#### 4.4 newSingleThreadExecutor ####

创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

public static ScheduledExecutorService newSingleThreadScheduledExecutor() { 
            return new DelegatedScheduledExecutorService
                (new ScheduledThreadPoolExecutor(1));
     }

首先new了一个线程数目为 1 的ScheduledThreadPoolExecutor，再把该对象传入DelegatedScheduledExecutorService中，看看DelegatedScheduledExecutorService的实现代码：

DelegatedScheduledExecutorService(ScheduledExecutorService executor) { 
                super(executor);
                e = executor;
    }

再看看它的父类

DelegatedExecutorService(ExecutorService executor) {  
               e = executor; 
    }

其实就是使用装饰模式增强了ScheduledExecutorService（1）的功能，不仅确保只有一个线程顺序执行任务，也保证线程意外终止后会重新创建一个线程继续执行任务。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3phaV94aWE_size_16_color_FFFFFF_t_70]: /images/20211210/46db7ca9d35640c88227809288d1e9ed.png