高并发技术之java线程池源码解析

高并发技术之Java线程池源目题 问高并发技术之Java线程池源目题 问 线程池框 第一层结 接口简 实现 辅助 完成服 源码原...........................................................................................................................线程池执行原 调度线程池原 异步结果源码分 1.线程池的模块结2.示例&原1.线程池包含哪些东西2.1.线程池包含哪些东西2.线程池原理3.调度线程池原理4.调度线程池怎么实现FixRate，FixDelay？和他们之间的区别5.怎么取消的第一层结 （启动一个线程行 （more执行器，直接执行 （启动一个线程行 （more执行器，直接执行 （线性执行器java.util.concurrent.ExecutorService执行器服务java.util.concurrent.Executor（执行器，执行方法java.util.concurrent.ExecutorService（执行服务）包含服务的生命周java.util.concurrent.ScheduledExecutorService（调度相关的服务 （普通的的线程池实现类oolExecutor（调度实现类 channel–AIO辅助完成服线程池辅助完成服线程池执行原构造数量，任务队列容器，存活时间，线程工厂，处理器//固定线程ExecutorServiceexecutorService= executorService.execute(newRunnable()初始化构造publicThre longTimeUnitBlockingQueue<Runnable>ThreadFactoryif(corePoolSize<0keepAliveTime<thrownewif(workQueue==null||threadFactory==thrownewthis.acc=System.初始化构造publicThre longTimeUnitBlockingQueue<Runnable>ThreadFactoryif(corePoolSize<0keepAliveTime<thrownewif(workQueue==null||threadFactory==thrownewthis.acc=System.getSecurityManager()==nullnullActhis.corePoolSize= umPoolSize this.workQueue=this.keepAliveTime=this.threadFactory=this.handler=}publicvoidrun()}});//runnable接 ubmit(newCallable<String>()publicStringcall()throwsExceptionreturn}});//callable接if(compareAndIncrementWorkerCount(cworkreturnfalsewc>=(core?corePoolSize if(wc>=CAPACITYwc=for(;;)if(compareAndIncrementWorkerCount(cworkreturnfalsewc>=(core?corePoolSize if(wc>=CAPACITYwc=for(;;)returnfalse1如果非运行状态2（nullworkQueue!Task==null!(rs==SHUTDOWNif(rs>=SHUTDOWN//Checkifqueueemptyonlyif rs= c=for(;;)c= if(workerCountOf(c)<corePoolSize){if(addWorker( d,true))cctl.get}if(isRunning(c)&&workQueue.offer( d)){recheck=ctl.get();if(!isRunning(recheck)&&remove( elseif(workerCountOf(recheck)00nullworker。addWorker(null,false);}elseif(!addWorker( d,false))//直接增加worker如果不成功直接reject thrownewNullPoif d==if(!}finally}}workerStarted=if(workerAdded)workstart}}finally}workerAdded=largestPoolSize=if(s>s=workers.add(w);thrownewIllegalThreadSif(t.isAlive())// ttis(rs==if(!}finally}}workerStarted=if(workerAdded)workstart}}finally}workerAdded=largestPoolSize=if(s>s=workers.add(w);thrownewIllegalThreadSif(t.isAlive())// ttis(rs==SHUTDOWN Task==null))if(rs<SHUTDOWNrs=runS//shutdownbeforelockacquired锁定后并重新检查下是否 //BackoutonThreadFactoryfailureor//RecheckwhileholdingtryfinalReentrantLockmainLock=if(tnullfinalThreadt=wnew tryWorkerw=workerAdded=workerStarted=}}//elseCASfailedduetoworkerCountchange;retryinnercontinueif eOf(c)!=cctl.get();Re-readctl增加workbreaktask=}finally}afterExecute(task,thrown}finallythrown=x;thrownew}catch(Throwablex)thrown=x;throw}catch(Errorx)thrown=x;throw}catch(RuntimeExceptiontask=}finally}afterExecute(task,thrown}finallythrown=x;thrownew}catch(Throwablex)thrown=x;throw}catch(Errorx)thrown=x;throw}catch(RuntimeExceptionx)tryThrowablethrown=beforeExecute(wt,task);trywt.errupt();//!wt.isif((runSeeast(ctl.get(),STOP)//shutdownNowracewhile //requiresarecheckinsecondcasetodeal//ifnot,ensurethreadis errupted//Ifpoolis ,ensurethread while(tasknull||taskgetTask(nulltrycompletedAbruptly=w.unlock();// Task=Runnabletask= ThreadwtThread.currentThread();//1.return}if(min==0&&!min=if(workerCountOf(c)>=return;//replacementnot}addWorker(null,}if(!completedAbruptly)min=allowCoreThreadTimeOut?0:if nif(min==0&&!min=if(workerCountOf(c)>=return;//replacementnot}addWorker(null,}if(!completedAbruptly)min=allowCoreThreadTimeOut?0:if n(cSTOPif(completedAbruptly)//Ifabrupt,thenworkerCountwasn'tfinalReentrantLockmainLock=trycompletedTaskCount workers.remove(w);//移除}finally}c=}}}completedAbruptly=}finallyprosWorkerExit(w,调度线程池原调构造DelayedWorkQueue延迟队通过DelayedWorkQueue延迟队列实现offer获取对象的延if(x==throw调度线程池原调构造DelayedWorkQueue延迟队通过DelayedWorkQueue延迟队列实现offer获取对象的延if(x==thrownewNullPoRunnableScheduledFuture<?>eRunnableScheduledFuture<?>)x;对finalReentrantLocklock=tryi=if(iqueue.length扩ifelsesuper.getQueue().add(task);//增加任if(isShutdown()!canRunInCurrentRunSe(task.isPeriodic())}publicScheduledThre corePoolSize) eger.MAX_VALUE,0,new}oolExecutor.DelayedWorkQueue#siftUp（二堆算法）保证相同if(other==this)//comparezeroifsamereturnif(otherinstanceofScheduledFutureTask)ScheduledFutureTask<?>x=longdifftimex.time判断if(diff<oolExecutor.DelayedWorkQueue#siftUp（二堆算法）保证相同if(other==this)//comparezeroifsamereturnif(otherinstanceofScheduledFutureTask)ScheduledFutureTask<?>x=longdifftimex.time判断if(diff<return-while(k>0)parent=(k-1)>>>RunnableScheduledFuture<?>e=if pareTo(e)>=queue[k]=setIndex(e,k=}queue[k]=setIndex(key,size=i+if(i==0)queue[0]=setIndex(e0第一个直接设置索引和下标}elsesiftUp(i,e);选到}if(queue[0]==e)leader=available.signal();//唤醒所有的被挤压的wait线}}finally}return确保有work执work运行的时候调用queue的take方finalReentrantLocklock=lock.locktryfor(;;) =queue确保有work执work运行的时候调用queue的take方finalReentrantLocklock=lock.locktryfor(;;) =queue[0];//获取第一个对if ==elselongdelay .getDelay(NANOSECONDS);//延迟时if(delay0)//到时间return =null;//don'tretainrefwhileif(leader!=available.await();//因为没有执行线程初始化，所以等等什么时候有了自己被他人唤elsewc=if(wc<addWorker(null,elseif(wc==addWorker(null,elseif(diff>returnelseif(sequenceNumber<return-return}longdiff=getDelay(NANOSECONDS)-return(diff<0)?-1:(diff>0)?1:怎么实现固定率的periodic=if elseif怎么实现固定率的periodic=if elseifelseif(ScheduledFutureTask.super.runAndReset()){//有periodreturns=--RunnableScheduledFuture<?>xqueue[s];//重排序队queue[s]=if(s!=siftDown(0,setIndex(f,-ThreadthisThread=Thread.currentThread();leader=thisThread;tryavailable.awaitNanos(delay);//各condition}finallyif(leader==thisThread)leader=null;}}}}}finallyif(leader==null&&queue[0]!=null)}怎么拿到的异步任务结果finallongdeadline=timed?System.nanoTime()+nanos:WaitNodeq=怎么拿到的异步任务结果finallongdeadline=timed?System.nanoTime()+nanos:WaitNodeq=queued=for(;;)if(Thread.errupted/check中throw }s=sif(sCOMPLETING){//判断是否完if(q!=q.thread=returnlongp=if(p>timep;//假如延迟了这个时间早过了，+当前时候肯定还是过的timetriggerTime(-p);前的任务延scheduleAtFixedRatescheduleWithFixedDelay有什么区别吗}什么时候回填的结果那if(se!=NEWpareAndSwapObject(this,nullThread.currentThread态不是new者runner置不成功直接退tryCallable<V>c=if(c!=null&&什么时候回填的结果那if(se!=NEWpareAndSwapObject(this,nullThread.currentThread态不是new者runner置不成功直接退tryCallable<V>c=if(c!=null&&se==NEW)Vtryresultc.call();//运行ok时候返回ran=}catch(Throwableex)result=ran=}if(ran)//正常成功setresult}elseif(s==COMPLETING)//cannottimeoutelseif(q==q=newWaitNode();//生成一个 对els