多线程基础学习五：线程池基础学习

柔情只为你懂 2022-06-07 05:12 309阅读 0赞

今天学习一下java的线程池的基础知识。  
我的工作中基本上没有涉及到并发的问题，开发的都是XXX后台管理系统之类的项目，用户就几十个人，不存需要使用线程池的场景。

我看了一下线程池相关的类和接口都在java.util.concurrent包中，其中Callable接口也在这里面。

### 顶级接口Executor ###

通过ThreadPoolExecutor这个类向上查找了解到，ThreadPoolExecutor继承了AbstractExecutorService这个抽象类，AbstractExecutorService实现了ExecutorService接口，ExecutorService继承了Executor，所以我觉的有必要了解一下这个接口。

在Java源码中可以看到这个接口只有一个方法：

void execute(Runnable command)

从方法名和参数上可以看到，执行一个Runnable线程，非常的简洁，具体怎么实现，就由子类决定了。  
如果用快捷键看一下它的子类：  
![子类][SouthEast]

有的是抽象类，有的是接口，也有实现的子类。  
在Java api文档的上明确标出常用的两个接口和四个实现类，  
接口分别是：  
– **ExecutorService**  
– **ScheduledExecutorService**

实现类：  
– **AbstractExecutorService**  
– **ForkJoinPool**  
– **ScheduledThreadPoolExecutor**  
– **ThreadPoolExecutor**

这是文档上关于这个类的注释：

An object that executes submitted Runnable tasks. This interface provides a way of decoupling task submission from the mechanics of how each task will be run, including details of thread use, scheduling, etc. An Executor is normally used instead of explicitly creating threads. For example, rather than invoking new Thread(new(RunnableTask())).start() for each of a set of tasks, you might use:
     Executor executor = anExecutor;
     executor.execute(new RunnableTask1());
     executor.execute(new RunnableTask2());

中文翻译（不保证百分百正确）：

一个执行提交的Runnable任务的类。这个接口提供了一种从任务如何运行的机制中分离任务提交的方法，包括线程的使用、调度等细节等。一个Executor通常更常用，而不是创建新的线程。例如，执行线程可以这样：
     Executor executor = anExecutor;
     executor.execute(new RunnableTask1());
     executor.execute(new RunnableTask2());
    而不是new RunnableTask.start()

现在，对这个接口有基本的理解了。

### ExecutorService ###

这个接口继承了Executor，提供结束的方法和返回值的方法。  
它有这些方法：  
![这里写图片描述][SouthEast 1]

可以非常容易理解有三类方法，结束方法，提交任务方法，执行任务方法；方法带时间参数的，都是超时时间。

– **void shutdown()**  
在先前提交执行的任务中，启动一个结束命令；方法执行后，不再接受新的任务。  
这个方法应该会执行完所有提交的任务再结束。  
– **List shutdownNow()**  
尝试停止所有正在积极执行的任务，停止所有正在等待的任务，并返回这些等待的任务。  
但是不是所有执行的任务都会被终结，对中断没有响应的任务，可能永远都不会结束。  
从注释上看，应该是通过线程中断的方法结束任务的，所以才会出现对中断没响应的任务不会结束。  
– **awaitTermination**  
调用方法后，会处于堵塞状态直到所有任务执行完成或者超过等待时间，或者遇到中断抛出中断异常。  
注释上没说是否还会接收新的任务，但是个人理解，处于堵塞状态的话，应该也不能接收新任务了。

另外从注释上看到：

@return {
       @code true} if this executor terminated and
            {
       @code false} if the timeout elapsed before termination

执行器正常执行任务结束，会返回true，如果超时结束，返回false。

– **Future submit(Callable task)**  
提交一个Callable任务，返回值为泛型为返回值的Future，通过Future的get方法可以获取到返回值，当然也可能为抛出异常。

注释上提到一种使用方式：

If you would like to immediately block waiting
         * for a task, you can use constructions of the form
         * {@code result = exec.submit(aCallable).get();}

