线程池配置参数详解案例

Myth丶恋晨 2022-03-15 10:36 366阅读 0赞

### 目录 ###

*  1.自定义一个线程池
 *   *  1.1 Excutor.execute执行过程
 *  2.newFixedThreadPool
 *  3.newCachedThreadPool
 *  4.设置单任务线程池
 *  keepaliveTime是如何监控回收
 *  LinkedBlockingQueue和ArrayBlockingQueue的异同
 *  SynchronousQueue
 *  『不允许』使用Executors创建线程池
 *   *  Executors为什么存在缺陷
     *   *  如newFixedThreadPool
         *  再看newCachedThreadPool
 *  创建线程池正确方式
 *  线程池大小设置
 *  excute和submit的区别
 *  线程池用的是独占锁
 *  拒绝策略
 *  合理设置线程池大小
 *  线程池容量的动态调整
 *  线程池的监控

*  自定义一个线程池
 *  newFixedThreadPool
 *  newCachedThreadPool
 *  设置单任务线程池

# 1.自定义一个线程池 #

*  API
    
    ThreadPoolExecutor(int corePoolSize,  
    int maximumPoolSize,  
    long keepAliveTime,  
    TimeUnit unit,  
    BlockingQueue workQueue)
    
     *  corePoolSize核心线程数：任务数小于核心线程数，会新建线程处理
     *  maximumPoolSize 最大线程数
     *  keepAliveTime 线程允许的空闲时间（核心线程默认不会超时，使用allowCoreThreadTimeout：允许核心线程超时）
     *  **workQueue 阻塞队列大小(spring参数queueCapacity)，当任务数大于corePoolSize，多余的任务存放在队列等待，当队列满了时，若当前的总线程数小于maximumPoolSize则创建线程，若大于maximumPoolSize，则调用拒绝策略（默认抛异常）**
     *  当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭。如果为false，那么只有当线程超过corePoolSize，并且时间达到keepAliveTime才会销毁

## 1.1 Excutor.execute执行过程 ##

1.  首先判断当前线程数是否小于核心线程数，则创建线程执行
2.  如果当前线程数大于核心线程数，队列没有满，则将当前任务runnable加入到任务队列中。
3.  如果队列也满了，就判断当前线程数是否大于最大线程数，如果不大于，就创建线程执行；如果大于最大线程数，就执行拒绝策略。默认策略是抛出异常。
4.  创建线程里面逻辑里面，会把当前线程数+1，同时会新建一个work线程，将客户端的任务runnable传递进来，然后执行runwork里面方法，核心线程会不断的从队列的取任务，取的方法是take，拿不到任务就阻塞，除非设置allowCoreThreadTimeout。非核心线程，取任务是通过poll方法，里面设置有超时时间，超过时间拿不到就返回null，然后runwork里面跳出while循环销毁线程

*  测试：
    
    比如，这里定义了15个任务，调度时，先有5个任务根据corepoolsize创建，然后放入3个到blocking（这3个任务排队等待执行），剩余7个任务，但是当前最大值时11，所以，因为之前已经创建了5个，这里最多会再创建6个，剩余的会拒绝掉。最终调度的任务只有前面的5+3+6=14个任务

