详解java中的同步工具类CyclicBarrier

简介: 之前介绍了java中另一个同步工具类CountDownLatch,这篇文章主要介绍CyclicBarrier。

一、概念理解


CyclicBarrier允许一组线程在到达某个栅栏点(common barrier point)互相等待,直到最后一个线程到达栅栏点,栅栏才会打开,处于阻塞状态的线程恢复继续执行。


就比如说我们在打王者的时候,十个人必须全部加载到100%,才可以开局。否则只要有一个人没有加载到100%,那这个游戏就不能开始。先加载完成的玩家必须等待最后一个玩家加载成功才可以。如果你实在记不住,你可以想象成人满发车的长途,就算你是第一个上车的人,也要等待车满才可以发车。否则车上所有人都要等待。


与CountDownLatch的区别就是是否相互等待。举一个例子,CountDownLatch就好比是马拉松比赛,跑完的人不用等待其他选手是否结束,而CyclicBarrier需要等最后一个玩家加载结束。这就是区别。


我们直接代码演示一下这个例子。


二、代码演示


在这里我们同样使用的是打王者的例子。

首先我们定义main线程:

public class CyclicBarrierTest {
    public static void main(String[] args) {
        //第一步:定义玩家,这里写了5个
        String[] heros  = {"孙悟空","猪八戒","狄仁杰","鲁班","甄姬"};
        //第二步:使用线程池来运行,也是5个。
        ExecutorService service = Executors.newFixedThreadPool(5);
        //第三步:常见围栏,也是5个
        final CyclicBarrier barrier = new CyclicBarrier(10);
        //第四步:通过for循环传递给每一个玩家和围栏
        for (int i = 0; i < 10; i++) {
            service.execute(new Player(heros[i], barrier));
        }
        service.shutdown();
    }
}

上面的代码我们已经解释清楚了,主要是通过线程池运行玩家,并传递给每一个玩家名字和围栏。下面我们就看看这个Player玩家线程如何实现的。

public class Player implements Runnable {
    private final String hero;
    private final CyclicBarrier barrier;
    public Player(String hero, CyclicBarrier barrier) {
        this.hero = hero;
        this.barrier = barrier;
    }
    @Override
    public void run() {
        try {
            //每一个英雄加载成功的时间不一样,所以这里用了随机数
            TimeUnit.SECONDS.sleep(1 + (new Random().nextInt(5)));
            System.out.println(hero + "开始加载==========等待其他玩家加载成功");
            barrier.await();
            System.out.println(hero + ":看到所有玩家加载成功,比赛开始");
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    }
}

在这个player线程中,我们使用随机数来表示每个玩家不同的加载时间,在休眠时间结束之前,player一直处于等待的状态,也就是调用了await方法。现在测试一遍。

v2-ff7828422355d1be885391c8c1d189ba_1440w.jpg

现在相信通过这个案例,大家都能掌握其用法,很简单。下面我们从源码的角度来分析一下其实现原理。


三、源码分析


为了分析的彻底,我们先从构造方法开始:

//构造方法1: parties主要是需要拦截的线程数
public CyclicBarrier(int parties) {
    this(parties, null);
}
//构造方法2:不仅有parties还有barrierAction
//主要是为了处理更加复杂的场景,当线程到达围栏时候
//优先执行barrierAction
public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

以上就是两个构造方法,下面我们主要分析await方法我们进入源码看看:

//不带超时时间
    public int await() throws InterruptedException, 
                            BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }
    //带有超时时间
    public int await(long timeout, TimeUnit unit)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        return dowait(true, unit.toNanos(timeout));
    }

返回的值是加载成功的玩家数量,既然内部是通过dowait方法实现的,不如我们再跟进去看看。

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()) {
                breakBarrier();
                throw new InterruptedException();
            }
            int index = --count;
            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();
                }
            }
            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();
        }
    }

上面的代码有点长,不过也是CyclicBarrier的核心,我在这里说一下上面代码的主要功能:


(1)通过ReentrantLock获取独占锁。

(2)通过try里面的Generation判断当前代是否损坏, 通过Thread的interrupted方法判断是否线程中断,如果中断通过breakBarrier方法告诉其他线程。

