我们在前面的文章学习CountDownLatch和CyclicBarrier的时候了解到,CountDownLatch计数器只能使用一次,不能重复利用,CyclicBarrier虽然可以重复利用但是parties在初始化时指定后就不能再修改了,所以用法上不是很灵活。
Phaser是JDK1.7引入的多线程同步工具,其功能类似于CountDownLatch和CyclicBarrier的合集,它提供的方法更灵活和丰富,但是使用难度也比较大。
Phaser用作CountDownLatchCountDownLatch能完成的功能,Phaser也能实现,其实现方式如下:
publicstaticvoidmain(String[]args){//初始化不指定parties数量,parties默认为0finalPhaserphaser=newPhaser();//主线程调用register,数量 1phaser.register();IntStream.range(0,5).forEach(i->newThread(()->{//线程调用register方法,数量 1phaser.register();try{//模拟业务执行时间TimeUnit.SECONDS.sleep(newRandom().nextInt(2));System.out.println(Thread.currentThread().getName() "完成任务");//线程完成任务调用arrive方法,不阻塞phaser.arrive();}catch(InterruptedExceptione){e.printStackTrace();}},"T-" i).start());//主线程arrive,但是它要等待第一个阶段结束,其前提是所有的线程都要arrivephaser.arriveAndAwaitAdvance();System.out.println("注册的任务数量:" phaser.getRegisteredParties());System.out.println(Thread.currentThread().getName() "被唤醒");}
初始化parties数量为0,启动5个线程执行业务,每个线程调用register使得parties数量加1,并在业务执行结束之后调用arrive方法。
main线程调用arriveAndAwaitAdvance方法一直阻塞直到所有的线程都arrive。
运行结果为:
main线程直到所有任务执行结束才会被唤醒,实现了CountDownLatch的效果。
Phaser用作CyclicBarrierCyclicBarrier可以使子任务阻塞直到所有线程到达屏障,在Phaser中只需要把子任务到达的方法arrive修改为arriveAndAwaitAdvance即可实现CyclicBarrier的效果。
publicstaticvoidmain(String[]args){//初始化不指定数量finalPhaserphaser=newPhaser();phaser.register();IntStream.range(0,5).forEach(i->newThread(()->{//线程调用register使得数量加1phaser.register();try{TimeUnit.SECONDS.sleep(newRandom().nextInt(5));System.out.println(Thread.currentThread().getName() "完成任务");//线程完成任务调用arriveAndAwaitAdvance阻塞直到所有线程都到达phaser.arriveAndAwaitAdvance();System.out.println(Thread.currentThread().getName() "继续执行任务");}catch(InterruptedExceptione){e.printStackTrace();}},"T-" i).start());//主线程arrive,但是它要等待第一个阶段结束,其前提是所有的线程都要arrivephaser.arriveAndAwaitAdvance();System.out.println("注册的任务数量:" phaser.getRegisteredParties());System.out.println(Thread.currentThread().getName() "被唤醒");}
初始化parties数量为0,启动5个线程执行业务,每个线程调用register使得parties数量加1,并在业务执行结束之后调用arriveAndAwaitAdvance,此方法会使线程阻塞直到所有的线程都到达之后才继续运行。
运行结果为:
通过结果我们看到main线程和所有T线程在所有线程到达之后继续并发执行,实现了CyclicBarrier的效果。
Phaser(阶段)编号在Phaser中有一个比较重要的概念:Phaser(阶段)编号。
一个Phaser中所有相关联的parties都arrive(到达后),Phaser就会从下一个阶段开始,除非Phaser已经被终止或者销毁。
Phaser为每一个Phaser都提供了一个编号,初始化为0,所有任务分片到达之后加1。
Phaser阶段可以理解为每次重用Phaser都会为其分配一个阶段编号,每次全部任务到达都顺序加1。
我们通过一段代码查看不同情况下Phaser的编号。
publicstaticvoidmain(String[]args){finalPhaserphaser=newPhaser(3);System.out.println("Phaser阶段编号为:" phaser.getPhase());phaser.arrive();phaser.arrive();phaser.arrive();System.out.println("Phaser阶段编号为:" phaser.getPhase());//继续重用phaserphaser.bulkRegister(1);//现在parties数量为4phaser.arrive();phaser.arrive();phaser.arrive();System.out.println("Phaser阶段编号为:" phaser.getPhase());phaser.arrive();System.out.println("Phaser阶段编号为:" phaser.getPhase());}
getPhase()方法返回Phaser的当前Phaser编号,最大的Phaser编号为Integer.MAX_VALUE,超过最大值将重新从0开始。当Phaser被终止时,Phaser编号将返回负值。
Phaser中大部分方法都会返回Phaser编号,理解Phaser编号是理解Phaser方法的基础。
Phaser方法总结Phaser构造器
parties为分片任务数量,可以构造时指定,也可以通过调用register方法对其加1。
parent为Phaser的层级,如果指定了parent,那么子Phaser的Phaser编号会以父Phaser为准,且父Phaser的分片任务等于所有子Phaser的分片任务和父Phaser的分片任务总和。
因parent会使线程管理比较复杂,一般很少使用,所以我们开发中最常用的是以下构造中的前2个。
Phaser():不指定parties数量,默认为0;并且不指定parent,parent默认为null。
Phaser(int parties):指定parties数量,不指定parent。
Phaser(Phaser parent):指定parent,不指定parties数量。
Phaser(Phaser parent, int parties):同时指定parent和parties数量。
register()方法
publicintregister(){returndoRegister(1);}
register方法的作用是为Phaser增加一个未到达的parties,并且返回Phaser的编号。使用该方法时,有的时候会陷入阻塞,比如前一个Phaser执行onAdvance方法耗时比较长,那么如果此时想要一个新的parties通过register方法加入就会陷入阻塞。
publicstaticvoidmain(String[]args){finalPhaserphaser=newPhaser(1){@OverrideprotectedbooleanonAdvance(intphase,intregisteredParties){try{TimeUnit.SECONDS.sleep(30);}catch(InterruptedExceptione){e.printStackTrace();}returnsuper.onAdvance(phase,registeredParties);}};newThread(()->{phaser.arrive();}).start();TimeUnit.SECONDS.sleep(2);longtime=System.currentTimeMillis();phaser.register();System.out.println("当前阶段编号为:" phaser.getPhase());System.out.println("耗时:" (System.currentTimeMillis()-time) "ms");}
上述代码中重写了onAdvance方法,初始化parties为1,启动一个线程调用arrive方法分片任务到达,此时onAdvance得以被执行。
main线程调用register会被阻塞,因为此时onAdvance正在做收尾工作,它在执行完逻辑之后才会接纳新的分片进来,因此onAdvance尽量不要写太复杂的代码。
onAdvance在每一个Phaser阶段中所有的任务分片达到之后都会被执行,它还决定了Phaser是否终止:当onAdvance方法返回true表明该Phaser将被终止,接下来将不能再使用。onAdvance可以实现CyclicBarrier中Runnable的作用。
bulkRegister(int parties)方法
publicintbulkRegister(intparties){if(parties<0)thrownewIllegalArgumentException();if(parties==0)returngetPhase();returndoRegister(parties);}
该方法的作用与register()一样,通过源码我们可以看到他们调用的都是doRegister方法。不同点是register只允许注册一个分片到Phaser,但是bulkRegister方法可以允许0个或者多个分片到达Phaser。
arrive()和arriveAndAwaitAdvance()方法
arrive方法是到达Phaser的下一个阶段,不会等待其他分片,它返回的阶段编号为当前Phaser的编号。
arriveAndAwaitAdvance方法是到达Phaser的下一个阶段,需要阻塞等待其他线程的分片都到达。而且此方法返回的是下一个Phaser的编号。
publicstaticvoidmain(String[]args){finalPhaserphaser=newPhaser(2);System.out.println("当前Phaser的编号:" phaser.getPhase());phaser.arrive();System.out.println("第一次arrive后的Phaser的编号:" phaser.getPhase());phaser.arrive();System.out.println("第二次arrive后的Phaser的编号:" phaser.getPhase());}
getPhase编号初始化为0,第一次调用arrive方法,返回当前编号为0,因我们定义了两个切片,此时只有一个分片任务到达。第二次调用arrive方法,返回当前编号为1,因为此时所有的分片任务都已到达,Phaser已经进入下一个阶段了。
arriveAndDeregister()方法
该方法可以到达下一个Phaser阶段之外,还会将当前Phaser的parties数量减少一个。
publicstaticvoidmain(String[]args){finalPhaserphaser=newPhaser(2);System.out.println("当前Phaser的编号:" phaser.getPhase() "&&当前的注册的parties:" phaser.getRegisteredParties());phaser.arriveAndDeregister();System.out.println("当前Phaser的编号:" phaser.getPhase() "&&当前的注册的parties:" phaser.getRegisteredParties());phaser.arriveAndDeregister();System.out.println("当前Phaser的编号:" phaser.getPhase() "&&当前的注册的parties:" phaser.getRegisteredParties());}
初始化定义parties数量为2,每次调用arriveAndDeregister打印出的parties数量都会减1,最后一次输出的Phaser为负数,表示当前的Phaser已经被销毁。
awaitAdvance(int phase)方法
awaitAdvance有3种不同的表现:awaitAdvance(int phase)、awaitAdvanceInterruptibly(int phase)、awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)。
这三个方法更多的是关注阻塞而不关注arrive,等待某个Phaser关联的所有分片是否已经到达某个指定的阶段。使用这3个方法不会影响Phaser内部分片的arrive和unarrive变化的。
awaitAdvance(int phase)方法的主要作用是等待与Phaser相关联的分片都到达某个指定的Phaser阶段编号,如果有某个分片任务未到达,那么此方法就会进入阻塞,且不能打断。如果入参Phaser与当前Phaser的阶段编号不一致,该方法会立即返回,如果Phaser已经被销毁,该方法的返回值为负数。
publicstaticvoidmain(String[]args){finalPhaserphaser=newPhaser(1);newThread(()->{System.out.println("当前Phaser的阶段编号为:" phaser.getPhase());intphaserNum=phaser.awaitAdvance(phaser.getPhase());System.out.println("phaserNum为:" phaserNum);}).start();TimeUnit.SECONDS.sleep(30);phaser.arriveAndAwaitAdvance();System.out.println("当前Phaser的阶段编号为:" phaser.getPhase());}
thread线程调用awaitAdvance方法一直阻塞直到main线程调用arriveAndAwaitAdvance方法分片任务到达后,thread线程退出阻塞,且awaitAdvance方法返回的Phaser编号为下一个阶段的Phaser编号。
awaitAdvanceInterruptibly(int phase)与awaitAdvance(int phase)方法作用相同,只是其可以被打断。
awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)在可以被打断的基础上增加了设置超时等待的功能。
Phaser的介绍到这里就结束了,对其方法的理解还需在使用中加深印象。
JUC包中还有一个并发工具类Exchanger,可以实现线程间的数据交换,感兴趣的小伙伴自行查找资料了解喔。
本周勾勾会继续学习并发编程知识,陆续推出并发容器和线程池的文章,期待和你一起进步!!!