异步线程池的使用总结

一.使用场景

       当前主流程中的业务处理愈来愈复杂，须要进行概括总结是否能够异步去作来下降关键接口的耗时和风险，目前咱们在分流调度降级，接单，拒单主流程中将一些镜像信息异步保留下来，固然咱们能够将接单过程的打标等一系列不影响主流程的业务处理通通异步去作。

二.使用风险

     使用过程当中，咱们初步会遇到线程池的参数设置(会影响到资源浪费，任务堆积时是否会引发oom(生产者和消费者的问题)，如何根据生产设置最优参数)，异常监控和处理(如何监控运行异常和蔼后 业务是否能接受抛弃仍是重试 莫名其妙丢失任务等)，模块化设计(统一入口，统一监控)等问题

三.使用过程问题总结

    1.最初的参数设置  

       咱们都知道线程池的参数设置有几个重要的参数，分别是核心线程数 最大线程数 任务队列 线程工厂  线程空闲时间 任务拒绝策略

       First:

                咱们先来看看几个参数的意义：

       (1) 核心线程数：默认状况下核心线程会一直存活，即便处于闲置状态也不会受存keepAliveTime限制即没有任务须要执行。除非将allowCoreThreadTimeOut设置为true，默认是false,活动线程数小于核心线程数时，即便有线程空闲，线程池也会优先建立新线程处理。

       (2)最大线程数：

        当线程数>=corePoolSize  小于最大线程数时，且任务队列已满时，线程池会建立新线程来处理任务 

当线程数=maxPoolSize，且任务队列已满时，线程池会拒绝处理任务而抛出异常

       (3)任务队列：

                最大线程数的大小设置和任务队列的选择以及大小限制有关，经常使用队列 好比SynchronousQueue,LinkedBlockingQueue，咱们来看看线程池的执行规则的比较

                       假设任务队列没有大小限制

                                        1.若是线程数量<=核心线程数量，那么直接启动一个核心线程来执行任务，不会放入队列中。

          2.若是线程数量>核心线程数，但<=最大线程数，而且任务队列是LinkedBlockingQueue的时候，超过核心线程数量的任务会放在任务队列中排队。因此没有大小限制的话会一直添加到任务队列中，因此不会存在启动新线程。也就是说最大线程数的大小设置无心义，可是这种模式 若是生产速度过快 可能会消耗内存太多从而引oom。

                 3.若是线程数量>核心线程数，但<=最大线程数，而且任务队列是SynchronousQueue的时候，线程池会建立新线程执行任务，这些任务也不会被放在任务列中。这些线程属于非核心线程，在任务完成后，闲置时间达到了超时时间就会被清除。

                                        4.若是线程数量>核心线程数，而且>最大线程数，当任务队列是SynchronousQueue的时候，会由于线程池拒绝添加任务而抛出异常。

                               假设任务队列存在大小限制

                                        1.当LinkedBlockingQueue塞满时，新增的任务会直接建立新线程来执行，当建立的线程数量超过最大线程数量时会抛异常。

 2.SynchronousQueue没有数量限制。由于他根本不保持这些任务，而是直接交给线程池去执行。当任务数量超过最大线程数时会直接抛异常。                SynchronousQueue 大小限制无心义。

       (4)线程工厂：线程工厂，提供建立新线程的功能。通常默认。

       (5)线程空闲时间设置：   keepAliveTime,默认状况下核心线程会一直存活，即便处于闲置状态也不会受存keepAliveTime限制即没有任务须要执行。

除非将allowCoreThreadTimeOut设置为true，默认是false。可是为true的时候 线程会退出直到为0，false时 线程会退出到活动线程等于核心线程数。

       (6)任务拒绝策略：

                        首先咱们看下什么状况 新的任务会被拒绝 

                                    1.当线程池线程数量达到最大线程数而且队列已满的时候  

                                    2.当调用shutdown和shutdownnow方法的时候，新进的任务都会被拒绝，前者会执行完剩余任务，后者会试图中止剩余任务 并返回任务列表。

那么这种状况发生的时候，新的任务会被怎样处理呢？线程池提供了哪些策略呢？

                                   1.AbortPolicy 默认 会抛出异常 丢弃任务

                                   2.CallerRunsPolicy 没有shutdown的话 主线程继续执行任务

                                   3.DiscardPolicy 什么都不作 丢弃任务

                                   4.DiscardOldestPolicy 判读shutdown 并移除队列的最后一个任务 同时线程池执行新任务而不是主线程执行新任务。

                      这几种策略都是内部类  实现了同一个接口 RejectedExecutionHandler 若是你愿意的话 也能够本身实现  注意判断是否shutdown的状态就行了。

源码中 提供了两个参数 一个主线程 一个当前线程池 进行处理