(3)if(index==0)判断当前是否是最后一个线程调用了await方法,如果是则把之前等待的线程全部唤醒。就好比是最后一个运动员到达了终点,告诉其他选手比赛结束了。

(4)for(;;)循环执行等待,如果没有超时时间,那就一直等待直到被唤醒,有超时时间,那就等时间过后自动被唤醒。就好比是在运动员在路上跑步,没有时间限制的时候那就一直跑,一直到达终点。如果有时间限制,不管是否跑到终点,比赛都结束。

(5)通过ReentrantLock释放独占锁。

这段代码读起来比较麻烦,因为里面涉及到了两个其他的类ReentrantLock和Generation,我们只需要知道其作用即可。还有一个点,那就是线程被中断了,如何告诉其他线程的breakBarrier方法。

/**
     * Sets current barrier generation as broken and wakes up everyone.
     * Called only while holding lock.
     */
    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }

我们可以看到,在breakBarrier()中除了将broken设置为true,还会调用signalAll将在CyclicBarrier处于等待状态的线程全部唤醒。


OK,今天的文章就先到这,关于CyclicBarrier原理我们已经解释了,其用途可以根据自己的业务场景来决定了。

相关文章
|
3天前
|
Java 关系型数据库 MySQL
Elasticsearch【问题记录 01】启动服务&停止服务的2类方法【及 java.nio.file.AccessDeniedException: xx/pid 问题解决】(含shell脚本文件)
【4月更文挑战第12天】Elasticsearch【问题记录 01】启动服务&停止服务的2类方法【及 java.nio.file.AccessDeniedException: xx/pid 问题解决】(含shell脚本文件)
28 3
|
1天前
|
安全 Java 程序员
|
2天前
|
Java
Java Class类
Java Class类
8 0
|
6天前
|
安全 Java
深入理解 Java 多线程和并发工具类
【4月更文挑战第19天】本文探讨了Java多线程和并发工具类在实现高性能应用程序中的关键作用。通过继承`Thread`或实现`Runnable`创建线程,利用`Executors`管理线程池,以及使用`Semaphore`、`CountDownLatch`和`CyclicBarrier`进行线程同步。保证线程安全、实现线程协作和性能调优(如设置线程池大小、避免不必要同步)是重要环节。理解并恰当运用这些工具能提升程序效率和可靠性。
|
8天前
|
Java 编译器
Java Character 类
4月更文挑战第13天
|
8天前
|
存储 缓存 安全
Java并发基础之互斥同步、非阻塞同步、指令重排与volatile
在Java中,多线程编程常常涉及到共享数据的访问,这时候就需要考虑线程安全问题。Java提供了多种机制来实现线程安全,其中包括互斥同步(Mutex Synchronization)、非阻塞同步(Non-blocking Synchronization)、以及volatile关键字等。 互斥同步(Mutex Synchronization) 互斥同步是一种基本的同步手段,它要求在任何时刻,只有一个线程可以执行某个方法或某个代码块,其他线程必须等待。Java中的synchronized关键字就是实现互斥同步的常用手段。当一个线程进入一个synchronized方法或代码块时,它需要先获得锁,如果
24 0
|
9天前
|
存储 Java
Java基础教程(7)-Java中的面向对象和类
【4月更文挑战第7天】Java是面向对象编程(OOP)语言,强调将事务抽象成对象。面向对象与面向过程的区别在于,前者通过对象间的交互解决问题,后者按步骤顺序执行。类是对象的模板,对象是类的实例。创建类使用`class`关键字,对象通过`new`运算符动态分配内存。方法包括构造函数和一般方法,构造函数用于对象初始化,一般方法处理逻辑。方法可以有0个或多个参数,可变参数用`类型...`定义。`this`关键字用于访问当前对象的属性。
|
13天前
|
Java Shell
Java 21颠覆传统:未命名类与实例Main方法的编码变革
Java 21颠覆传统:未命名类与实例Main方法的编码变革
13 0
|
13天前
|
Java
Java 15 神秘登场:隐藏类解析未知领域
Java 15 神秘登场:隐藏类解析未知领域
16 0
|
14天前
|
安全 Java
append在Java中是哪个类下的方法
append在Java中是哪个类下的方法
23 9