线程池总结
线程池0.线程池0.1 线程池原理0.2 线程池参数0.3 线程池队列1.线程池总结2.阻塞队列和非阻塞队列2.线程池原理3.线程池的状态和状态之间的切换4.线程池有哪些工作队列5.项目中怎么创建线程池6.如何合理设置线程池参数7.线程池模拟分段抓取Mysql数据0.线程池0.1 线程池原理## 线程池工作流程 提交任务 ↓ 核心线程数 corePoolSize ├──是→ 创建核心线程执行 └──否→ 任务队列是否已满 ├──否→ 放入队列排队 └──是→ 总线程数 maxPoolSize ├──是→ 创建非核心线程执行 └──否→ 执行拒绝策略 丢弃/抛异常/丢弃最老/调用者自己处理0.2 线程池参数参数说明corePoolSize常驻核心线程maxPoolSize总线程数 核心线程数 临时非核心线程workQueue存来不及执行的任务常用ArrayBlockingQueue、LinkedBlockingQueuekeepAliveTime非核心线程存活时间空闲的非核心线程超过这个时间就被回收RejectedExecutionHandler线程满 队列满 - 新的任务无法处理时的拒绝策略抛异常/丢弃/丢弃最老/调用者自己处理0.3 线程池队列阻塞队列会 “等待” 的线程安全队列满了等空位空了等数据1.队列满入队阻塞 队列已经放满了再尝试添加元素 - 线程卡住等待直到队列有空位才继续。 2.队列空出队阻塞 队列里面没有元素再尝试取出元素 - 线程卡住等待直到有新元素入队才继续。 把它当成排队取餐窗口 * 窗口最多同时放 5 份餐队列容量 5 * 厨师生产者 不停做餐放窗口窗口满了厨师就停下等待有人取餐再继续做 * 顾客消费者 排队取餐窗口没餐了顾客就站着等有新餐再取 这就是典型的生产者 - 消费者模型阻塞队列天生就是为解决这个场景设计的队列对比ArrayBlockingQueueLinkedBlockingQueue底层结构数组链表容量特性必须指定容量有界队列不指定容量就是无界指定就是有界锁机制入队/出队共用一把锁入队出队各一把锁并发性能较低锁竞争大高入队出队互不干扰GC影响小大频繁创建节点不指定容量高并发容易OOM非核心线程创建任务满会创建非核心线程不指定容量永远不会创建非核心线程因为队列无限大安全性安全可控不炸内存不指定容量高并发疯狂堆积任务导致OOM生产环境推荐用ArrayBlockingQueue必须用指定容量的 LinkedBlockingQueue1.线程池总结线程池的好处线程复用控制最大并发数管理线程2.阻塞队列和非阻塞队列非阻塞队列 和 阻塞队列 的入队offer 和出队poll// 非阻塞队列publicinterfaceQueueEextendsCollectionE{booleanoffer(Ee);Epoll();}// 阻塞队列publicinterfaceBlockingQueueEextendsQueueE{booleanoffer(Ee,longtimeout,TimeUnitunit)throwsInterruptedException;Epoll(longtimeout,TimeUnitunit)throwsInterruptedException;}入队非阻塞队列队列满添加元素立马返回失败阻塞队列添加元素如果队列满可以等待一段时间如果在等待的过程中队列有空闲位置就添加成功出队非阻塞队列队列为空poll出队立马返回null阻塞队列poll出队如果队列为空可以等待一段时间等待期间出现新元素那么就出队成功。//非阻塞队列ConcurrentLinkedQueueStringclqnewConcurrentLinkedQueue();//入队clq.add(a);clq.add(b);clq.add(c);//查看队列头Stringpeekclq.peek();System.out.println(peek peek);//aSystem.out.println(clq.size() clq.size());//3//出队Stringpollclq.poll();System.out.println(poll poll);//aSystem.out.println(clq.size() clq.size());//2System.out.println();//阻塞队列BlockingQueueStringbqnewLinkedBlockingDeque(2);// bq.add(a);//bq.add(b);// bq.add(c);//java.lang.IllegalStateException: Deque full//todo:offer 和 add一样都是入队但是offer可以等待一定时间看能否可以插入bq.offer(a);bq.offer(b);System.out.println(bq.poll() bq.poll());//阻塞3秒,看能不能把c放进队列中bq.offer(c,3,TimeUnit.SECONDS);System.out.println(bq.poll() bq.poll());System.out.println(bq.poll() bq.poll());//阻塞5秒,等待是否有可被取出的元素System.out.println(bq.poll() bq.poll(5,TimeUnit.SECONDS));2.线程池原理线程池ThreadPoolExecutor七大构造参数参数说明corePollSize核心线程数量线程池维护线程的最少数量maximumPoolSize线程池维护线程的最大数量keepAliveTime线程池除核心线程外的其他线程的最长空闲时间超过该时间的空闲线程会被销毁unitkeepAliveTime的单位TimeUnit中的几个静态属性NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDSworkQueue线程池所使用的任务缓冲队列存储待执行的任务threadFactory用于创建线程是生成线程池中工作线程的工厂handler拒绝策略任务阻塞队列满且工作线程数达到maximumPoolSize触发拒绝策略publicThreadPoolExecutor(intcorePoolSize,intmaximumPoolSize,longkeepAliveTime,TimeUnitunit,BlockingQueueRunnableworkQueue,ThreadFactorythreadFactory,RejectedExecutionHandlerhandler)ThreadPoolExecutor提供了四种拒绝策略饱和策略说明ThreadPoolExecutor.AbortPolicy丢弃任务并抛出RejectedExecutionException异常默认策略ThreadPoolExecutor.DiscardPolicy丢弃任务但是不抛出异常ThreadPoolExecutor.DiscardOldestPolicy丢弃队列最前面的任务然后重新尝试执行任务重复此过程ThreadPoolExecutor.CallerRunsPolicy由调用线程处理该任务自定义可以通过实现RejectedExecutionHandler接口自定义处理方式线程池工作原理3.线程池的状态和状态之间的切换线程池状态RUNNING接收新任务处理排队任务SHUTDOWN不接收新任务但是处理排队任务STOP不接收新任务也不处理排队任务并且中断正在进行的任务TIDYING所有任务都已经终止线程池中工作线程为0线程会转换为TIDYING状态即将运行钩子方法 terminated()TERMINATEDterminated()方法执行完成时进入TERMINATED状态RUNNING:Acceptnewtasks and process queued tasksSHUTDOWN:Dont acceptnewtasks,but process queued tasksSTOP:Dont acceptnewtasks,dont process queued tasks,and interrupt in-progress tasksTIDYING:Alltasks have terminated,workerCount is zero,the thread transitioningtostateTIDYINGwill run theterminated()hook methodTERMINATED:terminated()has completed线程池状态转换RUNNING-SHUTDOWNOninvocation ofshutdown(),perhaps implicitly infinalize()(RUNNINGorSHUTDOWN)-STOPOninvocation ofshutdownNow()SHUTDOWN-TIDYINGWhenboth queue and pool are emptySTOP-TIDYINGWhenpool is emptyTIDYING-TERMINATEDWhentheterminated()hook method has completed4.线程池有哪些工作队列ArrayBlockingQueue基于数组结构的有界阻塞队列FIFO。LinkedBlockingQueue基于链表结构的无界阻塞队列FIFO。SynchronousQueue不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作否则插入操作一直处于阻塞状态。DelayedWorkQueue具有优先级的延迟无界阻塞队列。可以根据任务自身的优先级顺序延迟执行。为什么要使用DelayedWorkQueue呢定时任务执行时需要取出最近要执行的任务所以任务在队列中每次出队时一定要是当前队列中执行时间最靠前的所以自然要使用优先级队列。Executors工具类// 可缓存publicstaticExecutorServicenewCachedThreadPool(){returnnewThreadPoolExecutor(0,Integer.MAX_VALUE,60L,TimeUnit.SECONDS,newSynchronousQueueRunnable());}// 定长publicstaticExecutorServicenewFixedThreadPool(intnThreads){returnnewThreadPoolExecutor(nThreads,nThreads,0L,TimeUnit.MILLISECONDS,newLinkedBlockingQueueRunnable());}// 单例publicstaticExecutorServicenewSingleThreadExecutor(){returnnewFinalizableDelegatedExecutorService(newThreadPoolExecutor(1,1,0L,TimeUnit.MILLISECONDS,newLinkedBlockingQueueRunnable()));}// 定时 来自 ScheduledThreadPoolExecutor extends ThreadPoolExecutorpublicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,newDelayedWorkQueue());}5.项目中怎么创建线程池使用ThreadPoolExecutor的原生构造函数创建这样可以更加明确每项参数的意义不使用Executors工具类创建。FixedThreadPool 和 SingleThreadPool:允许的任务队列LinkedBlockingQueue长度为 Integer.MAX_VALUE可能会堆积大量的请求从而导致 OOM。CachedThreadPool 和 ScheduledThreadPool:允许的创建线程数量为 Integer.MAX_VALUE可能会创建大量的线程从而导致 OOM。6.如何合理设置线程池参数线程池ThreadPoolExecutor参数设置7.线程池模拟分段抓取Mysql数据publicstaticvoidmain(String[]args){xxx(13000);}// CompletableFuture// long selectCount 13000;//假设 为 select count(*) from t 的返回结果publicstaticvoidxxx(longselectCount){ThreadPoolExecutorexecutornewThreadPoolExecutor(20,20,60,TimeUnit.SECONDS,newArrayBlockingQueue(200),newThreadPoolExecutor.AbortPolicy());classRUNimplementsRunnable{privateStringsql;Overridepublicvoidrun(){System.out.println(Thread.currentThread().getName() ~ sql);}publicRUN(Stringsql){this.sqlsql;}}// 模拟从数据库分段抓取数据 同步到ESlongstep2000;// 每次处理2000条数据longrsselectCount/step;if(rs0){CompletableFuture.runAsync(newRUN(select * from t limit 0,selectCount),executor);}else{longvselectCount%step;if(v!0){rs;}longlimitHead0;for(longi0;irs;i){CompletableFuture.runAsync(newRUN(select * from t limit limitHead,step),executor);limitHead2000;}}}pool-1-thread-1~select*from t limit0,2000pool-1-thread-4~select*from t limit6000,2000pool-1-thread-3~select*from t limit4000,2000pool-1-thread-6~select*from t limit10000,2000pool-1-thread-5~select*from t limit8000,2000pool-1-thread-7~select*from t limit12000,2000pool-1-thread-2~select*from t limit2000,2000
)
)
