Java并发-CyclicBarrier

发表于 | 评论数： | 热度： ℃   |   字数统计: 2.3k字   |   阅读时长: ≈ 9分钟

概述

现实生活中经常会遇到这样的情景，在进行某个活动前需要等待人全部都齐了才开始。例如吃饭时要等全家人都上座了才动筷子，旅游时要等全部人都到齐了才出发，比赛时要等运动员都上场后才开始。在J.U.C包中提供了一个同步工具类能够很好的模拟这类场景，它就是CyclicBarrier类。利用CyclicBarrier类可以实现一组线程相互等待，当所有线程都到达某个屏障点后再进行后续的操作。下图演示了这一过程。
img

成员变量

在CyclicBarrier类的内部有一个计数器，每个线程在到达屏障点的时候都会调用await方法将自己阻塞，此时计数器会减1，当计数器减为0的时候所有因调用await方法而被阻塞的线程将被唤醒。这就是实现一组线程相互等待的原理，下面看下CyclicBarrier有哪些成员变量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
//同步操作锁
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();
//计数器
private int count;

//静态内部类Generation
private static class Generation {
    boolean broken = false;
}

CyclicBarrier内部是通过条件队列trip来对线程进行阻塞的，并且其内部维护了两个int型的变量parties和count。
parties表示每次拦截的线程数，该值在构造时进行赋值。
count是内部计数器，它的初始值和parties相同，以后随着每次await方法的调用而减1，直到减为0就将所有线程唤醒。
CyclicBarrier有一个静态内部类Generation，该类的对象代表栅栏的当前代，就像玩游戏时代表的本局游戏，利用它可以实现循环等待。
barrierCommand表示换代前执行的任务，当count减为0时表示本局游戏结束，需要转到下一局。在转到下一局游戏之前会将所有阻塞的线程唤醒，在唤醒所有线程之前你可以通过指定barrierCommand来执行自己的任务。

构造器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
//构造器1
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
	// parties表示“必须同时到达barrier的线程个数”。
    this.parties = parties;
	// count表示“处在等待状态的线程个数”。
    this.count = parties;
	// barrierCommand表示“parties个线程到达barrier时，会执行的动作”。
    this.barrierCommand = barrierAction;
}

//构造器2
public CyclicBarrier(int parties) {
    this(parties, null);
}

CyclicBarrier有两个构造器，其中构造器1是它的核心构造器，在这里可以指定要拦截的线程数以及结束时要执行的任务，还可以看到计数器count的初始值被设置为parties。

等待方法

1
2
3
4
5
6
7
8
9
10
11
12
13
//非定时等待
public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe);
    }
}

//定时等待
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
    return dowait(true, unit.toNanos(timeout));
}

不管是定时等待还是非定时等待，它们都调用了dowait方法，只不过是传入的参数不同而已。下面看下dowait方法都做了些什么。

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
//核心等待方法
private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
    final ReentrantLock lock = this.lock;
    //获取独占锁
	lock.lock();
    try {
	    //保存当前的generation
        final Generation g = generation;
        //检查当前栅栏是否被打翻
        if (g.broken) {
            throw new BrokenBarrierException();
        }
        //检查当前线程是否被中断
        if (Thread.interrupted()) {
            //如果当前线程被中断会做以下三件事
            //1.打翻当前栅栏
            //2.唤醒拦截的所有线程
            //3.抛出中断异常
            breakBarrier();
            throw new InterruptedException();
        }
       //每次都将计数器的值减1
       int index = --count;
       //计数器的值减为0则需唤醒所有线程并转换到下一代
       if (index == 0) {
           boolean ranAction = false;
           try {
               //唤醒所有线程前先执行指定的任务
               final Runnable command = barrierCommand;
               if (command != null) {
                   command.run();
               }
               ranAction = true;
               //唤醒所有线程并转到下一代
               nextGeneration();
               return 0;
           } finally {
               //确保在任务未成功执行时能将所有线程唤醒
               if (!ranAction) {
                   breakBarrier();
               }
           }
       }

       //当前线程一直阻塞，直到“有parties个线程到达barrier” 或 “当前线程被中断” 或 “超时”这3者之一发生，当前线程才继续执行。
       for (;;) {
           try {
               //根据传入的参数来决定是定时等待还是非定时等待
               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) {
               throw new BrokenBarrierException();
           }
           //如果线程因为换代操作而被唤醒则返回计数器的值
           if (g != generation) {
               return index;
           }
           //如果线程因为时间到了而被唤醒则打翻栅栏并抛出异常
           if (timed && nanos <= 0L) {
               breakBarrier();
               throw new TimeoutException();
           }
       }
    } finally {
        lock.unlock();
    }
}

