原创转载请注明出处:http://agilestyle.iteye.com/blog/2344591
Phaser提供了动态增减parties计数,这点比CyclicBarrier类操作parties更加方便,通过若干个方法来控制多个线程之间同步运行的结果,还可以实现针对某一个线程取消同步运行的效果,而且支持屏障处等待,在等待时还支持中断或非中断等功能,使用Java并发类对线程进行分组同步控制时,Phaser比CyclicBarrier功能更加强大,推荐使用。
arriveAndAwaitAdvance()
arriveAndAwaitAdvance()作用是当前线程已经到达屏障,在此等待一段时间,等条件满足后继续向下一个屏障继续执行
PhaserTest1.java
package org.fool.java.concurrent.phaser; import java.util.concurrent.Phaser; public class PhaserTest1 { public static class Service { private Phaser phaser; public Service(Phaser phaser) { this.phaser = phaser; } public void testMethodA() { System.out.println(Thread.currentThread().getName() + " A1 begin " + System.currentTimeMillis()); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " A1 end " + System.currentTimeMillis()); System.out.println(Thread.currentThread().getName() + " A2 begin " + System.currentTimeMillis()); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " A2 end " + System.currentTimeMillis()); } public void testMethodB() { try { System.out.println(Thread.currentThread().getName() + " B1 begin " + System.currentTimeMillis()); Thread.sleep(5000); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " B1 end " + System.currentTimeMillis()); System.out.println(Thread.currentThread().getName() + " B2 begin " + System.currentTimeMillis()); Thread.sleep(5000); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " B2 end " + System.currentTimeMillis()); } catch (InterruptedException e) { e.printStackTrace(); } } } public static class ThreadA implements Runnable { private Service service; public ThreadA(Service service) { this.service = service; } @Override public void run() { service.testMethodA(); } } public static class ThreadB implements Runnable { private Service service; public ThreadB(Service service) { this.service = service; } @Override public void run() { service.testMethodA(); } } public static class ThreadC implements Runnable { private Service service; public ThreadC(Service service) { this.service = service; } @Override public void run() { service.testMethodB(); } } public static void main(String[] args) { Phaser phaser = new Phaser(3); Service service = new Service(phaser); Thread t1 = new Thread(new ThreadA(service)); t1.setName("Thread-A"); t1.start(); Thread t2 = new Thread(new ThreadB(service)); t2.setName("Thread-B"); t2.start(); Thread t3 = new Thread(new ThreadC(service)); t3.setName("Thread-C"); t3.start(); } }
Run
PhaserTest2.java
package org.fool.java.concurrent.phaser; import java.util.concurrent.Phaser; public class PhaserTest2 { public static class Service { private Phaser phaser; public Service(Phaser phaser) { this.phaser = phaser; } public void testMethodA() { System.out.println(Thread.currentThread().getName() + " A1 begin " + System.currentTimeMillis()); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " A1 end " + System.currentTimeMillis()); System.out.println(Thread.currentThread().getName() + " A2 begin " + System.currentTimeMillis()); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " A2 end " + System.currentTimeMillis()); } public void testMethodB() { try { System.out.println(Thread.currentThread().getName() + " B1 begin " + System.currentTimeMillis()); Thread.sleep(5000); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " B1 end " + System.currentTimeMillis()); } catch (InterruptedException e) { e.printStackTrace(); } } } public static class ThreadA implements Runnable { private Service service; public ThreadA(Service service) { this.service = service; } @Override public void run() { service.testMethodA(); } } public static class ThreadB implements Runnable { private Service service; public ThreadB(Service service) { this.service = service; } @Override public void run() { service.testMethodA(); } } public static class ThreadC implements Runnable { private Service service; public ThreadC(Service service) { this.service = service; } @Override public void run() { service.testMethodB(); } } public static void main(String[] args) { Phaser phaser = new Phaser(3); Service service = new Service(phaser); Thread t1 = new Thread(new ThreadA(service)); t1.setName("Thread-A"); t1.start(); Thread t2 = new Thread(new ThreadB(service)); t2.setName("Thread-B"); t2.start(); Thread t3 = new Thread(new ThreadC(service)); t3.setName("Thread-C"); t3.start(); } }
Run
Note:
从运行结果来看,当计数不足时,线程呈阻塞状态,不继续向下运行,因为线程C仅仅执行了一次arriveAndAwaitAdvance()方法导致了这样的结果,所以当出现这样无法继续向下一个屏障继续执行的情况,需要使用arriveAndDeregister()
arriveAndDeregister()
arriveAndDeregister()作用是使当前线程退出,并且使parties值减1。
PhaserTest3.java
package org.fool.java.concurrent.phaser; import java.util.concurrent.Phaser; public class PhaserTest3 { public static class Service { private Phaser phaser; public Service(Phaser phaser) { this.phaser = phaser; } public void testMethodA() { System.out.println(Thread.currentThread().getName() + " A1 begin " + System.currentTimeMillis()); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " A1 end " + System.currentTimeMillis()); System.out.println(Thread.currentThread().getName() + " A2 begin " + System.currentTimeMillis()); phaser.arriveAndAwaitAdvance(); System.out.println(Thread.currentThread().getName() + " A2 end " + System.currentTimeMillis()); } public void testMethodB() { try { System.out.println(Thread.currentThread().getName() + " B1 begin " + System.currentTimeMillis()); Thread.sleep(5000); System.out.println("before arriveAndDeregister(): " + phaser.getRegisteredParties()); phaser.arriveAndDeregister(); System.out.println("after arriveAndDeregister(): " + phaser.getRegisteredParties()); System.out.println(Thread.currentThread().getName() + " B1 end " + System.currentTimeMillis()); } catch (InterruptedException e) { e.printStackTrace(); } } } public static class ThreadA implements Runnable { private Service service; public ThreadA(Service service) { this.service = service; } @Override public void run() { service.testMethodA(); } } public static class ThreadB implements Runnable { private Service service; public ThreadB(Service service) { this.service = service; } @Override public void run() { service.testMethodA(); } } public static class ThreadC implements Runnable { private Service service; public ThreadC(Service service) { this.service = service; } @Override public void run() { service.testMethodB(); } } public static void main(String[] args) { Phaser phaser = new Phaser(3); Service service = new Service(phaser); Thread t1 = new Thread(new ThreadA(service)); t1.setName("Thread-A"); t1.start(); Thread t2 = new Thread(new ThreadB(service)); t2.setName("Thread-B"); t2.start(); Thread t3 = new Thread(new ThreadC(service)); t3.setName("Thread-C"); t3.start(); } }
Run
Reference
Java并发编程核心方法与框架
相关推荐
JavaEE源代码 concurrent-1.3.2JavaEE源代码 concurrent-1.3.2JavaEE源代码 concurrent-1.3.2JavaEE源代码 concurrent-1.3.2JavaEE源代码 concurrent-1.3.2JavaEE源代码 concurrent-1.3.2JavaEE源代码 concurrent-...
- copy %AXIS2_HOME%\lib\backport-util-concurrent-3.1.jar 到%ECLIPSE_HOME%\plugins\Axis2_Codegen_Wizard_1.3.0\lib - 注册此 jar 包: 編輯 %ECLIPSE_HOME%\plugins\Axis2_Codegen_Wizard_1.3.0\plugin.xml , ...
concurrent-1.3.4.jar
backport-util-concurrent-3.1.jar 和 geronimo-stax-api_1.0_spec-1.0.1.jar 复制到 MyEclipse 6.5\eclipse\plugins\Axis2_Codegen_Wizard_1.3.0\lib 文件夹下。 (3).注册此 jar 包: 修改MyEclipse 6.5\eclipse...
官方版本,亲测可用
3. **并发处理**:`org.jodconverter.concurrent`包提供了并发处理工具,允许在多线程环境中高效地执行大量转换任务。 4. **日志记录**:JodConverter集成了SLF4J(Simple Logging Facade for Java),方便开发者...
《并发编程库 concurrent-1.3.4-sources.jar 深度解析》 在Java编程领域,"concurrent"一词通常与多线程和并发处理相关,它指的是能够同时执行多个任务的能力。这里提到的`concurrent-1.3.4-sources.jar`是一个特定...
"backport-util-concurrent-3.1.jar" 文件到 Axis2_Codegen_Wizard_1.3.0 的 lib 目录中 , 同时修改 Axis2_Codegen_Wizard_1.3.0 下的 plugin.xml 文件 , 在 <runtime> 中添加 <library name="lib/geronimo-stax-...
Concurrent-Utils Utilities for Java concurrent library. This is a library contains some useful and smart utility class for Java concurrent library. Shelly, HermesEventBus and AndroidDataStorage are ...
concurrent-1.3.2.ja
《深入解析Atlassian Util Concurrent库:0.0.12版本》 在IT行业中,高效且可靠的并发处理是系统性能优化的关键因素之一。Atlassian公司,以其强大的协作工具如Jira、Confluence等闻名,也提供了许多开源工具来支持...
《并发编程全方位解析》 并发编程是现代计算机系统中不可或缺的一部分,尤其是在多核处理器和分布式系统环境下。本章将深入探讨Java并发编程的核心概念、工具和最佳实践。 1. **Java并发组件概述** ...
- 在3.1版本中,backport-util-concurrent引入了Phaser,这是一个可重用的同步帮助器类,支持多个参与者之间的有界同步,类似于CyclicBarrier和CountDownLatch,但更灵活。Phaser可以自动调整参与者的数量,并且在...
backport-util-concurrent-1.0.jar,
patterns-for-concurrent-and-networked-objects