Java如何使用并发工具CyclicBarrier呢?

下文笔者讲述CyclicBarrier的方法及示例分享
学习完本篇之后,你将掌握CyclicBarrier对象的使用方法及示例

CyclicBarrier对象的场景:
有若干个线程
  如:
  有五个线程，需要它们都到达了某一个点之后才能开始一起执行
	 也就是说假如其中只有四个线程到达了这个点
		还差一个线程没到达
		此时这四个线程都会进入等待状态
		直到第五个线程也到达了这个点之后
		这五个线程才开始一起进行执行状态
这个临界点就是使用CyclicBarrier进行设置

例:CyclicBarrier示例

package com.java265;
 
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
  
public class TestCyclicBarrier {
	
	private static String str = "";
	
//	private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
  //参数>0（可以reset重置）

	
	private static CyclicBarrier cyclicBarrier = new CyclicBarrier(3, () -> {
	    //带action，阻塞个数达到后执行（由最后一个await线程执行）
		str.length();//测试异常
		System.out.println("所有await停止阻塞。");
	});
 
	/**
	 * @param args
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) {
		System.out.println("parties:" + cyclicBarrier.getParties());
		System.out.println("======1=====");
		test1();
		threadsleep(1000);
		System.out.println("======2=====");
		test2();
		threadsleep(1000);
		System.out.println("======3=====");
		test3();
		threadsleep(1000);
		System.out.println("======4=====");
		test4();
		threadsleep(1000);
		System.out.println("main finished");
	}
	
	public static void test1() {
		awaitThread(1, cyclicBarrier);
		awaitThread(2, cyclicBarrier);
		awaitThread(3, cyclicBarrier);
	}
	
	public static void test2() {//超时异常
		cyclicBarrier.reset();
		awaitThread(4, cyclicBarrier);//其他线程超时
		awaitTimeThread(5, cyclicBarrier, 1);//超时
		System.out.println("Broken？" + cyclicBarrier.isBroken());
		threadsleep(1100);
		System.out.println("已经超时异常了，Broken?" + cyclicBarrier.isBroken());
		awaitThread(6, cyclicBarrier);//barrier已经毁坏
	}
	
	public static void test3() {//reset异常
		cyclicBarrier.reset();
		awaitThread(7, cyclicBarrier);//test reset
		threadsleep(100);
		cyclicBarrier.reset();
	}
	
	public static void test4() {//action异常
		cyclicBarrier.reset();
		str = null;
		awaitThread(8, cyclicBarrier);
		awaitThread(9, cyclicBarrier);
		awaitThread(10, cyclicBarrier);//action由该线程执行，异常也由该线程抛出
	}
 
	public static void awaitThread(final int i, CyclicBarrier cyclicBarrier) {
		new Thread(() -> {
				try {
					cyclicBarrier.await();
					System.out.println("停止阻塞-" + i);
				} catch (InterruptedException e) {
					//当前线程等待时被中断抛出
					System.out.println("x被中断-" + i);
				} catch (BrokenBarrierException e) {
					//当前线程等待时其他线程中断或者超时
					//当barrier被reset时
					//调用await时barrier已经毁坏时
					//回调action抛出异常时
					System.out.println("xbarrier毁坏-" + i);
				}
		}).start();
	}
	
	public static void awaitTimeThread(final int i, CyclicBarrier cyclicBarrier, int awaitSeconds) {
		new Thread(() -> {
			try {
				cyclicBarrier.await(awaitSeconds, TimeUnit.SECONDS);//带超时时间
				System.out.println("停止超时阻塞-" + i);
			} catch (InterruptedException e) {
				System.out.println("x被中断=" + i);
			} catch (BrokenBarrierException e) {
				System.out.println("xbarrier毁坏=" + i);
			} catch (TimeoutException e) {
				//超时抛出异常
				System.out.println("x超时异常=" + i);
			}
		}).start();
	}
	
	public static void threadsleep(long millis) {
		try {
			Thread.sleep(millis);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}