JUC之ThreadPoolExecutor（1）总体介绍

偏执的太偏执、 2023-01-23 03:57 124阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

**ThreadPoolExecutor**是Java中提供线程池功能的一个类，在类中实现了线程池的功能。不仅仅是线程池，还是其他的池化技术，比如数据库连接池等池化技术，实现思想都大致相同。它们都是用**空间去换取时间**。线程池在内存中提前准备好处于NEW状态的线程，当需要执行任务的时候便可以直接从线程池中拿出一个线程使用，当任务执行完成之后，再将线程归还到线程池中。这样的一个过程减少了线程创建和销毁的一个开销，能够提高运行效率，但在内存中提前准备好线程而不去销毁自然也就需要消耗一部分内存了。

--------------------

# **█ **Executor #

ThreadPoolExecutor间接继承了接口**Executor**，Executor中文翻译为“执行者”，执行什么呢？当然是执行线程了。在Executor中定义了一个方法**execute**便是执行线程的入口，该方法的参数类型为**Runnable**。

public interface Executor {
    
        void execute(Runnable command);
    }

execute的功能是在将来的某个时间执行Runnable中的**run**方法。取决于Executor子类对该方法的实现，所以可以实现在一个新的线程中、或在线程池中，或在调用该方法的线程中去执行run方法。如果参数command为null，方法则抛出空指针异常，如果无法继续执行run方法（比如线程池不够用的情况）则抛出**RejectedExecutionException**异常。

①在新的线程中执行command。

每次执行command，都创建新的线程，在新的线程中去执行。

public class NewThread implements Executor {
        @Override
        public void execute(Runnable command) {
            new Thread(command).start();
        }
    }

②在线程池中去执行command。

此类使用线程池的功能便是ThreadPoolExecutor的实现。在学习过程中慢慢了解ThreadPoolExecutor是如何实现的。

public class ThreadPool implements Executor {
    
        private BlockingQueue<Thread> threadsPool = new LinkedBlockingQueue<>();
    
        @Override
        public void execute(Runnable command) {
            // 利用线程池池化技术去执行command
           
        }
    }

③在调用方法的线程中去执行command。

在①②中，执行方法与调用方法的线程不是一个线程，可以看做是异步执行方法。此种使用调用方法的线程去执行方法，便是同步执行了。

public class InvokeThread implements Executor {
        @Override
        public void execute(Runnable command) {
            command.run();
        }
    }

--------------------

# **█ **ExecutorService #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

接口**ExecutorService**继承了Executor，定义了更多的方法，对Executor进行扩展。ExecutorService提供了方法可以关闭线程池等功能。

public interface ExecutorService extends Executor {
    
        void shutdown();
    
        List<Runnable> shutdownNow();
    
        boolean isShutdown();
    
        boolean isTerminated();
    
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
    
        <T> Future<T> submit(Callable<T> task);
    
        <T> Future<T> submit(Runnable task, T result);
    
        Future<?> submit(Runnable task);
    
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException;
    
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                      long timeout, TimeUnit unit)
            throws InterruptedException;
    
        <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException;
    
        <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                        long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

*  **shutdown**

关闭线程池。执行了该方法后，线程池不会再接受新的任务，但是先前已经提交到线程池中的任务还会被执行。

*  **shutdownNow**

关闭线程池。执行了该方法后，线程池不会再接受新的任务，并且先前已经提交到线程池中的任务也不会被执行。

*  **isShutdown**

线程池是否已经关闭，返回true表示线程池已关闭，返回false表示线程池没有被关闭。

*  **isTerminated**

在线程池执行关闭（shutdown）之后，线程池中所有的任务是否已经完成。返回true表示任务全部完成。

*  **awaitTermination**

boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;

执行该方法会一直阻塞，当满足以下条件是方法会返回。

①执行关闭线程池请求后，线程池中所有的任务都被执行完成。

②设置的时间参数达到超时时长。

③当前线程被阻断（interrupted）

*  **submit**

①参数为**Callable**，任务带有返回值

<T> Future<T> submit(Callable<T> task);

提交一个任务到线程池中去执行。task为Callable类型，可以有返回值。方法返回**Future**对象，可以通过Future的get方法去获取task的返回值。

②参数为Runnable，任务不带返回值，但可以指定执行结果。

<T> Future<T> submit(Runnable task, T result);

提交一个任务到线程池中去执行。task为Runnable类型，不带有返回值。方法的第二个参数result用来定义当task执行完成执行后，Future的get方法返回的内容。

③参数为Runnable，任务不带返回值，并且不能指定执行结果。

Future<?> submit(Runnable task);

与②中可以指定任务执行结果不同，此方法不能指定执行结果。当task任务执行完成后，Future的get方法返回null。

