15、详解java中的同步工具类CyclicBarrier

短命女 2023-05-31 07:30 2阅读 0赞

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

**一、概念理解**

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

就比如说我们在打王者的时候，十个人必须全部加载到100%，才可以开局。否则只要有一个人没有加载到100%,那这个游戏就不能开始。先加载完成的玩家必须等待最后一个玩家加载成功才可以。如果你实在记不住，你可以想象成人满发车的长途，就算你是第一个上车的人，也要等待车满才可以发车。否则车上所有人都要等待。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70]  
**与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方法。现在测试一遍。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70 1]  
现在相信通过这个案例，大家都能掌握其用法，很简单。下面我们从源码的角度来分析一下其实现原理。

**三、源码分析**

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

//构造方法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原理我们已经解释了，其用途可以根据自己的业务场景来决定了。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70]: /images/20230531/123f8bbbe4494a3a9c72e5cb69249ea0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70 1]: /images/20230531/12e6e3f7cc344dfcb82c31afc026ad2b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1NERERMTEw_size_16_color_FFFFFF_t_70 2]: /images/20230531/eb4cf8595f5046c2a5db175543721f0b.png