一、基本介绍与应用1.1 基本介绍线程使用上的问题new Thread().start();1、线程的频繁创建销毁比较浪费资源2、线程的数量过多会造成CPU频繁的进行上下文切换。(上下文切换过程并没有进行运算)思考 如何实现线程的复用池化技术举例: 连接池、对象池、内存池、线程池 。。。池化技术的核心复用线程池提前创建一系列的线程保存在这个线程池中。有任务要执行的时候从线程池中取出线程来执行。没有任务的时候放回线程池。Java中提供的线程池Executors 线程池创建工厂类创建线程池对象的方法:newFixedThreadPool 固定线程数量public static ExecutorService newFixedThreadPool(int nThreads){returnnew ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueueRunnable());}newSingleThreadExecutor 只有一个线程的线程池public static ExecutorServicenewSingleThreadExecutor(){returnnew FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1,1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueueRunnable()));}newCachedThreadPool 可以缓存的线程池 -理论上来说有多少请求该线程池就可以创建多少的线程来处理。public static ExecutorServicenewCachedThreadPool(){returnnew ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueueRunnable());}newScheduledThreadPool 提供了按照周期执行的线程池. -Timerpublic static ScheduledExecutorService newScheduledThreadPool(int corePoolSize){returnnew ScheduledThreadPoolExecutor(corePoolSize);}1.2 应用如果有些任务执行时间很长想要关闭可以利用守护线程的方式强制关闭。被守护的线程是主线程只要主线程执行完成线程池就会强制关闭可以配合awaitTermination方法使用只了解有这种用法即可不建议用到实时业务场景因为守护线程有可能会导致任务没执行结束就退出造成安全性问题。package cn.zxj.threadpool;importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.ThreadPoolExecutor;importjava.util.concurrent.TimeUnit;public class ThreadPoolExecutorDemo{public static void main(String[]args)throws InterruptedException{ThreadPoolExecutor threadPoolExecutorbulidThreadPoolExecutor();for(int i0;i10;i){threadPoolExecutor.execute(new Task());}threadPoolExecutor.shutdown();threadPoolExecutor.awaitTermination(5, TimeUnit.SECONDS);System.out.println(强制关闭);}public static ThreadPoolExecutorbulidThreadPoolExecutor(){ThreadPoolExecutor threadPoolExecutornew ThreadPoolExecutor(4,10,30, TimeUnit.SECONDS,new ArrayBlockingQueue(5), r -{Thread tnew Thread(r);t.setDaemon(true);returnt;},new ThreadPoolExecutor.AbortPolicy());returnthreadPoolExecutor;}static class Task implements Runnable{Override public voidrun(){try{Thread.sleep(1000*30);}catch(InterruptedException e){e.printStackTrace();}}}}二、线程池设计猜想需求实现线程的重复使用.分析如何实现线程的复用?让线程实现复用的唯一的方法就是让线程不结束。那如何让线程执行新的任务呢也就是说任务怎么给它执行[共享内存]- List.add()线程一直处于运行状态合理吗有任务来的时候执行没有任务的时候阻塞结论通过阻塞队列的方式来实现线程池中线程的复用。线程池的实现原理的过程推演三、线程池原理分析3.1 线程池参数3.1.1 线程池设置参数public ThreadPoolExecutor(int corePoolSize, //核心线程数 int maximumPoolSize, //最大线程数 long keepAliveTime, //存活时间 TimeUnit unit, //存活单位 BlockingQueueRunnableworkQueue, //阻塞队列 ThreadFactory threadFactory, //线程工厂用来创建工作线程 的。 默认实现自定义线程池中线程的名字 RejectedExecutionHandler handler){//拒绝执行策略 。默 认实现if(corePoolSize0||maximumPoolSize0||maximumPoolSizecorePoolSize||keepAliveTime0)throw new IllegalArgumentException();if(workQueuenull||threadFactorynull||handlernull)throw new NullPointerException();this.accSystem.getSecurityManager()null ? null:AccessController.getContext();this.corePoolSizecorePoolSize;this.maximumPoolSizemaximumPoolSize;this.workQueueworkQueue;this.keepAliveTimeunit.toNanos(keepAliveTime);this.threadFactorythreadFactory;this.handlerhandler;}注意:核心线程数和工作线程数只是参数数量的配置实际源码中并不区分核心线程和工作线程比如创建或者销毁是不区分核心线程和工作线程的只要达到了参数配置效果即可。3.1.2 AtomicInteger变量ThreadPoolExecutor有一个AtomicInteger变量叫ctlcontrol的简写一共32位高3位存线程池状态runState一共5种状态RunningShutdownStopTidyingTerminate低29位存当前有效线程数workerCountAtomicInteger ctlnew AtomicInteger(ctlOf(RUNNING,0));int COUNT_BITSInteger.SIZE -3;//COUNT_BITS32int CAPACITY(1COUNT_BITS)-1;//高3位000低29位全为1 int RUNNING-1COUNT_BITS;//高3位111低29位全为0 int SHUTDOWN0COUNT_BITS;//高3位000低29位全为0 int STOP1COUNT_BITS;//高3位001低29位全为0 int STOPPING2COUNT_BITS;//高3位010低29位全为0 int TERMINATED3COUNT_BITS;//高3位011低29位全为0 //得到运行状态 int runStateOf(int c){returnc~CAPACITY;//c 与 CAPACITY的取反}//得到工作线程数量 int workerCountOf(int c){returncCAPACITY;}//初始化ctl private static int ctlOf(int rs, intwc){returnrs|wc;}等价类实现class Control{int workCount;STATE runState;//初始化ctl static getControl(int workCount, STATE runState){returnnew Control(workCount, runState);}//得到工作线程数量 static int getWorkCount(Control ctl){returnctl.workCount;}//得到运行状态 static STATE getRunState(Control ctl){returnctl.runState;}}enum STATE{RunningShutdownStopTidyingTerminate}3.2 源码分析线程池中的核心线程是延迟初始化的。先初始化核心线程并执行任务(不是从阻塞队列获取的)。调用阻塞队列的方法把task存进去。offer() - true/false如果true 说明当前的请求量不大 核心线程就可以搞定。false增加工作线程非核心线程)并执行任务(不是从阻塞队列获取的)。如果增加工作线程失败说明当前的工作线程数量达到了最大的线程数直接调用拒绝策略。3.2.1 execute方法public void execute(Runnablecommand){if(commandnull)throw new NullPointerException();int cctl.get();//判断当前工作线程数是否小于核心线程数延迟初始化if(workerCountOf(c)corePoolSize){if(addWorker(command,true))//添加工作线程的同时执行commandreturn;cctl.get();}//workQueue.offer 添加到阻塞队列if(isRunning(c)workQueue.offer(command)){int recheckctl.get();if(!isRunning(recheck)remove(command))reject(command);elseif(workerCountOf(recheck)0)addWorker(null,false);}// 如果阻塞队列满了则添加工作线程扩容的线程elseif(!addWorker(command,false))reject(command);//拒绝策略}execute方法总结如果当前运行的线程数少于 corePoolSize则创建新线程来执行任务(延迟初始化)当任务的数量超过corePoolSize数量会将任务加入阻塞队列阻塞排队当阻塞队列也满了之后那么将会继续创建(maximumPoolSize-corePoolSize)个数量的线程来执行任务如果任务处理完成maximumPoolSize-corePoolSize额外创建的线程等待keepAliveTime之后被自动销毁如果达到maximumPoolSize阻塞队列还是满的状态那么将根据不同的拒绝策略对应处理3.2.2 addWorker方法private boolean addWorker(Runnable firstTask, boolean core){retry: //case1 通过原子操作来增加线程数量.for(;;){int cctl.get();int rsrunStateOf(c);// Checkifqueue empty onlyifnecessary.if(rsSHUTDOWN!(rsSHUTDOWNfirstTasknull!workQueue.isEmpty()))returnfalse;for(;;){intwcworkerCountOf(c);if(wcCAPACITY||wc(core ? corePoolSize:maximumPoolSize))returnfalse;if(compareAndIncrementWorkerCount(c))//通过原子操作来增加线程数量.breakretry;cctl.get();// Re-read ctlif(runStateOf(c)!rs)continueretry;//elseCAS failed due to workerCount change;retry inner loop}}//case2 初始化工作线程. boolean workerStartedfalse;boolean workerAddedfalse;Worker wnull;try{//构建一个工作线程,此时还没有启动. wnew Worker(firstTask);final Thread tw.thread;if(t!null){final ReentrantLock mainLockthis.mainLock;mainLock.lock();try{// Recheckwhileholding lock. // Back out on ThreadFactory failure orif// shut down before lock acquired. int rsrunStateOf(ctl.get());if(rsSHUTDOWN||(rsSHUTDOWNfirstTasknull)){if(t.isAlive())// precheck that t is startable throw new IllegalThreadStateException();workers.add(w);//添加到一个容器中。 int sworkers.size();if(slargestPoolSize)//重新更新largestPoolSize largestPoolSizes;workerAddedtrue;//添加成功。}}finally{mainLock.unlock();}if(workerAdded){t.start();//启动线程 workerStartedtrue;}}}finally{//失败 回滚。if(!workerStarted)addWorkerFailed(w);}returnworkerStarted;}3.2.3 worker.run()方法相关public voidrun(){runWorker(this);}final void runWorker(Worker w){Thread wtThread.currentThread();Runnable taskw.firstTask;w.firstTasknull;w.unlock();// allow interrupts boolean completedAbruptlytrue;try{//while循环保证当前线程不结束. 直到task为nullwhile(task!null||(taskgetTask())!null){//表示当前线程正在运行一个任务如果其他地方要shutdown().你必须要等我执行完成。 w.lock();//Worker继承了AQS -实现了互斥锁 // If pool is stopping, ensure thread is interrupted;//ifnot, ensure thread is not interrupted. This // requires a recheckinsecondcaseto deal with // shutdownNow racewhileclearing interruptif((runStateAtLeast(ctl.get(),STOP)||(Thread.interrupted()runStateAtLeast(ctl.get(),STOP)))!wt.isInterrupted())wt.interrupt();//是否应该触发中断 try{beforeExecute(wt, task);//空的实现 -Throwable thrownnull;try{task.run();//执行task.run}catch(RuntimeException x){thrownx;throw x;}catch(Error x){thrownx;throw x;}catch(Throwable x){thrownx;throw new Error(x);}finally{afterExecute(task, thrown);//空的实现 -}}finally{tasknull;w.completedTasks;w.unlock();}}completedAbruptlyfalse;}finally{processWorkerExit(w, completedAbruptly);}}private RunnablegetTask(){boolean timedOutfalse;// Did the last poll()timeout?for(;;){//cas 自旋 int cctl.get();int rsrunStateOf(c);// 如果线程池已经结束状态直接返回null. 需要清理掉所有的工作线程 // Checkifqueue empty onlyifnecessary.if(rsSHUTDOWN(rsSTOP||workQueue.isEmpty())){decrementWorkerCount();returnnull;//}intwcworkerCountOf(c);//是否允许超时 // * allowCoreThreadTimeOut 为true // * 如果当前的工作线程数量大于核心线程数 // Are workers subject to culling? boolean timedallowCoreThreadTimeOut||wccorePoolSize;if((wcmaximumPoolSize||(timedtimedOut))(wc1||workQueue.isEmpty())){if(compareAndDecrementWorkerCount(c))//cas 减少工作线程数量。returnnull;//表示要销毁当前工作线程.continue;}//获取任务的过程 try{Runnable rtimed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS):workQueue.take();//如果阻塞队列没有任务当前工作线程会阻塞在这里if(r!null)returnr;timedOuttrue;}catch(InterruptedException retry){//中断异常. timedOutfalse;}}}3.2.4 拒绝执行策略报错(默认)public void rejectedExecution(Runnable r, ThreadPoolExecutor e){throw new RejectedExecutionException(Task r.toString() rejected from e.toString());}普通线程调用task.run()public void rejectedExecution(Runnable r, ThreadPoolExecutor e){if(!e.isShutdown()){r.run();}}将队列中头部的任务丢弃,然后把当前任务添加到阻塞队列public void rejectedExecution(Runnable r, ThreadPoolExecutor e){if(!e.isShutdown()){e.getQueue().poll();e.execute(r);}}直接丢弃这个任务public void rejectedExecution(Runnable r, ThreadPoolExecutor e){}存储起来后续等待空闲之后重试(自定义去完成)四、线程数量设置常规设置IO密集型 CPU 2core1CPU利用率不高多设线程数等待时让给其它没有IO任务的线程执行提高效率。CPU密集型 CPU core 1CPU利用率很高避免过多线程上下文切换。动态设置在配置中心存放参数创建ThreadPoolExecutor对象读取配置文件设置参数自定义阻塞队列读取配置文件设置队列长度同时根据线程池监控数据及时调整参数配置。推荐设置线程数CPU核数×目标CPU利用率×(1平均等待时间/平均工作时间)