`

Concurrent - CyclicBarrier

 
阅读更多

原创转载请注明出处：http://agilestyle.iteye.com/blog/2343979

 

CyclicBarrier

CyclicBarrier允许一组线程互相等待，直到到达某个公共屏障点（common barrier point），这些线程必须实时地互相等待，这种情况下就可以使用CyclicBarrier来方便地实现这样的功能。另外，CyclicBarrier的公共屏障点可以重用。

CyclicBarrierTest1.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest1 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> System.out.println("All invoked!!!"));

        for (int i = 0; i < 5; i++) {
            Thread thread = new Thread(new MyThread(cyclicBarrier));
            thread.start();
        }
    }

    public static class MyThread implements Runnable {

        private CyclicBarrier cyclicBarrier;

        public MyThread(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run() {
            try {
                Thread.sleep((int) (Math.random() * 10000));

                System.out.println(Thread.currentThread().getName() + " invoked..." + System.currentTimeMillis());

                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
}

Note:

CyclicBarrier cyclicBarrier = new CyclicBarrier(5, () -> System.out.println("All invoked!!!"));

设置最大为5个parties同行者，也就是5个线程都执行了await()方法后程序才可以继续向下运行，否则这些线程彼此互相等待，一直呈阻塞状态。

 

线程个数大于parties数量进行分批处理

CyclicBarrierTest2.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest2 {
    public static void main(String[] args) {
        try {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> System.out.println("All invoked!!!"));

            for (int i = 0; i < 4; i++) {
                Thread thread = new Thread(new MyThread(cyclicBarrier));
                thread.start();
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static class MyThread implements Runnable {

        private CyclicBarrier cyclicBarrier;

        public MyThread(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run() {
            try {
                System.out.println(Thread.currentThread().getName() + " begin=" + System.currentTimeMillis() + " wait 2 threads to continue");

                cyclicBarrier.await();

                System.out.println(Thread.currentThread().getName() + " end=" + System.currentTimeMillis() + " have 2 threads to continue");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
}
Run 

Note:

从运行结果来看，每出现2个线程就开始RUN

 

getNumberWaiting()

getNumberWaiting()作用是获得有几个线程已经到达屏障点。

CyclicBarrierTest3.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest3 {
    public static void main(String[] args) {
        try {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(3);

            for (int i = 0; i < 10; i++) {
                Thread thread = new Thread(new MyThread(cyclicBarrier));
                thread.start();
                Thread.sleep(1000);
                System.out.println(cyclicBarrier.getNumberWaiting());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static class MyThread implements Runnable {

        private CyclicBarrier cyclicBarrier;

        public MyThread(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        @Override
        public void run() {
            try {
                cyclicBarrier.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }
}

Run


Note:

从运行结果来看，CyclicBarrier具有屏障重置性，也就是parties的值可以重置为0

 

使用CyclicBarrier屏障重置性实现多阶段的比赛实验

CyclicBarrierTest4.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest4 {
    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(2);

        Service service = new Service(cyclicBarrier);

        for (int i = 0; i < 4; i++) {
            Thread thread = new Thread(new MyThread(service));
            thread.setName("Thread " + (i + 1));
            thread.start();
        }

    }

    public static class Service {
        private CyclicBarrier cyclicBarrier;

        public Service(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        public void beginRun() {
            try {
                Thread.sleep((int) (Math.random() * 10000));

                System.out.println(Thread.currentThread().getName() + " "
                        + System.currentTimeMillis() + " begin No.1 phase "
                        + (cyclicBarrier.getNumberWaiting() + 1));

                cyclicBarrier.await();

                System.out.println(Thread.currentThread().getName() + " "
                        + System.currentTimeMillis() + " end No.1 phase "
                        + (cyclicBarrier.getNumberWaiting()));

                Thread.sleep((int) (Math.random() * 10000));

                System.out.println(Thread.currentThread().getName() + " "
                        + System.currentTimeMillis() + " begin No.2 phase "
                        + (cyclicBarrier.getNumberWaiting() + 1));

                cyclicBarrier.await();

                System.out.println(Thread.currentThread().getName() + " "
                        + System.currentTimeMillis() + " end No.2 phase "
                        + (cyclicBarrier.getNumberWaiting()));
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    public static class MyThread implements Runnable {

        private Service service;

        public MyThread(Service service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.beginRun();
        }
    }
}

Run


Note:

CyclicBarrier类的parties值从1到2，然后再恢复到0，证明CyclicBarrier类的屏障点是可以复用的，另外线程1234每到达一个屏障点时的组合是随机的，sleep勇士最少的互相组合，继续向下一个屏障行进。 

 

isBroken()

isBroken()作用是查询屏障是否处于损坏状态

CyclicBarrierTest5.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest5 {
    public static class MyService {
        private CyclicBarrier cyclicBarrier;

        public MyService(CyclicBarrier cyclicBarrier) {
            this.cyclicBarrier = cyclicBarrier;
        }

        public void testMethod() {
            try {
                System.out.println(Thread.currentThread().getName() + " invoked..., wait for others...");

                if (Thread.currentThread().getName().equals("Thread-2")) {
                    System.out.println("Thread-2 ran into an accident!!!");
                    Thread.sleep(3000);
                    Integer.parseInt("a");
                    //Thread.currentThread().interrupt();
                }

                cyclicBarrier.await();

                System.out.println("All invoked..., begin to execute...");

                System.out.println(Thread.currentThread().getName() + " done...");
            } catch (InterruptedException e) {
                System.out.println(Thread.currentThread().getName() + " In InterruptedException e " + cyclicBarrier.isBroken());
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                System.out.println(Thread.currentThread().getName() + " In BrokenBarrierException e " + cyclicBarrier.isBroken());
                e.printStackTrace();
            }
        }
    }

    public static class MyThread implements Runnable {
        private MyService service;

        public MyThread(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(4, () -> System.out.println("All invoked..."));

        MyService service = new MyService(cyclicBarrier);

        for (int i = 0; i < 4; i++) {
            new Thread(new MyThread(service)).start();
        }
    }
}

Run


Note:

有一个程序出现异常报错，则其他线程继续等待，并不影响程序运行的主流程。

 

更改MyService.java的testMethod，注释掉Integer.parseInt("a")，添加interrupt()

if (Thread.currentThread().getName().equals("Thread-2")) {
	System.out.println("Thread-2 ran into an accident!!!");
	Thread.sleep(3000);
	//Integer.parseInt("a");
	Thread.currentThread().interrupt();
}

Run


Note:

所有线程都进入了catch块，其中Thread-2进入了InterruptedException块，其他3个线程进入了BrokenBarrierException块。CyclicBarrier处理线程的interrupt会使用全有或全无的破坏模型，意思是如果有一个线程由于中断或者超时提前离开了屏障点，其他所有在屏障点等待的线程也会抛出BrokenBarrierException或InterruptedException异常，并且离开屏障点。

 

await(long timeout, TimeUnit unit)

await(long timeout, TimeUnit unit)作用是如果在指定的时间内达到parties数量，则程序继续向下运行，否则如果出现超时，则抛出TimeoutException异常。

CyclicBarrierTest6.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class CyclicBarrierTest6 {
    public static class MyService {
        private CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> System.out.println("All done! " + System.currentTimeMillis()));

        public void testMethod() {
            try {
                System.out.println(Thread.currentThread().getName() + " prepare... " + System.currentTimeMillis());

                if(Thread.currentThread().getName().equals("Thread-0")) {
                    System.out.println("Thread-0 invoked cyclicBarrier.await(5, TimeUnit.SECONDS);");
                    cyclicBarrier.await(5, TimeUnit.SECONDS);
                }

                if(Thread.currentThread().getName().equals("Thread-1")) {
                    System.out.println("Thread-1 invoked cyclicBarrier.await();");
                    cyclicBarrier.await();
                }

                System.out.println(Thread.currentThread().getName() + " start... " + System.currentTimeMillis());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }

    public static class MyThreadA implements Runnable {
        private MyService service;

        public MyThreadA(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static class MyThreadB implements Runnable {
        private MyService service;

        public MyThreadB(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static void main(String[] args) {
        MyService service = new MyService();

        Thread a = new Thread(new MyThreadA(service));
        a.start();

        Thread b = new Thread(new MyThreadB(service));
        b.start();
    }
}

Run


 

getNumberWaiting()和getParties()

getNumberWaiting()的作用是获取有几个线程已经到屏障点

getParties()的作用是获取parties的个数

CyclicBarrierTest7.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class CyclicBarrierTest7 {
    public static class MyService {
        private CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> System.out.println("All done! " + System.currentTimeMillis()));

        public void testMethod() {
            try {
                System.out.println(Thread.currentThread().getName() + " prepare... " + System.currentTimeMillis());

                if(Thread.currentThread().getName().equals("C")) {
                    Thread.sleep(Integer.MAX_VALUE);
                }

                cyclicBarrier.await();

                System.out.println(Thread.currentThread().getName() + " start... " + System.currentTimeMillis());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    public static class MyThreadA implements Runnable {
        private MyService service;

        public MyThreadA(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static class MyThreadB implements Runnable {
        private MyService service;

        public MyThreadB(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static class MyThreadC implements Runnable {
        private MyService service;

        public MyThreadC(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static void main(String[] args) {
        try {
            MyService service = new MyService();

            Thread a = new Thread(new MyThreadA(service));
            a.setName("A");
            a.start();

            Thread b = new Thread(new MyThreadB(service));
            b.setName("B");
            b.start();

            Thread c = new Thread(new MyThreadC(service));
            c.setName("C");
            c.start();

            Thread.sleep(3000);

            System.out.println(service.cyclicBarrier.getParties());

            System.out.println(service.cyclicBarrier.getNumberWaiting());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Run


 

reset()

reset()的作用是重置屏障 

CyclicBarrierTest8.java

package org.fool.java.concurrent.cyclicbarrier;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest8 {
    public static class MyService {
        private CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> System.out.println("All done! " + System.currentTimeMillis()));

        public void testMethod() {
            try {
                System.out.println(Thread.currentThread().getName() + " prepare... " + System.currentTimeMillis());

                cyclicBarrier.await();

                System.out.println(Thread.currentThread().getName() + " start... " + System.currentTimeMillis());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    }

    public static class MyThreadA implements Runnable {
        private MyService service;

        public MyThreadA(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static class MyThreadB implements Runnable {
        private MyService service;

        public MyThreadB(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static class MyThreadC implements Runnable {
        private MyService service;

        public MyThreadC(MyService service) {
            this.service = service;
        }

        @Override
        public void run() {
            service.testMethod();
        }
    }

    public static void main(String[] args) {
        try {
            MyService service = new MyService();

            Thread a = new Thread(new MyThreadA(service));
            a.setName("A");
            a.start();

            Thread b = new Thread(new MyThreadB(service));
            b.setName("B");
            b.start();

            // thread c uninitialized
            Thread.sleep(3000);
            service.cyclicBarrier.reset();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Run


Note:

屏障被重置后，2个线程出现BrokenBarrierException 

 

 

Reference

Java并发编程核心方法与框架




 

  • 大小: 26.8 KB
  • 大小: 18.6 KB
  • 大小: 28.6 KB
  • 大小: 14.4 KB
  • 大小: 37.3 KB
  • 大小: 37.7 KB
  • 大小: 70.5 KB
  • 大小: 41.8 KB
  • 大小: 14.2 KB
  • 大小: 38.5 KB
分享到：
| Concurrent - CountDownLatch
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    concurrent-1.3.4-sources.jar

    5. **CyclicBarrier和Semaphore**：这两个类是同步辅助工具，CyclicBarrier用于一组线程互相等待，直到所有线程都到达屏障点后才能继续执行；Semaphore则用于限制同时访问特定资源的线程数量，常用于限流和保护共享...

    concurrent-all-in-one.pdf

    - `CyclicBarrier`：允许多个线程同步开始，等待所有线程到达屏障点后一起继续。 - `Semaphore`：信号量，用于限制同时访问特定资源的线程数量。 2. **同步机制** - `Synchronized`：关键字确保同一时间只有一个...

    backport-util-concurrent-3.1.jar和geronimo-stax-api_1.0_spec-1.0.1.jar

    它提供了与Java 5及更高版本相同的并发工具类，如线程池、Future、CyclicBarrier等。这些工具可以提高多线程应用程序的性能和可维护性，尤其是在处理大量并发任务时。如果Eclipse Axis2 Codegen插件在旧版本的JDK上...

    backport-util-concurrent（2.2 /3.1）

    - 在3.1版本中，backport-util-concurrent引入了Phaser，这是一个可重用的同步帮助器类，支持多个参与者之间的有界同步，类似于CyclicBarrier和CountDownLatch，但更灵活。Phaser可以自动调整参与者的数量，并且在...

    concurrent-1.3.4.jar

    同步工具类：concurrent包提供了一些同步工具类，如CountDownLatch、CyclicBarrier、Semaphore等，可以帮助开发者实现复杂的线程协作和同步任务。 并发集合类：concurrent包提供了一些并发集合类，如...

    aduna-commons-concurrent-2.2.jar.zip

    5. **实用工具类**：库中还包括了一些实用工具类，如CountDownLatch、CyclicBarrier、Semaphore等，它们在多线程协调和同步方面提供了强大的支持，可以帮助开发者构建复杂的并发场景。 在实际项目中，要充分利用...

    concurrent-resource-test:Java并发包源码学习

    6. **CyclicBarrier**: `CyclicBarrier`允许多个线程等待，直到达到某个公共屏障点。当所有线程都到达屏障点后，它们会被一次性释放，继续执行。 7. **Atomic 类**: `Atomic`类如`AtomicInteger`, `AtomicLong`提供...

    backport-util-concurrent-Java50-3.1-src:我的项目只需要库文件

    此外，backport-util-concurrent还包括其他关键组件，如`Semaphore`（信号量）、`CyclicBarrier`（循环栅栏）、`Phaser`（屏障）等，它们都是协调多线程操作的强大工具。例如，Semaphore用于限制对资源的并发访问，...

    Concurrent-Web-Crawler:用Java实现的多线程Web搜寻器

    Java提供了多种并发控制机制，如线程、线程池、同步块、Semaphore、CyclicBarrier等。在这个项目中，很可能使用了ExecutorService（线程池）来管理多个线程，通过提交任务到线程池而不是直接创建新线程，可以更有效...

    java-concurrent-ways:演示几种用Java实现并发的方法

    8. **并发工具类**：`java.util.concurrent`包中包含很多并发工具，如`CountDownLatch`, `CyclicBarrier`, `Semaphore`等，用于控制线程的并发数量和同步点。 9. **原子类（Atomic Classes）**：如`AtomicInteger`,...

    java并发编程实战源码-concurrent-programming:《Java并发编程实战》源码整理

    3. **同步机制**：Java提供了多种同步机制，包括`synchronized`关键字、`Lock`接口及其实现（如`ReentrantLock`）、`Semaphore`信号量、`CyclicBarrier`和`CountDownLatch`等。这些机制用于控制并发访问资源，防止...

    Concurrent-programming:并发编程分配

    学习这些内容可以帮助你深入理解如何在Java中编写高效的并发程序，包括如何避免线程安全问题，如何设计高并发应用，以及如何使用高级并发工具如`Semaphore`(信号量)、`CountDownLatch`(计数器门)和`CyclicBarrier`...

    Concurrent-Assignment:并发编程的多人Snake分配

    Java提供了丰富的并发API，包括`java.util.concurrent`包中的各种类和接口，如`ExecutorService`、`Thread`、`Semaphore`、`CyclicBarrier`、`CountDownLatch`等，它们为并发编程提供了强大的支持。 1. **线程池...

    java并发工具包 java.util.concurrent中文版用户指南pdf

    13. 栅栏 CyclicBarrier 14. 交换机 Exchanger 15. 信号量 Semaphore 16. 执行器服务 ExecutorService 17. 线程池执行者 ThreadPoolExecutor 18. 定时执行者服务 ScheduledExecutorService 19. 使用 ForkJoinPool ...

    java并发源码-java-concurrent-test:和朱晔一起复习Java并发系列文章

    7. **并发工具类**：`CountDownLatch`、`CyclicBarrier`、`Semaphore`等，这些工具常用于多线程间的同步和协调。 8. **volatile关键字**：了解其内存可见性和禁止指令重排序的作用，以及在并发编程中的应用场景。 ...

    backport-util-concurrent_java_backport_

    《回溯：Java中的backport-util-concurrent》 在Java编程领域，backport-util-concurrent是一个重要的库，尤其对于那些需要在Java 5之前的版本中使用Java 5及更高版本引入的并发特性的开发者来说。这个库的目标是将...

    Java并发工具包java.util.concurrent用户指南中英文对照阅读版.pdf

    栅栏 CyclicBarrier 14. 交换机 Exchanger 15. 信号量 Semaphore 16. 执行器服务 ExecutorService 17. 线程池执行者 ThreadPoolExecutor 18. 定时执行者服务 ScheduledExecutorService 19. 使用 ForkJoinPool 进行...

    常用的两个jar包

    3. **CyclicBarrier和Semaphore**：这两个同步工具类在1.5版本之后被广泛使用，但在早期的`concurrent-1.3.2.jar`中已经存在。`CyclicBarrier`用于让一组线程等待彼此到达一个共同的屏障点，而`Semaphore`则用于控制...

    backport-util-concurrent

    "backport-util-concurrent" 是一个著名的开源库，它的主要目标是将Java 5和更高版本中的并发特性（如java.util.concurrent包）回移植到Java 1.3和1.4等早期版本。这个库使得那些无法或不想升级JVM的开发者也能享受...

    Java并发工具包java.util.concurrent用户指南中英文对照阅读版

    13. 栅栏 CyclicBarrier 14. 交换机 Exchanger 15. 信号量 Semaphore 16. 执行器服务 ExecutorService 17. 线程池执行者 ThreadPoolExecutor 18. 定时执行者服务 ScheduledExecutorService 19. 使用 ForkJoinPool ...

Magicbox
Global site tag (gtag.js) - Google Analytics