java多线程之CyclicBarrier源码解析

发表于

前言

本篇将分析CyclicBarrier的源码，分析结束后，会用一个示例展示CyclicBarrier，并比较CyclicBarrier和CountDownLatch的区别。

1、CyclicBarrier的简介

CyclicBarrier允许一组线程在触发屏障之前相互等待，直到达到某一条件才继续执行；CyclicBarrier在这些线程释放后，又可以重新使用，所以也称循环栅栏。

2、分析源码

2.1、构造方法

1
2
3
4
5
6
7
8
9
10
11
java复制代码//定义在触发屏障之前必须调用的线程数
public CyclicBarrier(int parties) {
	this(parties, null);
}
public CyclicBarrier(int parties, Runnable barrierAction) {
	if (parties <= 0) throw new IllegalArgumentException();
	    this.parties = parties;
	    this.count = parties;
	    //定义最后一个线程到达屏障时执行的操作
	    this.barrierCommand = barrierAction;
}

2.2、await()方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
java复制代码public int await() throws InterruptedException, BrokenBarrierException {
	try {
		//调用dowait方法，不需要定义超时时间
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}

private int dowait(boolean timed, long nanos)
	throws InterruptedException, BrokenBarrierException,
		   TimeoutException {
	final ReentrantLock lock = this.lock;
	//获取“独占锁”
	lock.lock();
	try {
		//获取“当前代”
		final Generation g = generation;
		//“当前代”损坏，则抛出异常
		if (g.broken)
			throw new BrokenBarrierException();
		// 如果当前线程被中断
		if (Thread.interrupted()) {
			//终止CyclicBarrier，唤醒CyclicBarrier中所有等待线程
			breakBarrier();
			throw new InterruptedException();
		}
		//计数
		int index = --count;
		//达到屏障（最后一个线程到达）
		//最后一个到达的线程不执行下面的for循环语句
		if (index == 0) {  // tripped
			boolean ranAction = false;
			try {
				final Runnable command = barrierCommand;
				if (command != null)
					//执行初始化传入的命令操作
					command.run();
				ranAction = true;
				//调用下一代方法
				nextGeneration();
				return 0;
			} finally {
				if (!ranAction)
					breakBarrier();
			}
		}

		// loop until tripped, broken, interrupted, or timed out
		//非最后一个到达的线程全部执行此语句，阻塞在trip.await()方法上
		for (;;) {
			try {
				//如果不是“超时等待”
				if (!timed)
					// 调用condition的await()方法
					trip.await();
				else if (nanos > 0L)
					nanos = trip.awaitNanos(nanos);
			} catch (InterruptedException ie) {
				// 如果等待过程中，线程被中断，则执行下面的函数
				if (g == generation && ! g.broken) {
					breakBarrier();
					throw ie;
				} else {
					// We're about to finish waiting even if we had not
					// been interrupted, so this interrupt is deemed to
					// "belong" to subsequent execution.
					Thread.currentThread().interrupt();
				}
			}

			if (g.broken)
				throw new BrokenBarrierException();
			// 如果“generation换代了”，则返回index。
			if (g != generation)
				return index;
			//检查超时
			if (timed && nanos <= 0L) {
				// 如果超时，则终止CyclicBarrier，并唤醒CyclicBarrier中所有等待线程
				breakBarrier();
				throw new TimeoutException();
			}
		}
	} finally {
		//释放“独占锁”，并唤醒AQS中等待的下一个线程
		lock.unlock();
	}
}

private void nextGeneration() {
    // 调用condition的signalAll()将其队列中的等待者全部转移到AQS的队列中
    trip.signalAll();
    // 重置count
    count = parties;
    // 进入下一代
    generation = new Generation();
}

3、CyclicBarrier与CountDownLatch的区别

  1. 两者都能实现阻塞一组线程，然后等待唤醒；
  2. 前者是最后一个线程到达直接唤醒，后者是调用countDown()方法；
  3. 前者是通过ReentrantLock的”独占锁”和Conditon来实现，后者是通过AQS的“共享锁”实现；
  4. 前者可以重复使用，后者只能使用一次；
  5. 前者只能实现多个线程到达后一起运行（多个条件成立才能一起运行）；
  6. 后者不仅可以实现一个线程等待多个线程（多个条件成立才能一起运行），还能实现多个线程等待一个线程（多个条件成立并且等待某个特殊信号才能一起运行）

4、示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
java复制代码public class CyclicBarrierTest {

    public static final CyclicBarrier WORK_THREAD = new CyclicBarrier(3);

    public static void main(String[] args) throws Exception{
        //主线程逻辑
        Thread.sleep(2000);
        for (int i=0; i<3; i++){
            String condition_name = "条件"+i;
            new Thread(() -> { method(); },condition_name).start();
        }
    }

    public static void method(){
        System.out.println("等待的条件是：" + Thread.currentThread().getName());
        try {
            WORK_THREAD.await();
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println("执行完的条件是：" + Thread.currentThread().getName());
    }
}

输出结果：

1
2
3
4
5
6
java复制代码等待的条件是：条件0
等待的条件是：条件1
等待的条件是：条件2
执行完的条件是：条件1
执行完的条件是：条件2
执行完的条件是：条件0

结束语

本篇介绍了CyclicBarrier，分析了源码，用了一段代码演示了使用方法，并比较了CyclicBarrier和CountDownLatch的区别。

JUC锁篇章到此就分析完了，下一篇将开启分析JUC集合。

如果你觉得本篇文章对你有帮助的话，请帮忙点个赞，再加一个关注。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%