Java并发包同步辅助器CyclicBarrier

春花秋月何时了

浏览: 42428 次
性别:
来自: 成都

最近访客更多访客>>

lilin

xumiller

jerry_shen

sway

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

Java 线程

引言及简介

上一章我们介绍了独占式的同步组件ReentrantLock，今天我们先不忙着继续介绍其他Lock接口的实现类同步组件，先来看看JDK基于ReentrantLock同步组件以及Condition条件等待机制实现的一个同步辅助器CyclicBarrier。

CyclicBarrier是一个同步辅助工具，它允许一组线程相互等待达到一个公共的屏障(barrier)点。CyclicBarrier的字面意思可以叫做可循环(Cyclic)使用的屏障(Barrier)，之所以说它是可循环(Cyclic)使用的屏障，是因为当所有相互等待的线程都释放之后(或者说都通过屏障之后)，它是可以被重复使用的。另外，CyclicBarrier还可以让这一组相互等待的线程在都到达这个公共的屏障(barrier)之后，让最晚到达屏障的线程先去完成一件特殊指定的任务，在这项指定的任务完成之后，再让这一组线程接着往下执行(可能已经执行完成就意味着结束)，包括最晚到达屏障的那一个线程也会继续往下执行自己本来的逻辑。

可能通过这样的描述，我们很难理解CyclicBarrier到底有什么作用，以我的理解可以简单的描述为：使用CyclicBarrier可以使一组线程在至少有其中一个线程没有就位之前都让已经就位的线程陷入等待，直到所有的线程都就位之后，大家再一起继续往下执行(如果已经执行完就意味着结束)。所谓的“就位”就相当于是一个屏障或者一扇门，只有当所有的参与者都来到这扇大门前，门才会开启让大家一起通过，在大家通过之后，这扇门又会关闭复原，使其能够被重复用于进行“就位”等待放行使用。作为可选的特殊任务就是在大家都就位之后，在门打开之前的这一个间隙之间，可以让最晚就位的参与者临时去完成一件特殊任务(谁让你来的最晚呢)，在这件任务完成之后，大家再一起通过这扇门，然后该干嘛干嘛。

上图是我理解的CyclicBarrier，线程A，B，C花费不同的时间之后达到一个公共的屏障之后，再一起继续往后执行(当然也可以是后面没有逻辑需要执行而结束)，在线程B达到之前，线程C已经等待了5秒，线程A已经等待了3秒，如果有指定一个在屏障点的特殊任务，那么在线程A,B,C通过屏障继续往下执行之前，将由线程B(因为它最晚到达)去完成该特殊任务。

使用示例

通过上面的简介，我相信应该对CyclicBarrier的作用有了了解，下面我们就使用几个示例来加深对其了解。在使用之前，我们先要了解一下CyclicBarrier的构造方法，它有两个构造方法：

public CyclicBarrier(int parties, Runnable barrierAction) { //接受一个在屏障点的特殊任务barrierAction
	if (parties <= 0) throw new IllegalArgumentException();
	this.parties = parties;
	this.count = parties;
	this.barrierCommand = barrierAction;
}
//直接构造一个CyclicBarrier，它将使指定个数的参与者(线程)都到达屏障点之前相互等待
public CyclicBarrier(int parties) {
	this(parties, null);
}

通过构造方法我们可以发现，在屏障点的特殊任务是可选的，该任务是一个实现了Runnable接口的实例，整形参数parties表示让多少个线程相互等待至屏障点。

示例一：假若有若干个线程都要进行写数据操作，并且只有所有线程都完成写数据操作之后，这些线程才能继续做后面的事情