*（关于Future的get方法，如果任务（Callable或Runnable）没有执行完成，调用get方法会一直阻塞等待，直到任务执行完成之后才会有返回。）*

*  **invokeAll**

①一直阻塞等待直到所有任务都被执行完成。

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;

submit方法每次只能提交一个任务到线程池中去执行，而invokeAll则提供了批量提交的功能，一次提交多个任务去线程池中去执行。当集合中所有的task都被执行完成后，方法返回List<Future<T>>。

②在超时时间内阻塞等待。

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

带有最大超时等待时间的批量提交任务。在超过设置的超时时间后，如果集合中的任务还没有被执行，则这些任务将被取消执行。方法返回在超时时间内已经执行完成的任务所对应的List<Future<T>>。

*  **invokeAny**

<T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

批量提交任务到线程池中去执行，只要有一个任务执行完成，方法就返回，返回其对应的结果。

<T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;

待有超时等待时间的执行任务。在设置的超时等待时间内等待其中一个任务执行完成。

--------------------

# **█ **AbstractExecutorService #

抽象类**AbstractExecutorService**实现了ExecutorService接口，对ExecutorService中定义的部分方法提供了实现以此形成模板方法。某些方法则由具体的实现类去实现。ThreadPoolExecutor直接继承了AbstractExecutorService，关于AbstractExecutorService中的方法，将在介绍ThreadPoolExecutor的方法中去一同介绍。

public abstract class AbstractExecutorService implements ExecutorService

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

--------------------

# ** █ ThreadPoolExecutor** #

![20210712213000941.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3][]

### **构造器** ###

在ThreadPoolExecutor中重写了四个构造器，分别为：

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

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

从构造器中可以看出线程池需要7大核心参数：

*  **corePoolSize**

线程池的核心worker的数量，而worker则负责执行被提交到线程池中的任务。核心worker一旦被创建之后便不会被销毁，永远保留在线程池中等待下一次执行任务。

*  **maximumPoolSize**

线程池允许的最多worker的数量。这里的worker数量不仅包含了核心worker，而且还包含了临时工worker。关于临时工worker，当没有任务执行可执行空闲下来时，这些临时工worker要被销毁的。

*  **keepAliveTime**

在前面提到临时工worker，当空闲下来需要将他们销毁。那多久没干活了叫空闲呢？这个参数就是用来定义空闲时长的，当达到空闲时长还没有任务要执行的话，就将临时工worker回收了。

*  **unit**

keepAliveTime定义了空闲时间的长度，而unit则定义空闲时间的单位，是毫秒还是秒还是小时呢？

*  **workQueue**

等待队列。当核心worker已经来不及处理任务了（所有的核心worker都在处理任务中），新加的任务需要先存放到等待队列中，等待核心worker去处理。workQueue的类型为BlockingQueue。

*  **threadFactory**

创建核心线程与临时工线程的线程工厂。

*  **handler**

拒绝策略。当核心worker都在工作，等待队列已经被任务塞满了，并且临时工worker也都在工作，此时线程池已经没有余力来执行新提交的任务了。此时则需要定义一个拒绝策略，如何处理新提交的任务。

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.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;
    }

从上面的逻辑可以看出，线程池的几个核心参数需要满足以下条件：

①corePoolSize，核心worker的数量不能小于0。

②maximumPoolSize，最大worker的数量必须大于0，即核心worker与临时工worker的数量之和必须大于0，也就是必须存在能干活的worker。

③maximumPoolSize不能小于corePoolSize。即最大的worker数量不能比核心worker的数量少。

④keepAliveTime，空闲时长不能小于0。

⑤workQueue等待队列、threadFactory创建worker的工厂、handler拒绝策略都不允许为空，必须指定。

--------------------

举一个简单的例子来模拟线程池的实现逻辑，理解了这个过程方便理解ThreadPoolExecutor实现线程池的逻辑。

①银行开放了三个柜台（核心worker，即corePoolSize=3）以便为用户提供服务

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4][]

②此时进来了一个用户办理业务，银行便派一个工作人员为该用户提供服务。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 5][]

③若用户办理业务很快，在办理期间没有其他的用户进来。等该用户走了之后，进来的一个用户可以继续在该柜台办理业务。

④若在一个用户办理业务期间又进来了其他用户，或者说一次进来了多个用户。但是此时用户的数量还没有超过开通柜台的数量。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 6][]

⑤此时的柜台工作人员和用户还能维持一对一的关系。如果此时还有其他用户进来了，但是已经没有剩余的工作人员可以提供服务了。那怎么办？可以设置一个等待休息区（workQueue，等待队列，大小为4），提供几把椅子让用户在此等待。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 7][]