在dowait方法中每次都将count减1，减完后立马进行判断看看是否等于0，如果等于0的话就会先去执行之前指定好的任务，执行完之后再调用nextGeneration方法将栅栏转到下一代，在该方法中会将所有线程唤醒，将计数器的值重新设为parties，最后会重新设置栅栏代次，在执行完nextGeneration方法之后就意味着游戏进入下一局。如果计数器此时还不等于0的话就进入for循环，根据参数来决定是调用trip.awaitNanos(nanos)还是trip.await()方法，这两方法对应着定时和非定时等待。如果在等待过程中当前线程被中断就会执行breakBarrier方法，该方法叫做打破栅栏，意味着游戏在中途被掐断，设置generation的broken状态为true并唤醒所有线程。同时这也说明在等待过程中有一个线程被中断整盘游戏就结束，所有之前被阻塞的线程都会被唤醒。线程醒来后会执行下面三个判断，看看是否因为调用breakBarrier方法而被唤醒，如果是则抛出异常；看看是否是正常的换代操作而被唤醒，如果是则返回计数器的值；看看是否因为超时而被唤醒，如果是的话就调用breakBarrier打破栅栏并抛出异常。这里还需要注意的是，如果其中有一个线程因为等待超时而退出，那么整盘游戏也会结束，其他线程都会被唤醒。下面贴出nextGeneration方法和breakBarrier方法的具体代码。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
//切换栅栏到下一代
private void nextGeneration() {
    //唤醒条件队列所有线程
    trip.signalAll();
    //设置计数器的值为需要拦截的线程数
    count = parties;
    //重新设置栅栏代次
    generation = new Generation();
}

//打翻当前栅栏
private void breakBarrier() {
    //将当前栅栏状态设置为打翻
    generation.broken = true;
    //设置计数器的值为需要拦截的线程数
    count = parties;
    //唤醒所有线程
    trip.signalAll();
}

应用示例

赛马:通过在控制台不停的打印各赛马的当前轨迹，以达到动态显示的效果。整场比赛有多个轮次，每一轮次各个赛马都会随机走上几步然后调用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
class Horse implements Runnable {
    
    private static int counter = 0;
    private final int id = counter++;
    private int strides = 0;
    private static Random rand = new Random(47);
    private static CyclicBarrier barrier;
    
    public Horse(CyclicBarrier b) { barrier = b; }
    
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                synchronized(this) {
                    //赛马每次随机跑几步
                    strides += rand.nextInt(3);
                }
                barrier.await();
            }
        } catch(Exception e) {
            e.printStackTrace();
        }
    }
    
    public String tracks() {
        StringBuilder s = new StringBuilder();
        for(int i = 0; i < getStrides(); i++) {
            s.append("*");
        }
        s.append(id);
        return s.toString();
    }
    
    public synchronized int getStrides() { return strides; }
    public String toString() { return "Horse " + id + " "; }
    
}

public class HorseRace implements Runnable {
    
    private static final int FINISH_LINE = 75;
    private static List<Horse> horses = new ArrayList<Horse>();
    private static ExecutorService exec = Executors.newCachedThreadPool();
    
    @Override
    public void run() {
        StringBuilder s = new StringBuilder();
        //打印赛道边界
        for(int i = 0; i < FINISH_LINE; i++) {
            s.append("=");
        }
        System.out.println(s);
        //打印赛马轨迹
        for(Horse horse : horses) {
            System.out.println(horse.tracks());
        }
        //判断是否结束
        for(Horse horse : horses) {
            if(horse.getStrides() >= FINISH_LINE) {
                System.out.println(horse + "won!");
                exec.shutdownNow();
                return;
            }
        }
        //休息指定时间再到下一轮
        try {
            TimeUnit.MILLISECONDS.sleep(200);
        } catch(InterruptedException e) {
            System.out.println("barrier-action sleep interrupted");
        }
    }
    
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(7, new HorseRace());
        for(int i = 0; i < 7; i++) {
            Horse horse = new Horse(barrier);
            horses.add(horse);
            exec.execute(horse);
        }
    }
    
}

程序的运行结果如下：
img

本文标题:Java并发-CyclicBarrier

文章作者:王洪博

发布时间:2019年02月21日 - 17:02

最后更新:2019年09月12日 - 10:09

原始链接:http://whb1990.github.io/posts/3a683baa.html

▄︻┻═┳一如果你喜欢这篇文章，请点击下方"打赏"按钮请我喝杯 ☕
0%