这句话的意思是，如果想为了一个任务，立刻阻塞等待，可以这么写：

result = exec.submit(aCallable).get();

这样写的话，按照我的理解，是因为Future的get方法导致的阻塞，导致后面的代码无法执行。

–  **Future submit(Runnable task, T result)**  
看到这个方法的时候，觉得特别奇怪，提交的是Runnable，怎么能返回数据呢；因为这只是接口方法，没有实现，也不知道怎么实现的。

ps：后来看了AbstractExecutorService类中的默认实现，才知道通过工具类中的Executors方法，使用Callable的实现，把task和result包装成了Callable。

– **Future**

### ScheduledExecutorService ###

这个相对而言，方法少一点，只有四个：

![这里写图片描述][SouthEast 2]

前两个方法一看就明白，就是延迟执行一个任务。  
后两个方法的区别：  
– scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)  
这个方法含义延迟initialDelay执行任务，两次成功执行任务之间的间隔是initialDelay + 2 \* period。  
– scheduleWithFixedDelay(Runnable command, long initialDelay, long delay,TimeUnit unit)  
这个是一次任务的结束到下次任务的开始之间，间隔delay时间。

### AbstractExecutorService ###

这个类是ExecutorService的默认实现，实现了提交方法、执行方法，没有实现结束方法。

实现的方法：  
![这里写图片描述][SouthEast 3]

### ForkJoinPool ###

目前对Fork和Join基本不了解，有基本了解之后，再来看这个类。

### ScheduledThreadPoolExecutor ###

目前基本的都不会用，就先不看这个类了。

### ThreadPoolExecutor ###

这是本次学习的最终目标，了解这个类的基本情况和怎么使用。  
这个类继承了AbstractExecutorService，AbstractExecutorService实现了提交任务和执行任务，这两类方法在ThreadPoolExecutor中没有重写，使用的是AbstractExecutorService中的默认实现。

在这个类中定义了这些实例变量：  
![这里写图片描述][SouthEast 4]

以我可怜的英语水平，大概也能看懂这些变量的含义，主要的有一下三个变量：  
– corePoolSize  
核心池数量，池中保持存货的最小数量，不会出现超时的情况。  
– maximumPoolSize  
池中任务最大的数量 ，int类型，所以最大不会超过Integer.MAX\_VALUE  
– keepAliveTime  
空闲线程的最大等待时间。

构造方法：

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,
                                  ThreadFactory threadFactory) {
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 threadFactory, 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);
        }
    
    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.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }

从变量上看，最终调用的都是最后一个方法，总共六个变量：  
– long corePoolSize  
– long maximumPoolSize  
– long keepAliveTime  
– TimeUnit unit  
– BlockingQueue workQueue  
– ThreadFactory threadFactory  
– RejectedExecutionHandler handler  
前四个变量都是基本的参数，workQueue是线程安全的任务阻塞队列，threadFactory的线程工程，handler是用来处理执行被阻塞的问题的。

默认的threadFactory是通过工具类Executors.defaultThreadFactory()的方法获取的，  
默认的handler是AbortPolicy，RejectedExecutionHandler有四种实现，都是ThreadPoolExecutor的内部类，分别是CallerRunsPolicy， AbortPolicy，DiscardPolicy，DiscardOldestPolicy；目前对这个东西了解不多，也没有用过。

到现在为止，我对线程池基础的东西，基本了解了。

### ThreadPoolExecutor的使用 ###

写一个简单的测试demo：

public class SimplePoolTest { 
    
        public static void main (String[] args) {
    
             BlockingDeque<Runnable> runnableBlockingDeque = new LinkedBlockingDeque<Runnable>();    
    
            for (int i = 0; i < 10; i++) {
                runnableBlockingDeque.add(new TestRunnable("Thread-" + i));            
            }
    
            ExecutorService  executorService = new ThreadPoolExecutor(2, 4, 5, TimeUnit.SECONDS,runnableBlockingDeque);    
    
            executorService.shutdown();
    
        }
    
        static class TestRunnable implements Runnable { 
    
            private String name;
    