public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {      }

         Second:

             了解了这些参数的含义后  咱们如何根据业务定义这些值呢 

              - tasks ：每秒的任务数，按照午高峰假设为500～1000
              - taskcost：每一个任务花费时间，假设为0.01s 
              - responsetime：系统容许容忍的最大响应时间，假设为1s 
              - corePoolSize = 每秒须要多少个线程处理？ 
                 * threadcount = tasks/(1/taskcost) =tasks*taskcout = (500~1000)*0.01 = 5~10 个线程。corePoolSize差很少这个范围 假设初步设为8
              - queueCapacity = (coreSizePool/taskcost)*responsetime 
                * 计算可得 queueCapacity = 8/0.01*1 = 800。意思是队列里的任务能够等待1s，超过了的须要新开线程来执行，否则来不及处理 超过系统容忍的最大响应时间
                * 切记不能设置为Integer.MAX_VALUE，这样队列会很大，线程数只会保持在corePoolSize大小，当任务陡增时，不能新开线程来执行，响应时间会随之陡增。 
              - maxPoolSize = (max(tasks)- queueCapacity)/(1/taskcost)＋corePoolSize
               * 计算可得 maxPoolSize = (1000-800)/100 ＋ 8= 10
               * （最大任务数-队列容量）/每一个线程每秒处理能力 + 核心线程数= 最大线程数 
              - rejectedExecutionHandler：根据具体状况来决定，任务不重要可丢弃，任务重要则要利用一些缓冲机制来处理 
              - keepAliveTime和allowCoreThreadTimeout采用默认一般能知足

               总结下来 初步上线的核心线程数8 最大线程数10 队列容量800

 2.监控后的参数设置

              那么实际线上的状况 可能不是最优  因此咱们须要监控相关指标 在午高峰 晚高峰的时候看看线程池中活跃线程到底有多少 超不超过核心线程数 

              超过以后  任务队列的大小又是多少？因此咱们统一一个入口  关心系统中不一样线程池的执行状况 以下

/**  * Created by huhaosumail on 17/8/25.  */ public class GalaxyThreadPool extends ThreadPoolExecutor{       private String poolName;       public GalaxyThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue,String poolName) {         super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);         this.poolName=poolName;     }       protected void beforeExecute(Thread paramThread, Runnable paramRunnable) {         //获取工做线程     }       protected void afterExecute(Runnable r, Throwable t) {           super.afterExecute(r, t);         //获取工做线程以及运行时异常进行处理 若是线上的话调用etrace 打点会不会访问太多 能够进行随机数来记录         if(System.currentTimeMillis()%100==0){             //获取活动的线程数             MetricUtil.recordIndex("threadPoolMonitor",this.getActiveCount(),poolName+".corePoolSize");             //程池曾经建立过的最大线程数量             MetricUtil.recordIndex("threadPoolMonitor",this.getLargestPoolSize(),poolName+".largestPoolSize");             //线程池的线程数量             MetricUtil.recordIndex("threadPoolMonitor",this.getPoolSize(),poolName+".poolSize");             //任务队列的长度             MetricUtil.recordIndex("threadPoolMonitor",this.getQueue().size(),poolName+".queueSize");         }     }       protected void terminated() {         //线程池关闭以前能够干一些事情；       }     //对于执行结果的获取     @Override     public Future<?> submit(Runnable task) {         return super.submit(task);     } }

咱们封装了本身的线程池  并实现了执行完任务的方法  在方法内 咱们能获取到相关线程池的指标 如 活跃线程数  任务队列大小等  结合线程池的执行策略 进行调优参数 

好比 活跃线程数始终不超过5

       

队列长度基本没有

就是说消费速度很快  核心线程数能够适当调小  基本够用

地址：

https://d.elenet.me/dashboard/db/galaxy-thread-pool-monitor?panelId=1&fullscreen&edit&editorTab=General

3.接下来咱们对线程池的执行结果进行处理 咱们知道提交任务后 咱们能够观察执行任务的结果 以下

List<Future> taskList ＝  new　ArrayList<Future>(); for () {     taskList.add(ruleMirrorImageThreadPool.submit(new RuleMirrorImageRunnable(order.getTrackingId(),currentRule.getChainContext().getScheduleUnitList(), order, currentSeqNum, currentRule.getId())); }   if (Assert.isEmptyList(taskList)) {     continue; }   for (Future task : taskList) {     try {         task.get();     } catch (Exception e) {        //是否从新处理  或者持久化失败信息 或者抛出异常等     } }

综上 咱们对于线程池的最初参数设置  参数调优  统一监控指标  线程池异常处理等有了初步的了解。

遗留问题：1.如何在代码层面监测到当前是否引发了oom或者线程池异常过多  而后当即关闭线程池 减少系统的风险性？

                  2.如何防止丢失任务？