Java并发——ThreadPoolExecutor详解（一）

阳光穿透心脏的1/2处 2021-08-30 18:32 443阅读 1赞

### ThreadPoolExecutor ###

类ThreadPoolExecutor可以非常方便地创建线程池对象，而不需要程序员设计大量的new实例化Thread相关的代码。

构造：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue)

*  `corePoolSize`:池中所保存的线程数，包括空闲线程，也就是核心线程池的大小
 *  `maximumPoolSize`:池中允许的最大线程数。
 *  `keepAliveTime`:当线程数量大于corePoolSize值时，在没有超过指定的时间内是不从线程池中将空闲线程删除的，如果超过此时间单位，则删除。
 *  `unit`: keepAliveTime 参数的时间单位。
 *  `workQueue` :执行前用于保持任务的队列。此队列仅保持由execute方法提交的Runnable任务。

为了更好地理解这些参数在使用上的- - 些关系，可以将它们进行详细化的注释:

1.  A代表execute(runnable)欲执行的runnable的数量;
2.  B代表corePoolSize;
3.  C代表maximumPoolSize;
4.  D代表A-B(假设A>=B);
5.  E代表new LinkedBlockingDeque0;队列，无构造参数;
6.  F代表SynchronousQueue队列;
7.  G代表keepAliveTime。

构造方法中5个参数之间都有使用上的关系，在使用线程池的过程中大部分会出现如下5种过程:

*  A`如果A<=B`,那么马上创建线程运行这个任务，并不放人扩展队列Queue中，其他参数功能忽略;
 *  B`如果A>B&&A<=C&&E`，则C和G参数忽略,并把D放入E中等待被执行;
 *  C`如果A>B&&A<=C&&F`,则C和G参数有效，并且马上创建线程运行这些任务，而不把D放人F中，D执行完任务后在指定时间后发生超时时将D进行清除;，
 *  D`如果A>B&&A>C&&E`,则C和G参数忽略，并把D放人E中等待被执行;
 *  E`如果A>B&&A>C&&F`,则处理C的任务，其他任务则不再处理抛出异常。

1. 前两个参数与getCorePoolSize()和getMaximumPoolSize()方法：

