公司建设网站申请,网站建设案例单招网,网页制作基础教程,黎城网站建设CyclicBarrier详解
CyclicBarrier
字面意思回环栅栏#xff08;循环屏障#xff09;#xff0c;通过它可以实现让一组线程等待至某个状态#xff08;屏障点#xff09;之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后#xff0c;CyclicBarrier可以被重用…CyclicBarrier详解
CyclicBarrier
字面意思回环栅栏循环屏障通过它可以实现让一组线程等待至某个状态屏障点之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后CyclicBarrier可以被重用 CyclicBarrier的使用 重要方法
// parties表示屏障拦截的线程数量每个线程调用 await 方法告诉 CyclicBarrier 我已经到达了屏障然后当前线程被阻塞。public CyclicBarrier(int parties)// 用于在线程到达屏障时优先执行 barrierAction方便处理更复杂的业务场景(该线程的 执行时机是在到达屏障之后再执行)public CyclicBarrier(int parties, Runnable barrierAction)//屏障 指定数量的线程全部调用await()方法时这些线程不再阻塞// BrokenBarrierException 表示栅栏已经被破坏破坏的原因可能是其中一个线程 await() 时被中断或者超时public int await() throws InterruptedException, BrokenBarrierExceptionpublic int await(long timeout, TimeUnit unit) throws InterruptedException, Bro kenBarrierException, TimeoutException//循环 通过reset()方法可以进行重置public void reset()案例
Slf4j
public class CyclicBarrierLOL {private static final CyclicBarrier cbr new CyclicBarrier(6);private static int aa ;private void ChooseHero(String route) {new Thread(()-{log.info(route 选择好了英雄等待进入游戏);try {cbr.await();} catch (InterruptedException e) {e.printStackTrace();log.info(route已被中断无法进行游戏);} catch (BrokenBarrierException e) {e.printStackTrace();log.info(route等待超时无法进行游戏);}}).start();}public static void main(String[] args) throws Exception {CyclicBarrierLOL lol new CyclicBarrierLOL();String[] rotes {上单,打野,中单,ADC,辅助};while (true){log.info(进入英雄选择界面等待英雄选择);for (int i 0; i 5; i) {String rote rotes[i];lol.ChooseHero(rote);}cbr.await();log.info(英雄选择完成开始进入游戏);aa;if (aa 2) return ;}}
}源码分析
CyclicBarrier的源码分析需要有ReentrantLock源码的基础。 CyclicBarrier的构造方法第一个参数资源数 第二个参数执行任务 当资源数为0的时候调用这个任务 这个资源数可以理解为Semaphore和CountDownLatch的资源数只不过Semaphore中持有资源数的线程才能通过CountDownLatch中资源数不为0的时候线程阻塞。 在CyclicBarrier中定义的parties和count都等于资源数。其中this.parties用于线程重置this.count 用于计数器控制线程是否阻塞 阻塞部分 当调用await()方法的时候进入阻塞路程
private int dowait(boolean timed, long nanos)//如果通过await调用dowait 那么timed false//如果通过await(long timeout, TimeUnit unit) 调用dowait 那么timed 就为truethrows InterruptedException, BrokenBarrierException,TimeoutException {//上锁操作 为什么要上锁 后续逻辑用到ReentrantLock条件锁的await方法 必须要先加锁才能使用await方法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;//当资源数变成0if (index 0) { // trippedboolean ranAction false;try {//判断构造方法中有没有传入Runnable final Runnable command barrierCommand;if (command ! null)//如果有 执行Runnable 任务command.run();ranAction true;//阻塞线程的唤醒 这里涉及到把条件等待队列的线程转到同步等待队列中 在进行唤醒nextGeneration();return 0;} finally {if (!ranAction)breakBarrier();}}// 当资源数不为0的时候执行自旋逻辑for (;;) {try {if (!timed)//这个trip 是 private final Condition trip lock.newCondition();//这里用到了ReentrantLock的条件队列 trip.await()阻塞当前线程并且进入条件阻塞队列//同时await方法会释放锁对象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 {//当资源数0的那部分代码执行完成后 最终还会执行这行代码//释放锁 这是ReentrantLock 释放锁的逻辑 唤醒同步等待队列中线程 这部分逻辑不在分析// 如果资源数不为0 不会执行该行代码 因为不为0 线程阻塞并且在条件等待队列中 不会往下执行lock.unlock();}}阻塞流程一个重点就是阻塞进入队列对应着就是trip.await();在看这部分源码之前先来看下AQS对于条件阻塞队列的实现 条件阻塞队列是一个单向链表如何创造这个链表就在addConditionWaiter()方法当中
private Node addConditionWaiter() {//定义一个节点等于链表的尾节点 一开始这个尾节点自然是null 因为还没有链表Node t lastWaiter;// If lastWaiter is cancelled, clean out.if (t ! null t.waitStatus ! Node.CONDITION) {unlinkCancelledWaiters();t lastWaiter;}//创建一个节点 Node.CONDITION 表示一个状态 等于-2 把当前线程和状态赋值给头节点Node node new Node(Thread.currentThread(), Node.CONDITION);//如果尾节点为nullif (t null)//链表的头节点等于创建出来的节点firstWaiter node;else//否则链表的下一个节点等于创建出来的节点t.nextWaiter node;//链表的尾节点等于创建出来的节点 lastWaiter node;//到此创建了一个新链表并且完成线程入队操作return node;}总结就是addConditionWaiter();方法如果没有链表创建链表完成入队如果已经创建了链表就直接入队 在回过头来看trip.await(); public final void await() throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();//创建链表并且如队Node node addConditionWaiter();//解锁操作int savedState fullyRelease(node);int interruptMode 0;//阻塞操作while (!isOnSyncQueue(node)) {//线程被阻塞在这个地方LockSupport.park(this);if ((interruptMode checkInterruptWhileWaiting(node)) ! 0)break;}//当线程被唤醒的时候重新获取锁acquireQueued(node, savedState)底层通过CAS获取锁//为什么要重新获取锁 dowait的finlly方法中时解锁操作 所以要先获取锁//如果此时有其他线程竞争CAS获取锁失败后继续阻塞//acquireQueued这个方法就是ReentrantLock底层获取锁的方法 不再分析了if (acquireQueued(node, savedState) interruptMode ! THROW_IE)interruptMode REINTERRUPT;if (node.nextWaiter ! null) // clean up if cancelledunlinkCancelledWaiters();if (interruptMode ! 0)reportInterruptAfterWait(interruptMode);}来看下解锁方法fullyRelease(node)
final int fullyRelease(Node node) {boolean failed true;try {//获取资源状态 之前有说过如果要加锁 通过CAS把State从0改成1int savedState getState();//解锁的逻辑 这部分和之前ReentrantLock解锁逻辑是一样的通过CAS把1改成0然后设置独占的线程为null 就不在展开了if (release(savedState)) {//解锁成功返回资源状态failed false;return savedState;} else {throw new IllegalMonitorStateException();}} finally {if (failed)node.waitStatus Node.CANCELLED;}}总结下来就是 trip.await(); 作用就是创建链表入队释放锁 阻塞入队线程。
dowait(boolean timed, long nanos)中当CyclicBarrier的资源数不为0的时候进行上述操作阻塞当前线程然后进入条件等待队列然后释放锁进行下一个线程的操作。 当资源数等于0的时候就执行了这部分代码重点是nextGeneration(); 这个是比较精髓的点 private void nextGeneration() {//重点就在于这个方法trip.signalAll();// 重复利用资源数开始新一轮的屏障count parties;generation new Generation();}public final void signalAll() {if (!isHeldExclusively())throw new IllegalMonitorStateException();//获取等待条件队列的头节点Node first firstWaiter;if (first ! null)//把头节点传入doSignalAll(first);}private void doSignalAll(Node first) {//把首尾节点设置为nulllastWaiter firstWaiter null;do {//设置一个节点等于 头节点的下一个节点Node next first.nextWaiter;//设置头节点的下一个节点为null 也就是说把头节点指向下一个节点的指针设置为null//由于之前就设置了头节点null 再把指针也设置为null 这样头节点就被删除了 也就是出队的意思first.nextWaiter null;//进入同步队列transferForSignal(first);//由于第一个头节点被删除了 那么第二个节点就要成为头节点first next;//精髓点在于 当头节点不为空进行了do while 的循环 直到把所有节点都出队并且进入到同步队列} while (first ! null);}doSignalAll方法就是单向链表出队进入同步等待队列 再来看进入同步队列的方法
final boolean transferForSignal(Node node) {if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))return false;//如果双向链表不存在创建双向链表同时把节点添加到双向链表中//这部分就是把条件等待队列的线程全部添加到同步等待队列中//这部分的源码在ReentrantLock源码解析的时候分析过就不在分析了Node p enq(node);//获取链表的一个状态(是否可以唤醒)int ws p.waitStatus;//如果状态满足唤醒阻塞线程if (ws 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))LockSupport.unpark(node.thread);return true;}当nextGeneration()执行完成后return 0 最后还会执行dowait(boolean timed, long nanos) finally中的lock.unlock();方法。这个方法就是唤醒同步等待队列的方法。这部分逻辑就是ReentrantLock的解锁逻辑。
CyclicBarrier源码流程总结
CyclicBarrier cbr new CyclicBarrier(3, new Runnable() {Overridepublic void run() {System.out.println(begin run);}});一上面构造出的CyclicBarrier为例子此时的资源数是3 当调用cbr.await();的时候底层调用dowait(false, 0L)方法。 dowait(false, 0L)的执行流程 一开始就加了ReentrantLock的锁资源数(一开始是3)减1判断资源数是不是等于0如果不等于0通过自旋调用 trip.await();其中trip是ReentrantLock的条件锁 trip.await();的作用
底层调用addConditionWaiter() 完成创建单向链表并且进入该队列调用fullyRelease(node) 完成释放锁调用LockSupport.park(this);阻塞线程 此时线程就是阻塞在这个地方当线程被唤醒的时候调用acquireQueued(node, savedState)通过CAS获取ReentrantLock的锁
当资源数判断为0时执行构造方法中的Runnable任务。然后条用nextGeneration(); nextGeneration();底层调用trip.signalAll(); 内部调用了doSignalAll。doSignalAll作用是条件等待队列的线程出队并且进入同步等待队列中。 nextGeneration();执行完成后 执行 dowait(false, 0L)的finally里面的方法 lock.unlock();此时释放ReentrantLock的锁。 释放锁自然会唤醒ReentrantLock同步等待队列中阻塞的线程。 到此CyclicBarrier完成一轮的阻塞-唤醒流程