public static void definedExecutor() { 
            // 参数 corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue
            // corePoolSize = 5 maximumPoolSize = 11 workQueue = 3
            Executor executor = new ThreadPoolExecutor(5, 11, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(3));
            executorsTest(executor);
        }
        public static void main(String a[]) { 
            definedExecutor();
        }
        // 模拟15个任务并发执行
        private static void executorsTest(Executor executor) { 
            for (int i = 0; i < 15; i++) { 
                executor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(Thread.currentThread().getName() + "正在执行");
                        try { 
                            Thread.sleep(1000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
        }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70]

# 2.newFixedThreadPool #

*  固定大小线程池，底层实现LinkedBlockingQueue
 *  实现原理最小值和最大值一致，使用**无界**阻塞队列，多余的任务全部都会进入队列排队

public static void fixedThreadPoolTest() { 
            ExecutorService executorService = Executors.newFixedThreadPool(5);
            executorsTest(executorService);
        }
            public static void main(String a[]) { 
            fixedThreadPoolTest();
        }
        // 模拟15个任务并发执行
        private static void executorsTest(Executor executor) { 
            for (int i = 0; i < 15; i++) { 
                executor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(Thread.currentThread().getName() + "正在执行");
                        try { 
                            Thread.sleep(1000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
        }

*  测试结果：可以看到线程数字始终只有5个  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70 1]

# 3.newCachedThreadPool #

*  设置无界线程池
 *  底层实现SynchronousQueue， 使用一个不存元素的阻塞队列，队列的插入操作put，必须另一个线程的调用移除操作take， 这个队列接收到任务的时候，会直接提交给线程处理，而不保留它，所以当此时大于核心线程的任务，maxsize又非常大时会新建线程
 *  SynchronousQueue：生产者消息执行put操作时候，必须要有一条消费者线程执行take，不然生产者现在put操作将会阻塞，直到消费者执行take。take操作将会唤醒该生产线程
 *  条件：maxsize无限大

public static void cacheThreadPoolTest() { 
            Executor executor = Executors.newCachedThreadPool();
            executorsTest(executor);
        }

public static void main(String a[]) { 
            cacheThreadPoolTest();
        }
        // 模拟15个任务并发执行
        private static void executorsTest(Executor executor) { 
            for (int i = 0; i < 15; i++) { 
                executor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(Thread.currentThread().getName() + "正在执行");
                        try { 
                            Thread.sleep(1000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
        }

*  测试  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70 2]

# 4.设置单任务线程池 #

*  等价于创建固定大小为1的线程池Executors.newFixedThreadPool(5);
 *  源码  
    ![在这里插入图片描述][201903020359504.png]

//单线程，始终只有一个
        public static void sigleThreadPool() { 
            Executor executor = Executors.newSingleThreadExecutor();
            executorsTest(executor);
    
        }
        public static void main(String a[]) { 
            sigleThreadPool();
        }
        // 模拟15个任务并发执行
        private static void executorsTest(Executor executor) { 
            for (int i = 0; i < 15; i++) { 
                executor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(Thread.currentThread().getName() + "正在执行");
                        try { 
                            Thread.sleep(1000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
        }

*  测试  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70 3]

# keepaliveTime是如何监控回收 #

参考另一篇[链接][Link 1]

# LinkedBlockingQueue和ArrayBlockingQueue的异同 #

1、队列大小有所不同。ArrayBlockingQueue是有界的初始化必须指定大小，而LinkedBlockingQueue可以是有界的也可以是无界的（Integer.Max\_Value）,对于后者，如果添加速度大于移除速度，在无界情况下，可能会造成内存溢出。  
2、数据存储结构不同，arrayBlockingQueue采用的是数组作为数据存储容器，而LinkedBlockingQueue采用的是链表  
3、由于ArrayBlockingQueue采用的是数组容器，因此在插入和删除元素时不会产生或者销毁任何额外的对象实例，而LinkedBlocking则会产生一个额外的Node对象，这在长时间高并发处理大量数据时候，对于GC可能会产生较大影响。  
4、两者实现队列添加和移除的锁不一样，ArrayBlockingQueue的实现中锁是没有分离的，添加和删除元素用的同一个锁，意味着两者无法真正并行运行。而LinkedBlockingQueue的实现中锁是分离的，其添加采用的是putLock锁，删除采用的是takeLock锁，这样能大大提高队列的吞吐量，生产者和消费者可以并行地操作队列中的数据。

*  ArrayBlockingQueue完全可以采用分离锁，从而实现生产者和消费者操作的完全并行运行。之所以没这样去做，猜测是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧，以至于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上完全占不到任何便宜。

# SynchronousQueue #

它的同步原理是，当生产者线程准备将元素放入这个队列时候，如果这时没有消费者线程过来，生产者就一直wait，一直等到有消费者过来取走元素，SynchronousQueue就可以返回true了。  
同样，当消费者准备从这队列取走元素时候，如果这时没有生产者过来，那么就一直wait，一直等到有生产者过来，消费者就把元素取走，返回true.

# 『不允许』使用Executors创建线程池 #

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

## Executors为什么存在缺陷 ##

### 如newFixedThreadPool ###

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

LinkedBlockingQueue如果不指定大小的情况下，默认是一个链表实现的无界阻塞队列，最大值是Integer.Max\_value，是可以不断的向其中添加任务，任务过多会导致OOM.  
newFixedThreadPool和newSingleThreadExecutor都用了无界的LinkedBlockingQueue

### 再看newCachedThreadPool ###

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

虽然这里用的SynchronousQueue，SynchronousQueue是一种无缓冲的等待队列，不会存任何线程，但是这里默认将最大线程数目设置了 Integer.MAX\_VALUE，那么大量请求下，必然会导致OOM.  
newCachedThreadPool和newScheduledThreadPool都讲最大线程数设置为了Integer.MAX\_VALUE

# 创建线程池正确方式 #

ThreadPoolExecutor threadPoolExecutor =  new ThreadPoolExecutor(0,10,60L,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(10));
     // threadPoolExecutor.allowCoreThreadTimeOut(true);

当设置allowCoreThreadTimeOut(true)时，线程池中corePoolSize线程空闲时间达到keepAliveTime也将关闭。  
allowCoreThreadTimeOut这个方法就像其字面的意思一样，允许Core Thread超时后可以关闭。  
书上说了，要想使线程池没有任务时销毁所有的进程，需要启用allowCoreThreadTimeOut(true)同时将core size设置为0，而实际上，core size设置成任意一个正数值就可以，设置成0时，加不加allowCoreThreadTimeOut(true)都没有影响，因为这个方法是对core thread产生影响，但此时core thread为0，而且当新任务进来时，必须等到workQueue满时才会创建新线程，这也不是我们想要的结果。

# 线程池大小设置 #

取决于硬件环境和软件环境  
CPU的核心数-》  
软件环境：线程的执行情况，  
IO密集型（CPU时间片的切换） CPU核心的2倍  
CPU密集型（CPU利用率非常高） 以CPU核心数为准，最大同时能够执行的线程数，设置最大线程数为CPU核心数+1  
（线程等待的时间+线程CPU的时间）/线程CPU时间\*CPU核心数

sevice.shutdown();  
sevice.shutdownNow();

# excute和submit的区别 #

submit（）；实现一个带返回值的线程。可以接受参数runnable 和callable。出现异常不会抛出，通过future.get抛出

比如：要依赖此线程的执行结果时，可以用submit

excuter(); 只能接受一个runnable接口的类。有异常可以抛出异常

1.  submit：

try { 
                Future<String> future = service.submit(new Callable<String>() { 
                    @Override
                    public String call() throws Exception { 
                        System.out.println("111");
                        System.out.println(1/0);
                        System.out.println("2222");
                        return "返回";
                    }
                });
            }catch (Exception e){ 
                System.out.println("xxxxxxxx");
                e.printStackTrace();
            }

执行结果：未抛出异常  
![在这里插入图片描述][20210317020817696.png]  
获取返回值：

future.get（）//  阻塞获取结果 ，线程run的时候park，run完了会unpark，future才会执行

2.execute

try { 
                service.execute(new Runnable() { 
                    @Override
                    public void run()   { 
                        System.out.println("aaa");
                        System.out.println(1/0);
                        System.out.println("bbb");
    
                    }
                });
            }catch (Exception e){ 
                System.out.println("xxxxxxxx");
                e.printStackTrace();
            }

执行结果：  
![在这里插入图片描述][20210317020945305.png]

# 线程池用的是独占锁 #

# 拒绝策略 #

默认策略AbortPolicy：直接抛出异常  
discardpolicy :任务直接丢弃  
discardOldestPolicy:直接丢弃最早的任务，并且执行当前任务  
callerRunsPolicy :直接由调用者的线程直接处理，例如直接用主线程运行run方法

# 合理设置线程池大小 #

1.  考虑任务是CPU密集型的，还是IO密集型的
    
    cpu密集型任务指进程绝大部份任务依靠cpu的计算能力完成，典型的如同科学计算，数值模拟等程序
    
    io密集型任务指绝大部分任务就是在读入，输出数据，典型的例如web后端程序，主要就是在根据url请求找到对应的资源并输出。mysql的大量读写属于io密集
2.  考虑线程的执行时间，

*  如果是CPU密集型的， CPU密集型可配置线程数为CPU个数+1
 *  IO密集型的任务，IO不占用CPU，所以可以加大线程数量，可以设置为两倍CPU个数+1
 *  如果任务对其他资源依赖，比如远程调用一个服务等，这里面产生一个等待时间。线程数量可以设置为（（等待时间/CPU时间）+1） \* CPU数  
    公式可以看出，等待时间越长，可以设置更多线程。CPU占用时间高，就减少线程

来自并发编程网的题

1.  高并发、执行时间短的任务怎么使用线程池？  
    高并发、执行时间短，说明会产生大量的上下文切换，可以使用CPU+1的线程数
2.  并发不高、执行时间长的怎么使用线程池？  
    这种要分两种情况：如果执行时间长是在IO操作上，IO不占CPU，可以设置大点，两倍CPU+1  
    如果是时间大部分耗在cpu上，则减少线程数，cpu+1
3.  并发高、执行时间长怎么使用线程池？  
    并发高、执行时间长，大量线程执行过慢，会导致服务器压力增加，此时应该考虑如何减少执行时间，或者增加服务器。对于线程池而言，应该根据执行时间，具体对待。参考（2）

# 线程池容量的动态调整 #

ThreadPoolExecutor 提 供 了 动 态 调 整 线 程 池 容 量 大 小 的 方 法 ： setCorePoolSize() 和  
setMaximumPoolSize()，setCorePoolSize：设置核心池大小 setMaximumPoolSize：设置线  
程池最大能创建的线程数目大小

# 线程池的监控 #

如果在项目中大规模的使用了线程池，那么必须要有一套监控体系，来指导当前线程池的状  
态，当出现问题的时候可以快速定位到问题。而线程池提供了相应的扩展方法，我们通过**重  
写**线程池的 beforeExecute、afterExecute 和 shutdown 等方式就可以实现对线程的监控

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70]: /images/20220315/b5c458710d4646f7a02a2bbbe183c2ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70 1]: /images/20220315/bd24a4d035144ad6a5092ac7acda0468.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70 2]: /images/20220315/e80133f2d1094855bcdff9b43e816dc2.png
[201903020359504.png]: /images/20220315/ed8f097573574e789b617d6abfd9ce25.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70 3]: /images/20220315/599cde78d8084e849dd147db2a004432.png
[Link 1]: https://blog.csdn.net/a718515028/article/details/114937755
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2E3MTg1MTUwMjg_size_16_color_FFFFFF_t_70_pic_center]: /images/20220315/04a3090834014e6bab49e8df9dc7d40c.png
[20210317020817696.png]: /images/20220315/1f18fe6e3b70407da3f0881b2023186e.png
[20210317020945305.png]: /images/20220315/be77246ea46844d6b4b9fac4f0c09958.png