java多线程工具类

　　准备继续巩固下知识，先更新一些关于java多线程的学习笔记
　　但是总觉得掌握得不扎实，所以重新来过，好事多磨嘛。
　　本篇准备学习的内容：ForkJoin并发工具类CountDownLatchCyclicBarrierSemaphoreExchangeCallable、Future和FutureTask
　　tip：本片偏向的应用，底层源码后在后面的博客更新ForkJoin
　　ForkJoin是分而治之。
　　规模为N的问题，N阈值，直接解决，N阈值，将N分解为K个小规模子问题，子问题互相对立，与原问题形式相同，将子问题的解合并得到原问题的解。内部可以理解为递归调用。
　　客官，先给您上菜authormark产生整形数组publicclassMakeArray｛数组长度publicstaticfinalintLENGTH1000000；publicstaticint〔〕makeArray（）｛new一个随机数发生器RandomrnewRandom（）；int〔〕resultnewint〔LENGTH〕；for（inti0；iLENGTH；i）｛用随机数填充数组result〔i〕r。nextInt（LENGTH3）；｝returnresult；｝｝Author：jimmyDate：202141720：19Description：使用forkjoin实现数字相加这里仅展示如何使用forkjion关于越界等其他问题不进行优化publicclassSumDemo｛privatestaticclassSumTaskextendsRecursiveTaskInteger｛阈值privatefinalstaticintTHRESHOLDMakeArray。LENGTH10；表示我们要实际统计的数组privateint〔〕data；开始统计的下标privateintfromIndex；统计到哪里结束的下标privateinttoIndex；publicSumTask（int〔〕data，intfromIndex，inttoIndex）｛this。datadata；this。fromIndexfromIndex；this。toIndextoIndex；｝计算方法return1、规模没有超过阈值则直接计算2、规模超过阈值则拆分为子任务invokeAlljoinOverrideprotectedIntegercompute（）｛if（toIndexfromIndexTHRESHOLD）｛intcount0；for（intifromIndex；itoIndex；i）｛countcountdata〔i〕；｝returncount；｝else｛拆分逻辑fromIndex。。。。mid。。。。toIndexintmid（fromIndextoIndex）2；SumTaskleftnewSumTask（data，fromIndex，mid）；SumTaskrightnewSumTask（data，mid1，toIndex）；invokeAll（left，right）；returnleft。join（）right。join（）；｝｝｝paramargspublicstaticvoidmain（String〔〕args）｛1、创建forkJoinPool实例ForkJoinPoolpoolnewForkJoinPool（）；int〔〕dataMakeArray。makeArray（）；2、创建任务SumTasktasknewSumTask（data，0，data。length1）；3、执行任务pool。invoke（task）；4、获取结果Integerresulttask。join（）；System。out。println（result）；｝｝
　　总结下forkjoin首先，创建一个自己的任务执行类（继承RecursiveTask），重写处理方法，这里面要注意阈值的处理及子任务的划分，当然这个要看具体的需求场景。这里两个重要的方法invokeAllRecursiveTask。join（）。用来将任务执行和获取任务执行结果
　　然后我们可以进行forkjoin的范式调用
　　1、创建forkJoinPool实例
　　2、创建任务
　　3、执行任务
　　4、获取结果并发工具类CountDownLatch
　　CyclicBarrier
　　Semaphore
　　Exchange
　　’’
　　这里在业务开发中相对比较常用的可能是CountDownLatch和CyclicBarrier。
　　现在我们想想有一场短跑比赛。一共10个选手参赛。（这里有11个线程，10个选手线程和一个开始比赛线程）
　　只有当所有选手都到场的时候才能开始比赛。这个时候就可以使用CountDownLatch。开始比赛线程必须等10个选手线程都完成才能开始。
　　现在换一个场景，玩moba类游戏，dota2、LOL啥的，都需要一个匹配，匹配成功后开始游戏。假设有10位玩家。匹配过程中所有玩家都必须加载完成才能继续进行后面的比赛。那么这里就可以使用CyclicBarrier。注意这里只有10个匹配线程，每个线程在加载完成的时候阻塞，当10个匹配线程都加载完成后，一起进入后面的比赛。
　　废话一堆，showmecodeCountDownLatchpackageconcurrentClass。countdown；importjava。util。concurrent。CountDownLatch；publicclassCutDownLatchTest｛publicstaticvoidmain（String〔〕args）｛finalCountDownLatchlatchnewCountDownLatch（2）；构造方法，放入2个线程newThread（）｛publicvoidrun（）｛try｛System。out。println（选手子线程Thread。currentThread（）。getName（）选手准备入场）；Thread。sleep（3000）；System。out。println（选手子线程Thread。currentThread（）。getName（）选手到场）；latch。countDown（）；｝catch（InterruptedExceptione）｛e。printStackTrace（）；｝｝；｝。start（）；newThread（）｛publicvoidrun（）｛try｛System。out。println（选手子线程Thread。currentThread（）。getName（）选手准备入场）；Thread。sleep（3000）；System。out。println（选手子线程Thread。currentThread（）。getName（）选手到场）；latch。countDown（）；｝catch（InterruptedExceptione）｛e。printStackTrace（）；｝｝；｝。start（）；使用await（）方法让主线程等待子线程执行完后在执行try｛latch。await（）；System。out。println（选手到齐，开始比赛）；｝catch（InterruptedExceptione）｛｝｝｝输出选手子线程Thread0选手准备入场选手子线程Thread1选手准备入场选手子线程Thread0选手到场选手子线程Thread1选手到场选手到齐，开始比赛cyclicBarrierpackageconcurrentClass。cyclicBarrier；importjava。util。concurrent。CyclicBarrier；publicclassCyclicBarrierDemo｛staticCyclicBarriercnewCyclicBarrier（2）；publicstaticvoidmain（String〔〕args）｛newThread（newRunnable（）｛Overridepublicvoidrun（）｛try｛System。out。println（玩家Thread。currentThread（）。getName（）加载完成）；Thread。sleep（1000）；c。await（）；System。out。println（玩家Thread。currentThread（）。getName（）时间：System。currentTimeMillis（）准备进入比赛）；｝catch（Exceptione）｛｝System。out。println（阿杜跟）；｝｝）。start（）；newThread（newRunnable（）｛Overridepublicvoidrun（）｛try｛System。out。println（玩家Thread。currentThread（）。getName（）加载完成）；Thread。sleep（3000）；c。await（）；System。out。println（玩家Thread。currentThread（）。getName（）时间：System。currentTimeMillis（）准备进入比赛）；｝catch（Exceptione）｛｝System。out。println（哈撒剋）；｝｝）。start（）；｝｝输出玩家Thread1加载完成玩家Thread0加载完成玩家Thread1时间：1618666778562准备进入比赛玩家Thread0时间：1618666778562准备进入比赛哈撒剋阿杜跟Semaphore控制同时访问某个特定资源的线程数量，用在流量控制packageconcurrentClass。semaphore；importjava。util。concurrent。Semaphore；publicclassSemaphoreDemo｛publicstaticvoidmain（String〔〕args）｛intN8；工人数SemaphoresemaphorenewSemaphore（5）；机器数目for（inti0；iN；i）｛newWorker（i，semaphore）。start（）；｝｝staticclassWorkerextendsThread｛privateintnum；privateSemaphoresemaphore；publicWorker（intnum，Semaphoresemaphore）｛this。numnum；this。semaphoresemaphore；｝Overridepublicvoidrun（）｛try｛semaphore。acquire（）；System。out。println（工人this。num占用一个机器在生产。。。）；Thread。sleep（1000）；System。out。println（工人this。num释放出机器）；semaphore。release（）；｝catch（InterruptedExceptione）｛e。printStackTrace（）；｝｝｝｝输出工人0占用一个机器在生产。。。工人1占用一个机器在生产。。。工人2占用一个机器在生产。。。工人3占用一个机器在生产。。。工人4占用一个机器在生产。。。工人1释放出机器工人4释放出机器工人3释放出机器工人5占用一个机器在生产。。。工人2释放出机器工人7占用一个机器在生产。。。工人0释放出机器工人6占用一个机器在生产。。。工人6释放出机器工人5释放出机器工人7释放出机器Exchange
　　两个线程间的数据交换（这个我接触很少）packageconcurrentClass。exchange；importjava。util。HashSet；importjava。util。Set；importjava。util。concurrent。Exchanger；publicclassExchangeDemo｛privatestaticfinalExchangerSetStringexchangenewExchangerSetString（）；publicstaticvoidmain（String〔〕args）｛第一个线程newThread（newRunnable（）｛Overridepublicvoidrun（）｛SetStringsetAnewHashSetString（）；存放数据的容器try｛添加数据set。add（。。。。。）setAexchange。exchange（setA）；交换set处理交换后的数据｝catch（InterruptedExceptione）｛｝｝｝）。start（）；第二个线程newThread（newRunnable（）｛Overridepublicvoidrun（）｛SetStringsetBnewHashSetString（）；存放数据的容器try｛添加数据set。add（。。。。。）set。add（。。。。。）setBexchange。exchange（setB）；交换set处理交换后的数据｝catch（InterruptedExceptione）｛｝｝｝）。start（）；｝｝
　　尾巴phaser类Callable、Future和FutureTask
　　这套工具可以获取子线程的结果信息。FutureTask可以理解为继承了Runnable和Future接口的Callable包装类Future是用来提供获取线程处理结果的接口packagefuture；importjava。util。Random；importjava。util。concurrent。Callable；importjava。util。concurrent。ExecutionException；importjava。util。concurrent。FutureTask；publicclassFutureDemo｛实现Callable接口，允许有返回值privatestaticclassUseCallableimplementsCallableInteger｛privateintsum；OverridepublicIntegercall（）throwsException｛System。out。println（Callable子线程开始计算）；Thread。sleep（2000）；for（inti0；i300；i）｛sumsumi；｝System。out。println（Callable子线程计算完成，结果sum）；returnsum；｝｝publicstaticvoidmain（String〔〕args）throwsInterruptedException，ExecutionException｛UseCallableuseCallablenewUseCallable（）；FutureTaskIntegerfutureTasknewFutureTaskInteger（useCallable）；newThread（futureTask）。start（）；RandomrnewRandom（）；Thread。sleep（1000）；if（r。nextBoolean（））｛随机决定是获得结果还是终止任务System。out。println（GetUseCallableresultfutureTask。get（））；｝else｛System。out。println（中断计算）；futureTask。cancel（true）；｝｝｝输出Callable子线程开始计算Callable子线程计算完成，结果44850GetUseCallableresult44850