Java线程池进阶

朴灿烈づ我的快乐病毒、 2021-09-20 20:06 283阅读 0赞

***在之前的文章[Java面试知识点（七十三）线程池 ][Java_]，已经说了线程池的基本情况，包括线程池的运行原理，线程池的创建，任务的提交，获取结果，线程池的关闭和配置，下面我们在深入的了解一下线程池并进行代码编写。***

### 一、线程池的继承架构 ###

Java 里面线程池的顶级接口是 Executor，但是严格意义上讲 Executor 并不是一个线程池，而只是一个执行线程的工具。

真正的线程池接口是 ExecutorService。下面这张图完整描述了线程池的类体系结构。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70]

*  Executor 是一个顶层接口（类似一个标记接口），在它里面只声明了一个方法 execute (Runnable)，返回值为 void，参数为 Runnable 类型，从字面意思可以理解，就是用来执行传进去的任务的；
 *  然后 ExecutorService 接口继承了 Executor 接口，并声明了一些方法：submit、invokeAll、invokeAny 以及 shutDown 等；
 *  抽象类 AbstractExecutorService 实现了 ExecutorService 接口，基本实现了 ExecutorService 中声明的所有方法；
 *  ThreadPoolExecutor 继承了类 AbstractExecutorService。

**整理得：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70 1]

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

### 二、线程池相关类 ###

**1.Executors 类**

该类里面提供了一些**静态工厂**，生成一些常用的线程池。

*  newSingleThreadExecutor：创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行，**使用的阻塞队列是LinkedBlockingQueue；**
 *  newFixedThreadPool：创建**固定大小的线程池**。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程，**使用的 LinkedBlockingQueue；**
 *  newCachedThreadPool：创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，那么就会回收部分空闲（60 秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。 此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说 JVM）能够创建的最大线程大小，newCachedThreadPool 将 corePoolSize 设置为 0，将 maximumPoolSize 设置为 Integer.MAX\_VALUE，**使用的 SynchronousQueue**，也就是说来了任务就创建线程运行，当线程空闲超过 60 秒，就销毁线程。
 *  newScheduledThreadPool：创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。
 *  newSingleThreadExecutor：创建一个单线程的线程池。此线程池支持定时以及周期性执行任务的需求。
 *  【源码】

