分布式系列并发编程executor老师笔记

看源码java内存模型(JavaMemoryModelJMM)本身是一种抽象的概念，并不真实存在，段和构成数组对象的元素）的方式。由于JVM运行程序的实体是线程，而每个线程创值)必须通过主内存来完成，其简要过程如下图CPU调度——这是一个耗费时间和Java当中，线程通常都有五种状态，创建、就绪、运行、阻塞和start是此时线程调度程序还没有把该线程设置为当前线程此时处于就绪状态程运行之后，run如说某项资源就绪)之后再继续运行。sleep,suspend，wait第五是状态。如果一个线程的run方法执行结束或者调用stop方法后，该线程就可以用过jstackideadebug"LowMemoryDetector"负责对可使用内存进行检测，如果发现可用内存低，分配新的内"CompilerThread0"JITingclass。"SignalDispatcher"负责分发内部。"Finalizer"Finalizer方法。"ReferenceHandler"负责处理。"main"是主线程。"VMThread""VMPeriodicTaskThread"NEW状态是指线程刚创建,RUNNABLE状态是线程正在正常运行中,当然可能会有某种耗时计算/IO等待的操作/CPU时间片切换等,这个状态下发生的等待一般是其他系统资源,而不是锁,Sleep等BLOCKED这个状态下,是在多个线程有同步操作的场景,比如正在等待另一个线程的synchronized块的执行释放,synchronizedwait()方法,也就是WAITING这个状态下是指线程拥有了某个锁之后,wait方法,等待其他线程/拥有者调用notify/notifyAll一遍该线程可以继续下一步操作,这里要区分BLOCKED和WATING的区别,一个是在临界点外面等待进入,waitnotify,joinjoin了另外的线程的时候,WAITING状态,join的TIMED_WAITING这个状态就是有限的(时间限制)的WAITING,一般出现在调用wait(long),join(long)等情况下,sleep后,也会进入TIMED_WAITING状态TERMINATED这个状态下表示该线程的run方法已经执行完毕了,基本上就等于了(当时如果线程被持久持有,可能不会被回收)线程的优先级是将该线程的重要性传给了调度器、cpu处理线程顺序有一定的不确定，RunnableThread区别：Callable1.5run方call方法哦为什么有线程还要用线程池Java.util.concurrentExecutor(中文翻译是执行器)ThreadExecutorService（执行器） 两种基础线程池的选ScheduledThreadPoolExecutor123 区别有返回值4调用（offer>task放Run方法调用runwork方法5 6publicpublicstaticvoidmain(String[]args)ExecutorServiceexecutorService=Executors.newFixedThreadPool(1);//构建并且创建线程池赋executorService.submit(newRunnable(){publicvoidrun(){ }}new构造器 publicpublicThreadPoolExecutor(int TimeUnitunit,BlockingQueue<Runnable>workQueue,//阻塞队ThreadFactorythreadFactory,//线程工){ifif(command==thrownewNullPointerException();intc=ctl.get();//判断是否小 数量，是直接新增work成功后直接退if(workerCountOf(c)<corePoolSize){if(addWorker(command,true))cctl.get();失败后继续获取标}//判断是运行状态并且扔到workQueue里成功if(isRunning(c)&&workQueue.offer(command)){intrecheck=ctl.get();//再次check判断运行状态如果是非运行状态就移除出去&rejectif(!isRunning(recheck)&&remove(command))elseif(workerCountOf(recheck)==0)//否则发现可能运行线程0么增加一个nullworkeraddWorker(null,}elseif(!addWorker(commandfalse))//直接增加worker果不成功直接addWorker(null,false)来创建一个临时线程去阻塞队列中获取任务来执行。for(;;)intc=intrs=//Checkifqueueemptyonlyifif(rs>=SHUTDOWN!(rs==SHUTDOWNfirstTask==null!returnfalse;种情况1.如果非关闭状态2.不是这种情况（停止状态并且是null象并且workQueue等于null）for(;;)intwc=workerCountOf(c);if(wc>=CAPACITY||wc>=(core?corePoolSizereturnfalse;是否饱和容if(compareAndIncrementWorkerCount(cworkbreakcctl.get();//增加work败后继续递if(runStateOf(c)!=rs)continueretry;//elseCASfailedduetoworkerCountchange;retryinner}}booleanworkerStarted=booleanworkerAdded=false;Workerw=null;trywnewWorker(firstTask);//增加一个finalThreadt=if(tnull/判断是否finalReentrantLockmainLock=this.mainLock;try//Recheckwhileholding////BackoutonThreadFactoryfailureor//shutdownbeforelockacquired. intrs=或if(rs<SHUTDOWN(rs==SHUTDOWN&&firstTask==null))if(t.isAlive())//precheckthattisthrownewworkers.add(w);ints=workers.size();if(s>largestPoolSize=s;workerAdded=true;}}finally}){workerStarted=true;}}}finallyif(!workerStarted)}return//内循环主要是个增加线程数的CAS操firstTasknullfirstTask对象，然后去阻塞corePoolSize作为池中线程数量的最大值；为假，则以umPoolSize作为池中线程数量T.startrun方法执行>runWorkerThreadThreadwtThread.currentThread();//1.取到当前线程Runnabletask=w.firstTask;w.firstTask=null;w.unlock();//allowinterruptsbooleancompletedAbruptly=trywhile(tasknull||taskgetTasknull//获取任务看是否能拿到进Task中的队列polltake是真正拿到//workQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)超时取//Ifpoolis//ifnot,ensurethreadisnotinterrupted.//requiresarecheckinsecondcasetodeal//shutdownNowracewhileclearingif((runStateAtLeast(ctl.get(),STOP)||(Terrupted()&&runStateAtLeast(ctl.get(),STOP)))&&errupt();保线程是能中断try,;Throwablethrown=null;try{task.run();//执行任}catch(RuntimeExceptionx){thrown=x;throwx;}catch(Errorx){thrown=x;throw}catch(Throwablex)thrown=x;thrownew}finallyafterExecute(taskthrown//任务后的}}}finallytask=}}completedAbruptly=}finally,);}ifif(completedAbruptly)//Ifabrupt,thenworkerCountwasn'tfinalReentrantLockmainLock=try{completedTaskCountworkers.remove(w);//移除}finally}intc=if essThan(cSTOP//判断是否还有if(!completedAbruptly)intmin=allowCoreThreadTimeOut?0:corePoolSize;if(min==0&&!workQueue.isEmpty())min=if(workerCountOf(c)>=return;// cementnot}addWorker(addWorker(null,}线程池的策AbortAbort(默认当任务添加到线程池中被时，它将抛RejectedExecutionException当任务添加到线程池中被时，会程池当前正在运行的Thread线程池中处理被拒publicpublicScheduledThreadPoolExecutor(intcorePoolSize){super(corePoolSize,Integer.MAX_VALUE,0,NANOSECONDS,new}publicpublicScheduledFuture<?>scheduleAtFixedRate(RunnablelonginitialDelay,longperiod,TimeUnitunit){if(command==null||unit==null)thrownewNullPointerException();if(period<=thrownewIllegalArgumentException();ScheduledFutureTask<Void>sft=newnewtriggerTime(initialDelayunit),RunnableScheduledFuture<Void>t=decorateTask(command,sft);sft.outerTask=t;delayedExecute(t);return}delayedExecute#ScheduledThreadPoolExecutor方ifif(isShutdown())elsesuper.getQueue().add(task);//增加任务子类实现if(isShutdown()!canRunInCurrentRunState(task.isPeriodic&&判断是否已经停止}Offer#DelayedWorkQueuepublicpublicbooleanoffer(Runnablex){if(x==null)thrownewRunnableScheduledFutureeRunnableScheduledFuture)x//内部类finalReentrantLocklock=this.lock;try{inti=if(iqueue.length)//判断是否扩容size=i+1;if(i==0)queue[0esetIndex(e0);}elsesiftUp(i,e);}if(queue[0]==e){leader=null;available.signal();}finally}return}whilewhile(k>0)intparent=(k-1)>>>1;RunnableScheduledFuture<?>e=queue[parent];ifpareTo(e)>=queue[k]=e;setIndex(e,k);k=parent;}queue[k]=key;setIndex(key,k);ifif(other==this)//comparezeroifsamereturnif(otherinstanceofScheduledFutureTask){ScheduledFutureTask<?>x=(ScheduledFutureTask<?>)other;longdifftimex.time//判断if(diff<return-1;elseif(diff>0)returnelseif(sequenceNumber<x.sequenceNumber)return-1;return}longdiff=getDelay(NANOSECONDS)-other.getDelay(NANOSECONDS);return(diff<0)?-1:(diff>0)?1:0;work可以运行了？workintintwc=workerCountOf(ctl.get());if(wc<corePoolSize)addWorker(null,true);elseif(wc==0)addWorker(null,runworktakepublicpublicRunnableScheduledFuturetake()throwsInterruptedException{finalReentrantLocklock=this.lock;try{for(;;)RunnableScheduledFuturefirst=queue[0];if(firstnullelse{=if(delay<=0)//到时间了returnfinishPoll(first);elseif(leader!=available.await();elseThreadthisThread=Thread.currentThread();leader=thisThread;tryavailable.awaitNanos(delay);conditionawaitNanos}finallyif(leader==thisThread)leader=null;}}}}}finallyif(leader==null&&queue[0]!=null)}}privateprivateRunnableScheduledFuturefinishPoll(RunnableScheduledFuturef)ints=--longlongp=if(p0)//看这个大于0和小于0区timep;//假如延迟了这个时间早过了，+当前时候肯定还是过的RunnableScheduledFutureRunnableScheduledFuturexqueue[s];//重新排列queue[s]=null;if(s!=siftDown(0,x);setIndex(f,-1);returnf;} scheduleWithFixedDelay有什么区别吗（publicpublicvoidrun()booleanperiodicisPeriodic//判断是否是周期if(!canRunInCurrentRunState(periodic))elseif(!periodic)elseif(ScheduledFutureTask.super.runAndReset()){reExecutePeriodic(outerTask);置task有异常捕捉}}finalfinallongdeadline=timed?System.nanoTime()+nanos:WaitNodeq=null;booleanqueued=false;for(;;){if(Terruptedcheck程中thrownew}ints=if(s>COMPLETING){ if(q!=null)q.thread=null;returns;}elseif(s==COMPLETING)//cannottimeoutelseif(q==qnewWaitNode();//生成一个waintelseifqueuedpareAndSwapObject(this,q.next=waiters,q)表的对象下一个置成当前的elseif(timed)nanos=deadline-System.nanoTime();if(nanos<=0L){returnstate;}Lo