            public TestRunnable (String name) {
                this.name = name;
            }
    
            @Override
            public void run () {
    
                //SleepUtil.sleep(3000);
                System.out.println(name);
            }
        }    
    }

测试发现，执行结果没有输出任何数据。后来百度了一下，发现BlockingQueue这个参数我用错了，有看了下源码的注释：

the queue to use for holding tasks before they are
         *        executed.  This queue will hold only the {@code Runnable}
         *        tasks submitted by the {@code execute} method

这个参数使用来，保存通过execute提交的任务的。测试代码修改成下面的样子：

public static void main (String[] args) {
    
            ArrayBlockingQueue<Runnable> runnableBlockingDeque = new ArrayBlockingQueue<Runnable>(2);
            ExecutorService  executorService = new ThreadPoolExecutor(2, 4, 5, TimeUnit.SECONDS, runnableBlockingDeque);
            for (int i = 0; i < 10; i++) {
                executorService.execute(new TestRunnable("Thread-" + i));
            }   
    
            executorService.shutdown();        
        }

执行结果如下：

Thread-0
    Thread-2
    Thread-3
    Thread-1
    Thread-4
    Thread-6
    Thread-7
    Thread-8
    Thread-5
    Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task learn.pooldemo.SimplePoolTest$TestRunnable@6d6f6e28 rejected from java.util.concurrent.ThreadPoolExecutor@135fbaa4[Running, pool size = 4, active threads = 0, queued tasks = 0, completed tasks = 9]
        at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
        at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
        at learn.pooldemo.SimplePoolTest.main(SimplePoolTest.java:23)

总共有是个线程，执行了9个，一个超时抛出了异常。

上面是按照构造方法自己写的一个测试例子。

当时我用过几次线程池，也在网上看了很多别人写的的测试代码，发现都是用工具类Executors生成的。

Executors这个类提供了很多有用的方法：  
![这里写图片描述][SouthEast 5]

方法分为几类，生成线程池的方法，生成线程工厂的方法以及把Runnable转成Callable的方法。

主要看一下线程池的相关方法，forkjoin与延时执行的就不看了（对这个还不了解），有下面几个：

newFixedThreadPool(int nThreads);
    newFixedThreadPool(int nThreads, ThreadFactory threadFactory);
    newSingleThreadExecutor();
    newSingleThreadExecutor(ThreadFactory threadFactory);
    newCachedThreadPool()
    newCachedThreadPool(ThreadFactory threadFactory);
    unconfigurableExecutorService(ExecutorService executor)

newFixedThreadPool这个方法生成的是固定数量的线程池；  
newSingleThreadExecutor这个方法生成的单例的线程池，不知道什么场景需要使用这种；  
newCachedThreadPool这个方法生成的是自适应大小的线程池，随着任务越来越多，线程数量也会增加；  
unconfigurableExecutorService这个方法生成的给定的实现的线程池（不确定是不是这样的，注释没看看明白）。

所以我在学习Callable的时候，看到很多生成线程池的时候都是这么写的：

ExecutorService executorService = Executors.newCachedThreadPool();

### 总结 ###

虽然对线程池底层执行逻辑还不清楚，但现在对线程池的基本使用清楚了。  
在网上看到几篇对线程池原理讲解的很详细的博文，有兴趣的可以看看。

[海子的博文][Link 1]

[SouthEast]: /images/20220607/f47aa25746ef44f3bc5d6838dc853858.png
[SouthEast 1]: /images/20220607/ebd20c99446a4655ad9fd46e678fae8b.png
[SouthEast 2]: /images/20220607/97afe23dbe864b8291cf04c74a291f80.png
[SouthEast 3]: /images/20220607/95601122fba948cb800c7546ec96ff91.png
[SouthEast 4]: /images/20220607/536edb52bb2e4310bfab17150642de4d.png
[SouthEast 5]: /images/20220607/21a00ffaeb6b400390c31acaf1b71b32.png
[Link 1]: http://www.cnblogs.com/dolphin0520/p/3932921.html