public static ExecutorService newFixedThreadPool(int nThreads) { 
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    public static ExecutorService newSingleThreadExecutor() { 
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    public static ExecutorService newCachedThreadPool() { 
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

**注意**：声明LinkedBlockingQueue的时候，可以指定大小，也可以不指定，不指定的时候，就是默认Integer.MAX\_VALUE的大小，所以当阻塞队列是linked并且不指定大小的时候，提交任务是否溢出是根据内存来确定的。而arrayblockingqueue必须指定大小

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

**2.Future类**

*  Future 表示异步计算的结果。  
    它提供了检查计算是否完成的方法，以等待计算的完成，并获取计算的结果。  
    计算完成后只能使用 get 方法来获取结果，如有必要，计算完成前可以阻塞此方法。
 *  取消则由 cancel 方法来执行。还提供了其他方法，以确定任务是正常完成还是被取消了。一旦计算完成，就不能再取消计算。
 *  如果为了可取消性而使用 Future 但又不提供可用的结果，则可以声明 Future<?> 形式类型、并返回 null 作为底层任务的结果。  
    ![在这里插入图片描述][20191017101555759.png]
 *  Future 就是对于具体的 Runnable 或者 Callable 任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过 get 方法获取执行结果，该方法会阻塞直到任务返回结果。
 *  也就是说 Future 提供了三种功能：
    
     *  判断任务是否完成；
     *  能够中断任务；
     *  能够获取任务执行结果。
 *  Future 类的方法
    
     *  boolean cancel (boolean mayInterruptIfRunning) 试图取消对此任务的执行。
     *  V get () 如有必要，等待计算完成，然后获取其结果。
     *  V get (long timeout, TimeUnit unit) 如有必要，最多等待为使计算完成所给定的时间之后，获取其结果（如果结果可用）。
     *  boolean isCancelled () 如果在任务正常完成前将其取消，则返回 true。
     *  boolean isDone () 如果任务已完成，则返回 true。

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

### 三、代码示例 ###

**1.无返回值的Runable示例**

【实现runnable接口的线程类】

package test.threadpool;
    
    public class Water implements Runnable { 
    
        private int num;
    
        public Water(int num) { 
            this.num = num;
        }
    
        @Override
        public void run() { 
            System.out.println("第" + num + "号选手入场");
            try { 
                Thread.currentThread().sleep(3000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            System.out.println("第" + num + "号选手黯然离场");
        }
    }

【生产环境】

package test.threadpool;
    
    import java.util.concurrent.*;
    
    public class Produce { 
        public static void main(String[] args) { 
            // 核心线程数
            int corePoolSize =5;
            // 线程池总大小
            int poolSize = 10;
            // 任务阻塞队列
            BlockingQueue<Runnable> queue = new ArrayBlockingQueue(5);
            // 创建线程池
            ThreadPoolExecutor executor = new ThreadPoolExecutor(
                    corePoolSize,
                    poolSize,
                    2000,
                    TimeUnit.MICROSECONDS,
                    queue);
    
    		// 注意这里的循环次数，当超过池和阻塞队列的和之后，线程池会拒绝，使用默认的拒绝策略，即抛出异常
            for (int i=1; i<16; i++) { 
                Water task = new Water(i);
                executor.execute(task);
                System.out.println("线程池中当前线程数："+executor.getPoolSize()+
                        ",等待执行的任务数："+executor.getQueue().size()+
                        ",已经完成人数数："+executor.getCompletedTaskCount());
            }
    
            // 关闭线程池
            executor.shutdown();
    
        }
    
    }

【运行结果（不唯一）】

线程池中线程数：1,等待执行的任务数：0,已经完成人数数：0
    第1号选手入场
    线程池中线程数：2,等待执行的任务数：0,已经完成人数数：0
    第2号选手入场
    线程池中线程数：3,等待执行的任务数：0,已经完成人数数：0
    第3号选手入场
    线程池中线程数：4,等待执行的任务数：0,已经完成人数数：0
    线程池中线程数：5,等待执行的任务数：0,已经完成人数数：0
    第4号选手入场
    线程池中线程数：5,等待执行的任务数：1,已经完成人数数：0
    线程池中线程数：5,等待执行的任务数：2,已经完成人数数：0
    第5号选手入场
    线程池中线程数：5,等待执行的任务数：3,已经完成人数数：0
    线程池中线程数：5,等待执行的任务数：4,已经完成人数数：0
    线程池中线程数：5,等待执行的任务数：5,已经完成人数数：0
    线程池中线程数：6,等待执行的任务数：5,已经完成人数数：0
    线程池中线程数：7,等待执行的任务数：5,已经完成人数数：0
    线程池中线程数：8,等待执行的任务数：5,已经完成人数数：0
    第11号选手入场
    第12号选手入场
    线程池中线程数：9,等待执行的任务数：5,已经完成人数数：0
    线程池中线程数：10,等待执行的任务数：5,已经完成人数数：0
    第13号选手入场
    第14号选手入场
    第15号选手入场
    第3号选手黯然离场
    第14号选手黯然离场
    第15号选手黯然离场
    第1号选手黯然离场
    第12号选手黯然离场
    第5号选手黯然离场
    第2号选手黯然离场
    第11号选手黯然离场
    第13号选手黯然离场
    第4号选手黯然离场
    第10号选手入场
    第9号选手入场
    第8号选手入场
    第7号选手入场
    第6号选手入场
    第9号选手黯然离场
    第7号选手黯然离场
    第10号选手黯然离场
    第8号选手黯然离场
    第6号选手黯然离场

从执行结果可以看出，当线程池中线程的数目大于 5 时，便将任务放入任务缓存队列里面，当任务缓存队列满了之后，便创建新的线程。  
如果上面程序中，将 for 循环中改成执行 20 个任务，就会抛出任务拒绝异常了。

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

**2.有返回值的Callable示例**

callable也是实现多线程的一种方式，但是单独用的情况很少，大部分情况是和线程池一起使用

Runnable 和 Callable 的区别

*  Runnable 执行方法是 run (),Callable 是 call ()
 *  实现 Runnable 接口的任务线程无返回值；实现 Callable 接口的任务线程能返回执行结果
 *  call 方法可以抛出异常，run 方法若有异常只能在内部消化

**【实现callable接口】**

package test.threadpool;
    
    import java.util.concurrent.Callable;
    
    public class CallabelImpl implements Callable { 
        private int num;
    
        public CallabelImpl(int num) { 
            this.num = num;
        }
    
        @Override
        public Integer call() throws Exception { 
            int result = 0;
            for (int i=0; i<=num; i++) { 
                result += i;
                System.out.println("这是："+num+" 的计算进度："+result);
            }
            System.out.println("<<<<<<<<计算结束>>>>>>>");
            return result;
        }
    }

**【生产环境】**

package test.threadpool;
    
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    public class Demo { 
        public static void main(String[] args) { 
            ExecutorService pool = Executors.newFixedThreadPool(1);
            Future<Integer> f1 = pool.submit(new CallabelImpl(5));
            Future<Integer> f2 = pool.submit(new CallabelImpl(10));
    
            try { 
                Integer i1 = f1.get();
                Integer i2 = f2.get();
    
                System.out.println(i1+"----"+i2);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            } catch (ExecutionException e) { 
                e.printStackTrace();
            } finally { 
    
            }
            pool.shutdown();
        }
    }

**【运行结果（线程池大小为1，结果唯一）】**

这是：5 的计算进度：0
    这是：5 的计算进度：1
    这是：5 的计算进度：3
    这是：5 的计算进度：6
    这是：5 的计算进度：10
    这是：5 的计算进度：15
    <<<<<<<<计算结束>>>>>>>
    这是：10 的计算进度：0
    这是：10 的计算进度：1
    这是：10 的计算进度：3
    这是：10 的计算进度：6
    这是：10 的计算进度：10
    这是：10 的计算进度：15
    这是：10 的计算进度：21
    这是：10 的计算进度：28
    这是：10 的计算进度：36
    这是：10 的计算进度：45
    这是：10 的计算进度：55
    <<<<<<<<计算结束>>>>>>>
    15----55

但是如果，把线程池的大小设置成大于等于2

ExecutorService pool = Executors.newFixedThreadPool(2);

**【结果（不唯一）】**

这是：10 的计算进度：0
    这是：5 的计算进度：0
    这是：10 的计算进度：1
    这是：10 的计算进度：3
    这是：10 的计算进度：6
    这是：10 的计算进度：10
    这是：5 的计算进度：1
    这是：10 的计算进度：15
    这是：10 的计算进度：21
    这是：10 的计算进度：28
    这是：10 的计算进度：36
    这是：10 的计算进度：45
    这是：10 的计算进度：55
    <<<<<<<<计算结束>>>>>>>
    这是：5 的计算进度：3
    这是：5 的计算进度：6
    这是：5 的计算进度：10
    这是：5 的计算进度：15
    <<<<<<<<计算结束>>>>>>>
    15----55

分析：当线程池只有一个的时候，提交两个任务，一个在corepool中一个在阻塞队列中，在阻塞队列中的任务会等待池中的任务运行结束，在进入池中开始执行。

[Java_]: https://blog.csdn.net/qq_33945246/article/details/102551704
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70]: /images/20210920/407d4a2255c649f0b776cd1fa9bf4e00.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70 1]: /images/20210920/67dfc61440d542daaba01ff05e0c6203.png
[20191017101555759.png]: /images/20210920/11581095ebaf4e0198bea5fd43dbfa9f.png