⑥此时若用户还一直进来，休息区都坐不下了呢？解决方式是增加几个临时窗口（临时worker）来增加工作人员数量，这样能够加到处理用户的事情，让用户造点办完事情早点离开。但临时柜台的数量也不能是无限增加的。（maximumPoolSize限制了总的柜台数量，也就限制了核心worker与临时worker的数量）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 8][] ⑦若休息区都满了，临时的柜台也增加了，可还是无法匹配用户的数量。此时当新来的用户只能拒绝了，说现在暂时无法提供服务，等一会再来吧。

⑧当高峰期过了之后，用户慢慢少了。此时原先的三个窗口完全能够应付用户的任务。那新增加的临时窗口则需要撤销了，只保留原先的三个窗口的工作人员即可。

上面的例子大致描述了ThreadPoolExecutor中线程池的实现逻辑，想想和生活中去银行办理业务是不是很相似。当人很少时，你到了银行不用排队，可以直接去柜台办理业务。当人很多时，你到了银行需要先取个号在休息区等待，等待柜员叫号，叫到你了你再去办理业务。不过区别的是，当来银行办理业务的人很多时，银行不会增加临时窗口来解决的，反正我没遇到过，哈哈！

--------------------

### **workQueue** ###

等待队列。在ThreadPoolExecutor中提供了一个BlockingQueue类型的字段作为等待队列，用来存放需要等待worker去执行的任务。

private final BlockingQueue<Runnable> workQueue;

*（关于BlockingQueue的介绍，请戳《[BlockingQueue初识][BlockingQueue]》《[BlockingQueue之ArrayBlockingQueue][BlockingQueue_ArrayBlockingQueue]》《[BlockingQueue之LinkedBlockingQueue][BlockingQueue_LinkedBlockingQueue]》）*

### **ThreadFactory** ###

线程工厂，用来创建一个线程，这里的一个线程就是一个worker。一个worker对应一个线程。

public interface ThreadFactory {
    
        Thread newThread(Runnable r);
    }

可以自定义ThreadFactory实现类来实现创建线程的逻辑，如果不指定自定义的实现类，ThreadPoolExecutor中也提供了ThreadFactory默认实现，在类**Executors**定义了一个静态内部类**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-";
        }
    
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

### **RejectedExecutionHandler** ###

拒绝策略。当新提交的任务太多了，超过了线程池的处理能力，那后面新提交的任务就需要拒绝策略来处理了。

public interface RejectedExecutionHandler {
    
        void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
    }

ThreadPoolExecutor中提供了四种拒绝策略，分别对应四个**RejectedExecutionHandler**实现类。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 9][]

*  **AbortPolicy**

直接抛出异常。此种拒绝策略也是ThreadPoolExecutor的**默认拒绝策略**。

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());
        }
    }

*  **DiscardPolicy**

啥都不做，直接丢弃。该类的实现特别有意思，方法没有逻辑，就一个空的实现。

public static class DiscardPolicy implements RejectedExecutionHandler {
        
        public DiscardPolicy() { }
    
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }
    }

*  **DiscardOldestPolicy**

将等待队列中等待时间最长的那个任务丢弃了，然后执行此时提交的任务。e.getQueue().poll()，也就是workerQueue.poll()会取出等待队列中的第一个等待元素，即等待时间最长的那个任务。

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

*  **CallerRunsPolicy**

将该任务交给调用线程去处理，即调用了ThreadPoolExecutor的execute方法提交了任务到线程池的那个线程。直接执行了r.run()，此方式没有产生新的线程，在调用线程中去执行run方法。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
    
        public CallerRunsPolicy() { }
    
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221004/60385f270a3c4c0db226dca1df5d0f6f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221004/fc0403585b1748f38509fa3f816f6c26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20221004/b5b2f41e896641cbb1ebde8bcefcd6a0.png
[20210712213000941.png]: /images/20221004/d4260d0a038d4b0ca1b428276196548d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 3]: /images/20221004/d5ba88ba5c8f4d23b40d974ea5df3527.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 4]: /images/20221004/b0667cfa5afd4c4194826364726f4409.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 5]: /images/20221004/939bd460059d46649b753e2d9cc4b939.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 6]: /images/20221004/e41cb4cb83724187bddd2a3e0f250fb2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 7]: /images/20221004/199de73353ef4a8180f809fc9eb27da0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 8]: /images/20221004/3160e77472eb43989adf9a4253dc10fe.png
[BlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/117789512?spm=1001.2014.3001.5501
[BlockingQueue_ArrayBlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/118162245?spm=1001.2014.3001.5501
[BlockingQueue_LinkedBlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/118195710?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 9]: /images/20221004/4351d4ffee7541c099ba4d1d3d820080.png