public class Run1 { 
        //获取基本属性corePoolSize和maximumPoolSize
        public static void main(String[] args) { 
            ThreadPoolExecutor executor = new ThreadPoolExecutor(7, 8,
                    5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            System.out.println(executor.getCorePoolSize());//7
            System.out.println(executor.getMaximumPoolSize());//8
            System.out.println("");
            executor = new ThreadPoolExecutor(7, 8, 5,
                    TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
            System.out.println(executor.getCorePoolSize());//7
            System.out.println(executor.getMaximumPoolSize());//8
        }
    }

2. 在线程池中添加的线程数量<=corePoolSize:

public class Run2_1 { 
        //队列使用LinkedBlockingDeque类
        //并且线程数量<=corePoolSize，所以keepAliveTime>5时也不清楚空闲线程
        public static void main(String[] args) throws InterruptedException { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        System.out.println(Thread.currentThread().getName() + "run!" + System.currentTimeMillis());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
            ThreadPoolExecutor executor = new ThreadPoolExecutor(7, 8, 5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            executor.execute(runnable);//1
            executor.execute(runnable);//2
            executor.execute(runnable);//3
            executor.execute(runnable);//4
            executor.execute(runnable);//5
            executor.execute(runnable);//6
            executor.execute(runnable);//7
            Thread.sleep(300);
            System.out.println("A:" + executor.getCorePoolSize());
            System.out.println("A:" + executor.getPoolSize());
            System.out.println("A:" + executor.getQueue().size());
    
            Thread.sleep(10000);
            System.out.println("B:" + executor.getCorePoolSize());
            System.out.println("B:" + executor.getPoolSize());
            System.out.println("B:" + executor.getQueue().size());
        }
    }

因为提交任务个数小于核心线程数，所以立即执行，并不放入队列；任务书<=核心线程数，所以核心池中的线程超过5秒也不清除；  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]

3. 数量>corePoolSize并且<=maximumPoolSize的情况：

public class Run3 { 
        //队列使用LinkedBlockingDeque类，也就是如果线程数量>corePoolSize时，将多余的任务放入队列
        //同一时间最多只有7个线程在运行
        //如果使用LinkedBlockingDeque类则maximumPoolSize参数作用将忽略
        public static void main(String[] args) throws InterruptedException { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        System.out.println(Thread.currentThread().getName() + "run!" + System.currentTimeMillis());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
            ThreadPoolExecutor executor = new ThreadPoolExecutor(7, 8, 5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            executor.execute(runnable);//1
            executor.execute(runnable);//2
            executor.execute(runnable);//3
            executor.execute(runnable);//4
            executor.execute(runnable);//5
            executor.execute(runnable);//6
            executor.execute(runnable);//7
            executor.execute(runnable);//8
            Thread.sleep(300);
            System.out.println("A:" + executor.getCorePoolSize());
            System.out.println("A:" + executor.getPoolSize());
            System.out.println("A:" + executor.getQueue().size());
    
            Thread.sleep(10000);
            System.out.println("B:" + executor.getCorePoolSize());
            System.out.println("B:" + executor.getPoolSize());
            System.out.println("B:" + executor.getQueue().size());
        }
    }

超过核心线程池的任务被放入队列中；  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]

//队列使用SynchronousQueue类
    //并且线程数量>corePoolSize时
    //将其余的任务也放入池中，总数量为8，
    //并且线程总数量也没有超过maximumPoolSize值的8
    //由于运行的线程数为8，数量上>corePoolSize为7的值
    //所以keepAliveTime>5时清除空闲线程
    //如果使用SynchronousQueue类则maximumPoolSize参数的作用将有效

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]

3. 数量>maximumPoolSize：

public class Run3 { 
        //队列使用LinkedBlockingDeque类，也就是如果线程数量>corePoolSize时，将多余的任务放入队列
        //同一时间最多只有corePoolSize个线程在运行
        //所以keepAliveTime>5时也不清除空闲线程
        public static void main(String[] args) throws InterruptedException { 
            Runnable runnable = new Runnable() { 
                @Override
                public void run() { 
                    try { 
                        System.out.println(Thread.currentThread().getName() + "run!" + System.currentTimeMillis());
                        Thread.sleep(1000);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            };
            ThreadPoolExecutor executor = new ThreadPoolExecutor(7, 8,
                    5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            executor.execute(runnable);//1
            executor.execute(runnable);//2
            executor.execute(runnable);//3
            executor.execute(runnable);//4
            executor.execute(runnable);//5
            executor.execute(runnable);//6
            executor.execute(runnable);//7
            executor.execute(runnable);//8
            executor.execute(runnable);//9
            Thread.sleep(300);
            System.out.println("A:" + executor.getCorePoolSize());
            System.out.println("A:" + executor.getPoolSize());
            System.out.println("A:" + executor.getQueue().size());
    
            Thread.sleep(10000);
            System.out.println("B:" + executor.getCorePoolSize());
            System.out.println("B:" + executor.getPoolSize());
            System.out.println("B:" + executor.getQueue().size());
        }
    }

结论：`如果A>B&&A>C&&E，则C和G参数忽略，并把D放人E中等待被执行`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]

如果使用SynchronousQueue：  
结论：`如果A>B&&A>C&&F,则处理C的任务，其他任务则不再处理抛出异常`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]

shutdown()和shutdownNow():

方法shutdown()的作用是使当前未执行完的线程继续执行，而不再添加新的任务Task，还有shutdown()方法不会阻塞，调用shutdown()方法后，主线程main就马上结束了，而线程池会继续运行直到所有任务执行完才会停止。如果不调用shutdown()方法，那么线程池会一直保持 下去，以便随时执行被添加的新Task任务。

public class MyRunnable implements Runnable { 
        @Override
        public void run() { 
            try { 
                System.out.println("begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
                Thread.sleep(4000);
                System.out.println("end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) { 
            MyRunnable myRunnable = new MyRunnable();
            ThreadPoolExecutor pool = new ThreadPoolExecutor(7, 10, 0L,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            pool.execute(myRunnable);
            pool.shutdown();
            System.out.println("main end");
        }
    }

4秒之后进程结束；  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]

public class MyRunnable implements Runnable { 
        @Override
        public void run() { 
            try { 
                System.out.println("begin " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
                Thread.sleep(4000);
                System.out.println("end " + Thread.currentThread().getName() + " " + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            MyRunnable myRunnable = new MyRunnable();
            ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 99999, 9999L,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            pool.execute(myRunnable);
            pool.execute(myRunnable);
            pool.execute(myRunnable);
            pool.execute(myRunnable);
            Thread.sleep(1000);
            pool.shutdown();
            pool.execute(myRunnable);//调用了shutdwon后不可以再提交任务
            System.out.println("main end");
        }
    }

通过运行结果可知，执行了4个任务，最后一个任务报异常，因为执行了shutdown)方法，并且当前程序进程销毁。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]

方法shutdownNow()的作用是中断所有的任务Task,并且抛出InterruptedException异常，前提是在Runnable中使用`if (Thread.currentThreadQ.isInterrupted) == true)` 语句来判断当前线程的中断状态，而未执行的线程不再执行，也就是从执行队列中清除。如果没有`if(Thread.currentThread().isInterrupted) == true)`语句及抛出异常的代码，则池中正在运行的线程直到执行完毕，而未执行的线程不再执行，也从执行队列中清除。

public class MyRunnable1 implements Runnable { 
        @Override
        public void run() { 
            try { 
                for (int i = 0; i < Integer.MAX_VALUE / 50; i++) { 
                    String newString = new String();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    Math.random();
                    if (Thread.currentThread().isInterrupted() == true) { 
                        System.out.println("任务没有完成，就中断了！");
                        throw  new InterruptedException();
                    }
                    System.out.println("任务成功完成！");
                }
            } catch (InterruptedException e) { 
                System.out.println("进入catch中断了任务");
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            MyRunnable1 myRunnable1 = new MyRunnable1();
            ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 99999, 9999L,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            pool.execute(myRunnable1);
            pool.execute(myRunnable1);
            pool.execute(myRunnable1);
            pool.execute(myRunnable1);
            Thread.sleep(1000);
            pool.shutdownNow();
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]

public class MyRunnable1 implements Runnable { 
        @Override
        public void run() { 
            for (int i = 0; i < Integer.MAX_VALUE / 1000; i++) { 
                String newString = new String();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
            }
            System.out.println("任务成功完成！");
        }
    
        public static void main(String[] args) throws InterruptedException { 
            MyRunnable1 myRunnable1 = new MyRunnable1();
            ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 99999, 9999L,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            pool.execute(myRunnable1);
            pool.execute(myRunnable1);
            pool.execute(myRunnable1);
            pool.execute(myRunnable1);
            Thread.sleep(1000);
            List<Runnable> runnables = pool.shutdownNow();//返回取消的任务
            System.out.println("main end");
            for (Runnable r : runnables)
                System.out.println(r.toString());
        }
    }

控制台信息代表2个任务被成功执行，其余2个任务被取消运行，并且进程销毁。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]

*  当线程池调用`shutdown()`方法时，线程池的状态则立刻变成`SHUTDOWN`状态，此时不能再往线程池中添加任何任务，否则将会抛出`RejectedExecutionException`异常。但是，此时线程池不会立刻退出，直到线程池中的任务都已经处理完成，才会退出。
 *  而`shutdownNow()`方法是使线程池的状态立刻变成`STOP`状态，并试图停止所有正在执行的线程(`如果有if 判断则人为地抛出异常`),不再处理还在池队列中等待的任务，当然，它会返回那些未执行的任务。

isShutdown()：  
判断线程池是否已经关闭。

isTerminating()和isTerminated()：

如果正在执行的程序处于shutdown或shutdownNow之后处于正在终止但尚未完全终止的过程中，调用方法isTerminating()则返回true。此方法可以比喻成，门是否正在关闭。门彻底关闭时，线程池也就关闭了。

如果线程池关闭后，也就是所有任务都已完成，则方法isTerminated()返回true。此方法可以比喻成，门是否已经关闭。

awaitTermination(long timeout, TimeUnit unit):  
方法awaitTermination(long timeout,TimeUnit unit)的作用就是查看在指定的时间之间，线程池是否已经终止工作，也就是最多等待多少时间后去判断线程池是否已经终止工作。此方法需要有shutdown()方法的配合。

`注意：`该方法具有阻塞特性。awaitTermination()被执行时，如果池中`有任务`在被执行时，则调用方法出现`阻塞`，等待指定的时间，如果`没有任务时则不再阻塞`。

public class MyRunnable2 implements Runnable { 
        @Override
        public void run() { 
            try { 
                System.out.println(Thread.currentThread().getName() + " " + System.currentTimeMillis());
                Thread.sleep(4000);
                System.out.println(Thread.currentThread().getName() + " " + System.currentTimeMillis());
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            MyRunnable2 myRunnable2 = new MyRunnable2();
            ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 99999, 9999L,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            pool.execute(myRunnable2);
            System.out.println("main begin! " + System.currentTimeMillis());
            System.out.println(pool.awaitTermination(10, TimeUnit.SECONDS));
            System.out.println("main end! " + System.currentTimeMillis());
        }
    }

打印false是因为线程池未调动shutdown()方法。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 9]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 10]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 11]  
从打印的结果来看，“main end !”打印的时间就是线程执行完毕后的时间，也就是说方法awaitTermination()被执行时，如果池中有任务在被执行时，则调用awaitTermination()方法出现阻塞，等待指定的时间，如果没有任务时则不再阻塞。

方法awaitTermination()与shutdown()结合时可以实现“等待执行完毕”的效果，原理就是应用awaitTermination()方法具有阻塞性，如果awaitTermination()方法正在阻塞的过程中任务执行完毕，则awaitTermination()取消阻塞继续执行后面的代码:

public static void main(String[] args) throws InterruptedException { 
            MyRunnable2 myRunnable2 = new MyRunnable2();
            ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 99999, 9999L,
                    TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
            pool.execute(myRunnable2);
            pool.execute(myRunnable2);
            pool.execute(myRunnable2);
            pool.execute(myRunnable2);
            pool.shutdown();
            System.out.println(pool.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS) +
                    " " + System.currentTimeMillis() + " 全部任务执行完毕！");
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 12]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70]: /images/20210813/9754a34f2fa84c0c913fe0feda118e3c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 1]: /images/20210813/56dff544df354286a41c9bf7f7e88f6f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 2]: /images/20210813/6fb69fc13d054338a0e3469219b7dadc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 3]: /images/20210813/779152656dae4a1a94a27e34507ddae2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 4]: /images/20210813/195687097cbc4ddfa4f0afb25e9eaacb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 5]: /images/20210813/278e7e9bfacd4c30a1d5d4bb6895761a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 6]: /images/20210813/123b59b25a4f47be8f68a61c42d21e06.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 7]: /images/20210813/7f0a79488aa0412d9a9721c6b763d2dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 8]: /images/20210813/3f5350a5b3004649b14e49f83e2a0059.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 9]: /images/20210813/48b1ab15ee7d422bbf7c6278f70b4cf6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 10]: /images/20210813/5f445a9d04934236a3138673767ee443.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 11]: /images/20210813/377b3911dffd4fd1a744b80175188ba3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NvbGRfX19wbGF5_size_16_color_FFFFFF_t_70 12]: /images/20210813/cace7d44e4e34243b1f0490a44cfc2dc.png