public static void main(String[] args) {  
	int N = 4;  
	CyclicBarrier barrier  = new CyclicBarrier(N, new Runnable(){  
  
		@Override  
		public void run() {  
			System.out.println("由线程"+Thread.currentThread().getName()+"开始执行屏障点特殊任务");    
			try {  
				Thread.sleep(2000);  
			} catch (InterruptedException e) {  
				e.printStackTrace();  
			}   
		}  
		  
	});  
	for(int i=0;i<N;i++)  
		new Writer(barrier).start();  
}  
static class Writer extends Thread{  
	private CyclicBarrier cyclicBarrier;  
	public Writer(CyclicBarrier cyclicBarrier) {  
		this.cyclicBarrier = cyclicBarrier;  
	}  
  
	@Override  
	public void run() {  
		System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");  
		try {  
			Thread.sleep((long)(Math.random() * 10000));      //以睡眠来模拟写入数据操作  
			System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");  
			int index = cyclicBarrier.await();  
			System.out.println("所有线程写入完毕，当前线程"+Thread.currentThread().getName()+"(await返回："+index+")继续处理其他任务...");  
		} catch (InterruptedException e) {  
			e.printStackTrace();  
		}catch(BrokenBarrierException e){  
			e.printStackTrace();  
		}  
	}  
}

执行结果：

线程Thread-2正在写入数据...
线程Thread-1正在写入数据...
线程Thread-3正在写入数据...
线程Thread-0正在写入数据...
线程Thread-0写入数据完毕，等待其他线程写入完毕
线程Thread-2写入数据完毕，等待其他线程写入完毕
线程Thread-3写入数据完毕，等待其他线程写入完毕
线程Thread-1写入数据完毕，等待其他线程写入完毕
由线程Thread-1开始执行屏障点特殊任务
所有线程写入完毕，当前线程Thread-1(await返回：0)继续处理其他任务...
所有线程写入完毕，当前线程Thread-0(await返回：3)继续处理其他任务...
所有线程写入完毕，当前线程Thread-2(await返回：2)继续处理其他任务...
所有线程写入完毕，当前线程Thread-3(await返回：1)继续处理其他任务...

通过示例一可以看到，我们构造了四个相互等待的线程，在线程Thread-1没有写入数据完毕之前，其它三个最快的线程一直等待，当四个线程都到达屏障点时，让最晚执行完的线程Thread-1执行了屏障点的特殊任务，特殊任务执行完之后，四个线程再一起继续往后执行。并且最早达到的线程Thread-0的await()的返回值为3，往后依次递减，最晚达到屏障点的线程Thread-1的await()的返回值为0，

示例二：继续改造示例一，演示 CyclicBarrier的可重用性。

public static void main(String[] args) {
	int N = 4;
	CyclicBarrier barrier  = new CyclicBarrier(N,new Runnable() {
		@Override
		public void run() {
			System.out.println("由线程"+Thread.currentThread().getName()+"开始执行屏障点特殊任务");
			try {
				Thread.sleep(2000); 
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	});
	 
	for(int i=0;i<N;i++)
		new Writer(barrier).start();
}
static class Writer extends Thread{
	private CyclicBarrier cyclicBarrier;
	public Writer(CyclicBarrier cyclicBarrier) {
		this.cyclicBarrier = cyclicBarrier;
	}

	@Override
	public void run() {
		int count = 2;
		try {
			while (count > 0) {
				count--;
				System.out.println("线程"+Thread.currentThread().getName()+"第"+(2-count)+"次正在写入数据...");
				Thread.sleep(5000);      //以睡眠来模拟写入数据操作
				System.out.println("线程"+Thread.currentThread().getName()+"第"+(2-count)+"写入数据完毕，等待其他线程写入完毕");
				cyclicBarrier.await();
			}
			System.out.println("所有线程写入完毕，当前线程"+Thread.currentThread().getName()+"继续处理其他任务...");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}catch(BrokenBarrierException e){
			e.printStackTrace();
		}
	}
}

执行结果：

线程Thread-0第1次正在写入数据...
线程Thread-2第1次正在写入数据...
线程Thread-3第1次正在写入数据...
线程Thread-1第1次正在写入数据...
线程Thread-3第1写入数据完毕，等待其他线程写入完毕
线程Thread-1第1写入数据完毕，等待其他线程写入完毕
线程Thread-0第1写入数据完毕，等待其他线程写入完毕
线程Thread-2第1写入数据完毕，等待其他线程写入完毕
由线程Thread-2开始执行屏障点特殊任务
线程Thread-2第2次正在写入数据...
线程Thread-0第2次正在写入数据...
线程Thread-1第2次正在写入数据...
线程Thread-3第2次正在写入数据...
线程Thread-3第2写入数据完毕，等待其他线程写入完毕
线程Thread-2第2写入数据完毕，等待其他线程写入完毕
线程Thread-1第2写入数据完毕，等待其他线程写入完毕
线程Thread-0第2写入数据完毕，等待其他线程写入完毕
由线程Thread-0开始执行屏障点特殊任务
所有线程写入完毕，当前线程Thread-0继续处理其他任务...
所有线程写入完毕，当前线程Thread-3继续处理其他任务...
所有线程写入完毕，当前线程Thread-2继续处理其他任务...
所有线程写入完毕，当前线程Thread-1继续处理其他任务...

在示例二中，我让这四个线程循环执行两次模拟数据写入操作，所以他们会遇到两次CyclicBarrier设定的屏障，再他们第一次通过屏障之后，屏障将会被复原，所以第二次遇到屏障的时候，依然会产生和第一次相同的效果，当然这四个线程具体的执行顺序几乎不会和第一次相同了，执行屏障点的特殊任务的线程还是由最晚达到屏障点的线程去完成，第一次是线程Thread-2，第二次是线程Thread-0.

通过以上的两个示例我们只能基本的了解CyclicBarrier循环屏障的简单使用，至于其深层次的其他特性，比如：万一存在线程在未到达屏障之前出现了未被捕获的异常，或者线程在执行屏障点的特殊任务的时候抛出了异常，那么又会对其他线程的执行有什么影响呢？这些问题我们稍后再来回答，虽然我们可以通过继续举例来得到验证，但是我还是想先看看CyclicBarrier的源码之后再进行说明，毕竟举例不可能把所有的情况都包含，通过源码分析才能得到最直接最有力的结论。

源码分析

首先我们看看CyclicBarrier的成员属性：

public class CyclicBarrier {
	private static class Generation {
		boolean broken = false;
	}
        //锁对象
	private final ReentrantLock lock = new ReentrantLock();
	private final Condition trip = lock.newCondition();
	/** 参与者个数，构造方法参数之一 */
	private final int parties;
	/* 屏障点特殊任务，构造方法参数之一 */
	private final Runnable barrierCommand;
	//用于表示屏障状态
	private Generation generation = new Generation();
        //表示还未达到屏障点的线程个数，为0表示所有线程都达到了屏障点
	private int count;
	....
}

通过以上对成员属性的分析，可以发现，CyclicBarrier确实是基于ReentrantLock同步组件以及Condition条件等待机制实现的，另外，CyclicBarrier还存在一个私有的静态内部类Generation，它只有一个布尔型的broken属性，用于表示当前屏障是否被破坏，在默认情况下或者所有参与者都通过屏障之后屏障都是没有被破坏的。Generation其实也是实现CyclicBarrier能够被重复使用的关键。

接着分析CyclicBarrier最核心的方法，通过上面的示例可以发现，CyclicBarrier最核心的关系方法就是await()方法，所以我们直接看这个方法就可以了，当然CyclicBarrier内部还提供了带有超时时间的await()方法。

public int await() throws InterruptedException, BrokenBarrierException {
	try {
		return dowait(false, 0L);//第一个参数为false.超时时间为0，表示没有超时设定
	} catch (TimeoutException toe) {
		throw new Error(toe); // 因为dowait()的时间参数为0， 所以这里永远不会抛出TimeoutException。
	}
}
//带有超时时间的await()方法
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
	return dowait(true, unit.toNanos(timeout)); //因为dowait()的时间参数由使用者传入，所以有可能会抛出TimeoutException异常}

从表面上可以看到await()是可以抛出中断异常和BrokenBarrierException异常的，甚至await(long, TimeUnit)方法还可能抛出TimeoutException超时异常，他们都是调用的dowait()方法，其返回值是一个由dowait()方法返回的整形数值，通过Java Doc的描述，await()/dowait()方法的返回值代表当前线程到达屏障点的序列号，假设有5个线程，最先到达屏障点的线程该方法返回值的就是4，往后依次递减，最后一个达到屏障点的线程调用该方法的返回值就是0.

接下来，我们重点分析dowait()方法。

private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
	final ReentrantLock lock = this.lock;
	lock.lock(); //通过ReentrantLock独占锁加锁，在finally语句中解锁
	try {
		final Generation g = generation;
                //如果当前屏障已经被损坏，直接抛出BrokenBarrierException异常
		if (g.broken)
			throw new BrokenBarrierException();
                //如果当前线程被设置了中断标记，先将当前屏障标记为“已损坏”，再唤醒所有在等待的线程
		if (Thread.interrupted()) { //中断状态将被复位清空
			breakBarrier();
			throw new InterruptedException();//同时还有抛出中断异常
		}
                //如果一切正常，计数器减1(因为已经上锁所以没有线程安全问题)
		int index = --count;
		if (index == 0) {  //计数器==0，表示所有线程都已经“就位”
			boolean ranAction = false;
			try {
				final Runnable command = barrierCommand;
				if (command != null)  //如果有特殊任务，执行其run方法
					command.run();//这里如果出现异常也就直接以异常形式返回了
				ranAction = true;
				nextGeneration(); //如果没有特殊任务或者特殊任务执行成功，先唤醒所有等待线程，再重置屏障状态
				return 0;//返回 0
			} finally {
				if (!ranAction) //如果有特殊任务并执行失败，先标记屏障“已损坏”，再唤醒所有等待线程
					breakBarrier();
			}
		}
                //走到这里，说明一切正常，并且还有线程没有“就位”
		for (;;) { // 注意这里是一个"自旋"
			try {
                                //根据调用者是否有指定超时时间，分别调用不同的await()方法，使当前线程陷入阻塞等待，并会释放锁
                                //直到被唤醒、被中断、或者带有超时时间的超时，才会退出等待
				if (!timed)
					trip.await();
				else if (nanos > 0L)
					nanos = trip.awaitNanos(nanos);
			} catch (InterruptedException ie) {
                               //如果是被中断唤醒
                               //检测最新的屏障状态，如果屏障依然完好，就损坏屏障再唤醒其他等待线程，还有抛出中断异常。
				if (g == generation && ! g.broken) {
					breakBarrier();
					throw ie;//这里直接以异常形式返回了
				} else {
                               //如果发现最新的屏障状态为“已损坏”，进行自我中断
					Thread.currentThread().interrupt();
				}
			}
                        //走到这里，说明是被其他线程唤醒，或者超时自动唤醒，或者中断唤醒但是发现屏障已经被损坏
			if (g.broken) //发现屏障损坏立即抛出BrokenBarrierException异常
				throw new BrokenBarrierException();

			if (g != generation)//发现屏障被更新了，返回index
				return index;

			if (timed && nanos <= 0L) { //如果是带有超时时间的等待被唤醒，发现已经没有时间剩下了
				breakBarrier(); //先损坏屏障，再唤醒所有等待线程
				throw new TimeoutException(); //抛出超时异常
			}
		}
	} finally {
		lock.unlock();
	}
}

//先设置屏障状态为“损坏”，再唤醒所有已经在等待的线程
private void breakBarrier() {
	generation.broken = true;
	count = parties;
	trip.signalAll();
}

//先唤醒所有等待的线程，再生成新的可用的屏障
private void nextGeneration() {
	trip.signalAll();
	count = parties;
	generation = new Generation();
}

通过以上dowait()的源码分析，我们可以大致梳理其逻辑如下：

在进入await()方法之后，如果发现屏障已被破坏，抛出BrokenBarrierException异常，结束。
在进入await()方法之后，如果屏障没有被破坏，但是当前线程被中断，则先破坏屏障，再唤醒所有线程，最后抛出中断异常，结束。
如果不是最后一个线程到达屏障点，就通过Condition对象的await/awaitNanos方法将当前线程阻塞。index记录了当前线程到达屏障点的序列号，最先达到的序号为总线程数减1，往后的线程依次减1
如果是最后一个线程到达屏障点，根据是否事先指定了屏障点的特殊任务做不同的事情，
如果有指定屏障点任务，则先执行屏障点任务，根据屏障点任务执行时是否抛出异常执行不同的逻辑。
如果在执行屏障点任务时抛出异常，则先损坏屏障，再唤醒所有等待在屏障点的其他线程。然后再将异常抛出，结束
如果在执行屏障点任务时没有异常抛出，则先唤醒所有等待在屏障点的其他线程，然后重新生成新的屏障。当前线程返回0，结束。
如果没有指定屏障点任务，那么也走执行屏障点任务没有抛出异常之后一样的逻辑：先唤醒所有等待在屏障点的其他线程，然后重新生成新的屏障。当前线程返回0，结束。

其他线程被唤醒之后的逻辑梳理如下：

如果是被中断唤醒，并且屏障没有破坏，就先损坏屏障再唤醒其他线程，抛出中断异常，结束。
如果是被中断唤醒，并且屏障已经破坏，先重新标记中断状态，然后抛出BrokenBarrierException异常，结束。
如果是被其他线程正常唤醒，并且屏障已经破坏，立即抛出BrokenBarrierException异常，结束。
如果是被其他线程正常唤醒，并且屏障没有破坏，但是屏障被更新了，返回之前记录的到达屏障点的序列号index，结束。
如果是被超时自动唤醒，并且屏障已经破坏，立即抛出BrokenBarrierException异常，结束。
如果是被超时自动唤醒，并且屏障没有破坏，但是屏障被更新了，返回之前记录的到达屏障点的序列号index，结束。
如果是被超时自动唤醒，并且屏障没有破坏，屏障也没有被更新，先破坏屏障在唤醒其他线程，抛出TimeoutException异常，结束。
被其他非中断情况唤醒，并且屏障没有破坏，屏障也没有被更新，如果设定了超时时间但超时时间也没到，通过自旋继续阻塞。

通过dowait()还可以发现对BrokenBarrierException异常的处理是优先于中断异常的，也就是说在既发生中断又发生了屏障损坏的时候，肯定是抛出BrokenBarrierException异常而不是中断异常，而中断只会被重新标记。那么除了在上面的逻辑中因为①屏障点任务抛出异常，②线程被中断唤醒，③线程被超时唤醒，这三种情况下，会自动破坏屏障外，其他线程有没有办法通过其它手段破坏屏障呢？答案是：有。那就是CyclicBarrier提供的rest()方法：

public void reset() {
	final ReentrantLock lock = this.lock;
	lock.lock();
	try {
		breakBarrier();   // 先破坏屏障，再唤醒所有等待线程
		nextGeneration(); // 先唤醒所有等待线程，再生成新的屏障
	} finally {
		lock.unlock();
	}
}

因为rest()方法是一个public的方法，所有任何线程在拿到CyclicBarrier的实例之后都可以在外部破坏屏障，唤醒所有等待的线程，然后生成新的屏障。虽然rest()方法中破坏屏障与生成新的屏障这几个操作在一个同步锁的块结构中，但是由于生成新的屏障时直接修改了屏障状态generation属性指向的内存地址，而原来旧的generation指向的内存地址表示的状态被修改为了“损坏”状态，所以对于那些已经处于等待状态的线程被rest()唤醒之后，由于他们在阻塞之前已经把generation作为临时变量存储于方法栈了，所以他们看到的将是被损坏的屏障而不是新的屏障，因此它们将会抛出BrokenBarrierException异常。

内存可见性

通过 CyclicBarrier的源码分析，await方法内部(不包括最后到达屏障的线程)主要是通过：lock->condition.await->unlock的模式运作，先到达屏障的线程在lock之后陷入condition.await，只有最后一个到达的线程执行逻辑是：lock->屏障点特殊任务->唤醒所有等待线程->unlock。最后一个到达的线程唤醒其他所有陷入condition.await阻塞的线程之后，那些被唤醒的线程需要依次重新获取锁之后才能执行unock。其中lock和unlock操作其实是对声明在AQS中volatile修饰的state变量的读取和修改操作，根据happens-before的锁定规则“一个unlock操作先行发生于后面对同一个锁的lock操作”，以及volatile变量规则“对一个volatile变量的写操作先行发生于后面对这个变量的读操作”和传递性可以得出如下的内存可见性：

所有线程在执行CyclicBarrier的await方法之前的操作happens-before屏障点特殊任务中的操作，屏障点特殊任务中的操作又happens-before所有那些从CyclicBarrier的await方法返回之后的操作。也就是说，所有线程在执行CyclicBarrier的await方法之前对共享变量的修改对执行屏障点特殊任务时是立即可见的，而在屏障点特殊任务中对共享变量的修改也对那些线程从CyclicBarrier的await方法返回之后的操作立即可见的。当然，所有线程在执行CyclicBarrier的await方法之前对共享变量的修改对所有线程在从CyclicBarrier的await方法返回之后的操作也是立即可见的。

CyclicBarrier运行情况总结

通过对源码的分析，很多未知的疑团终于有了答案，现将问答一一列举，在使用的时候了解了这些细节才能放心使用。

发生的情况(其它没有指明的情况默认没有发生)	对await()方法产生的结果
最后一个线程到达屏障，并成功执行屏障点任务( 如果有的话)	所有线程返回到达屏障点的序列号，继续往下执行。
最后一个线程到达屏障后，执行屏障点任务抛出异常	该线程将屏障点任务的异常从await()方法抛出；其他线程立即抛出BrokenBarrierException异常。
某个线程在即将进入await()方法之前被中断	该线程在执行await()方法时立即抛出中断异常；排在后面执行await()的线程在执行await()时和已经处于等待中的线程将立即抛出BrokenBarrierException异常。
中断某个处于等待中的某个线程	被中断的线程立即抛出InterruptedException异常；排在后面执行await()的线程在执行await()时和已经处于等待中的线程将立即抛出BrokenBarrierException异常。
某个等待线程超时，并且其他线程都处于等待	该线程直接返回到达屏障点的序列号，其他线程继续等待
某个等待线程超时，并且还有线程未到达屏障	该线程抛出TimeoutException异常，排在后面执行await()的线程在执行await()时和已经处于等待中的线程将立即抛出BrokenBarrierException异常。
某个线程执行了负值的await()方法	该线程抛出TimeoutException异常，排在后面执行await()的线程在执行await()时和已经处于等待中的线程将立即抛出BrokenBarrierException异常。
rest()方法被执行	在调用rest()之前已经处于等待状态的线程将立即抛出BrokenBarrierException异常；而在调用rest()方法之后调用await()的线程将陷入永久的等待，除非它们中至少存在一个带有超时时间。
某个线程在进入await()方法之前发生了异常，导致await()没有被执行。	那些已经处于等待状态的线程和排在后面执行了await()的线程都将陷入永久的等待，除非它们中至少存在一个带有超时时间。

查看图片附件

分享到：

Java并发包计数信号量Semaphore | Java并发包同步组件ReentrantLock

2018-04-30 16:35
浏览 484
评论(0)
分类:编程语言
查看更多

发表评论

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

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论