zoukankan      html  css  js  c++  java
  • 【JUC】JDK1.8源码分析之CyclicBarrier(四)

    一、前言

      有了前面分析的基础,现在,接着分析CyclicBarrier源码,CyclicBarrier类在进行多线程编程时使用很多,比如,你希望创建一组任务,它们并行执行工作,然后在进行下一个步骤之前等待,直至所有的任务都完成,和join很类似,下面,开始分析源码。

    二、CyclicBarrier数据结构

      分析源码可以知道,CyclicBarrier底层是基于ReentrantLockAbstractQueuedSynchronizer来实现的,所以,CyclicBarrier的数据结构也依托于AQS的数据结构,在前面对AQS的分析中已经指出了其数据结构,在这里不再累赘。

    三、CyclicBarrier源码分析

      3.1 类的继承关系

    public class CyclicBarrier {}

      说明:可以看到CyclicBarrier没有显示继承哪个父类或者实现哪个父接口,根据Java语言规定,可知其父类是Object。

      3.2 类的内部类

      CyclicBarrier类存在一个内部类Generation,每一次使用的CycBarrier可以当成Generation的实例,其源代码如下

    private static class Generation {
        boolean broken = false;
    }

      说明:Generation类有一个属性broken,用来表示当前屏障是否被损坏。

      3.3 类的属性 

    public class CyclicBarrier {
        
        /** The lock for guarding barrier entry */
        // 可重入锁
        private final ReentrantLock lock = new ReentrantLock();
        /** Condition to wait on until tripped */
        // 条件队列
        private final Condition trip = lock.newCondition();
        /** The number of parties */
        // 参与的线程数量
        private final int parties;
        /* The command to run when tripped */
        // 由最后一个进入 barrier 的线程执行的操作
        private final Runnable barrierCommand;
        /** The current generation */
        // 当前代
        private Generation generation = new Generation();
        // 正在等待进入屏障的线程数量
        private int count;
    }
    View Code

      说明:该属性有一个为ReentrantLock对象,有一个为Condition对象,而Condition对象又是基于AQS的,所以,归根到底,底层还是由AQS提供支持。

      3.4 类的构造函数

      1. CyclicBarrier(int, Runnable)型构造函数 

    public CyclicBarrier(int parties, Runnable barrierAction) {
            // 参与的线程数量小于等于0,抛出异常
            if (parties <= 0) throw new IllegalArgumentException();
            // 设置parties
            this.parties = parties;
            // 设置count
            this.count = parties;
            // 设置barrierCommand
            this.barrierCommand = barrierAction;
        }
    View Code

      说明:该构造函数可以指定关联该CyclicBarrier的线程数量,并且可以指定在所有线程都进入屏障后的执行动作,该执行动作由最后一个进行屏障的线程执行。

      2. CyclicBarrier(int)型构造函数 

    public CyclicBarrier(int parties) {
            // 调用含有两个参数的构造函数
            this(parties, null);
        }
    View Code

      说明:该构造函数仅仅执行了关联该CyclicBarrier的线程数量,没有设置执行动作。

      3.5 核心函数分析

      1. dowait函数

      此函数为CyclicBarrier类的核心函数,CyclicBarrier类对外提供的await函数在底层都是调用该了doawait函数,其源代码如下。

    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) {  // 正在等待进入屏障的线程数量为0,所有线程都已经进入
                    // 运行的动作标识
                    boolean ranAction = false;
                    try {
                        // 保存运行动作
                        final Runnable command = barrierCommand;
                        if (command != null) // 动作不为空
                            // 运行
                            command.run();
                        // 设置ranAction状态
                        ranAction = true;
                        // 进入下一代
                        nextGeneration();
                        return 0;
                    } finally {
                        if (!ranAction) // 没有运行的动作
                            // 损坏当前屏障
                            breakBarrier();
                    }
                }
    
                // loop until tripped, broken, interrupted, or timed out
                // 无限循环
                for (;;) {
                    try {
                        if (!timed) // 没有设置等待时间
                            // 等待
                            trip.await(); 
                        else if (nanos > 0L) // 设置了等待时间,并且等待时间大于0
                            // 等待指定时长
                            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();
    
                    if (g != generation) // 不等于当前代
                        // 返回索引
                        return index;
    
                    if (timed && nanos <= 0L) { // 设置了等待时间,并且等待时间小于0
                        // 损坏屏障
                        breakBarrier();
                        // 抛出异常
                        throw new TimeoutException();
                    }
                }
            } finally {
                // 释放锁
                lock.unlock();
            }
        }
    View Code

      说明:dowait方法的逻辑会进行一系列的判断,大致流程如下。

      2. nextGeneration函数 

      此函数在所有线程进入屏障后会被调用,即生成下一个版本,所有线程又可以重新进入到屏障中,其源代码如下  

        private void nextGeneration() {
            // signal completion of last generation
            // 唤醒所有线程
            trip.signalAll();
            // set up next generation
            // 恢复正在等待进入屏障的线程数量
            count = parties;
            // 新生一代
            generation = new Generation();
        }
    View Code

      在此函数中会调用AQS的signalAll方法,即唤醒所有等待线程。如果所有的线程都在等待此条件,则唤醒所有线程。其源代码如下 

    public final void signalAll() {
                if (!isHeldExclusively()) // 不被当前线程独占,抛出异常
                    throw new IllegalMonitorStateException();
                // 保存condition队列头结点
                Node first = firstWaiter;
                if (first != null) // 头结点不为空
                    // 唤醒所有等待线程
                    doSignalAll(first);
            }
    View Code

      说明:此函数判断头结点是否为空,即条件队列是否为空,然后会调用doSignalAll函数,doSignalAll函数源码如下  

    private void doSignalAll(Node first) {
                // condition队列的头结点尾结点都设置为空
                lastWaiter = firstWaiter = null;
                // 循环
                do {
                    // 获取first结点的nextWaiter域结点
                    Node next = first.nextWaiter;
                    // 设置first结点的nextWaiter域为空
                    first.nextWaiter = null;
                    // 将first结点从condition队列转移到sync队列
                    transferForSignal(first);
                    // 重新设置first
                    first = next;
                } while (first != null);
            }
    View Code

      说明:此函数会依次将条件队列中的节点转移到同步队列中,会调用到transferForSignal函数,其源码如下

        final boolean transferForSignal(Node node) {
            /*
             * If cannot change waitStatus, the node has been cancelled.
             */
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
    
            /*
             * Splice onto queue and try to set waitStatus of predecessor to
             * indicate that thread is (probably) waiting. If cancelled or
             * attempt to set waitStatus fails, wake up to resync (in which
             * case the waitStatus can be transiently and harmlessly wrong).
             */
            Node p = enq(node);
            int ws = p.waitStatus;
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);
            return true;
        }
    View Code

      说明:此函数的作用就是将处于条件队列中的节点转移到同步队列中,并设置结点的状态信息,其中会调用到enq函数,其源代码如下。  

    private Node enq(final Node node) {
            for (;;) { // 无限循环,确保结点能够成功入队列
                // 保存尾结点
                Node t = tail;
                if (t == null) { // 尾结点为空,即还没被初始化
                    if (compareAndSetHead(new Node())) // 头结点为空,并设置头结点为新生成的结点
                        tail = head; // 头结点与尾结点都指向同一个新生结点
                } else { // 尾结点不为空,即已经被初始化过
                    // 将node结点的prev域连接到尾结点
                    node.prev = t; 
                    if (compareAndSetTail(t, node)) { // 比较结点t是否为尾结点,若是则将尾结点设置为node
                        // 设置尾结点的next域为node
                        t.next = node; 
                        return t; // 返回尾结点
                    }
                }
            }
        }
    View Code

      说明:此函数完成了结点插入同步队列的过程,也很好理解。

      综合上面的分析可知,newGeneration函数的主要方法的调用如下,之后会通过一个例子详细讲解。

      3. breakBarrier函数

      此函数的作用是损坏当前屏障,会唤醒所有在屏障中的线程。源代码如下  

    private void breakBarrier() {
            // 设置状态
            generation.broken = true;
            // 恢复正在等待进入屏障的线程数量
            count = parties;
            // 唤醒所有线程
            trip.signalAll();
        }
    View Code

      说明:可以看到,此函数也调用了AQS的signalAll函数,由signal函数提供支持。

    四、示例

      下面通过一个例子来详解CyclicBarrier的使用和内部工作机制,源代码如下  

    package com.hust.grid.leesf.cyclicbarrier;
    
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    /**
     * 
     * @author leesf
     * @time   2016.4.16
     */
    class MyThread extends Thread {
        private CyclicBarrier cb;
        public MyThread(String name, CyclicBarrier cb) {
            super(name);
            this.cb = cb;
        }
        
        public void run() {
            System.out.println(Thread.currentThread().getName() + " going to await");
            try {
                cb.await();
                System.out.println(Thread.currentThread().getName() + " continue");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    public class CyclicBarrierDemo {
        public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
            CyclicBarrier cb = new CyclicBarrier(3, new Thread("barrierAction") {
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " barrier action");
                    
                }
            });
            MyThread t1 = new MyThread("t1", cb);
            MyThread t2 = new MyThread("t2", cb);
            t1.start();
            t2.start();
            System.out.println(Thread.currentThread().getName() + " going to await");
            cb.await();
            System.out.println(Thread.currentThread().getName() + " continue");
    
        }
    }

      运行结果(某一次): 

    t1 going to await
    main going to await
    t2 going to await
    t2 barrier action
    t2 continue
    t1 continue
    main continue

      说明:根据结果可知,可能会存在如下的调用时序。

      说明:由上图可知,假设t1线程的cb.await是在main线程的cb.barrierAction动作是由最后一个进入屏障的线程执行的。根据时序图,进一步分析出其内部工作流程。

      ① main(主)线程执行cb.await操作,主要调用的函数如下。

      说明:由于ReentrantLock的默认采用非公平策略,所以在dowait函数中调用的是ReentrantLock.NonfairSync的lock函数,由于此时AQS的状态是0,表示还没有被任何线程占用,故main线程可以占用,之后在dowait中会调用trip.await函数,最终的结果是条件队列中存放了一个包含main线程的结点,并且被禁止运行了,同时,main线程所拥有的资源也被释放了,可以供其他线程获取。

      ② t1线程执行cb.await操作,其中假设t1线程的lock.lock操作在main线程释放了资源之后,则其主要调用的函数如下。

      说明:可以看到,之后condition queue(条件队列)里面有两个节点,包含t1线程的结点插入在队列的尾部,并且t1线程也被禁止了,因为执行了park操作,此时两个线程都被禁止了。

      ③ t2线程执行cb.await操作,其中假设t2线程的lock.lock操作在t1线程释放了资源之后,则其主要调用的函数如下。

      说明:由上图可知,在t2线程执行await操作后,会直接执行command.run方法,不是重新开启一个线程,而是最后进入屏障的线程执行。同时,会将Condition queue中的所有节点都转移到Sync queue中,并且最后main线程会被unpark,可以继续运行。main线程获取cpu资源,继续运行。

      ④ main线程获取cpu资源,继续运行,下图给出了主要的方法调用。

      说明:其中,由于main线程是在AQS.CO的wait中被park的,所以恢复时,会继续在该方法中运行。运行过后,t1线程被unpark,它获得cpu资源可以继续运行。

      ⑤ t1线程获取cpu资源,继续运行,下图给出了主要的方法调用。

      说明:其中,由于t1线程是在AQS.CO的wait方法中被park,所以恢复时,会继续在该方法中运行。运行过后,Sync queue中保持着一个空节点。头结点与尾节点均指向它。

       注意:在线程await过程中中断线程会抛出异常,所有进入屏障的线程都将被释放。至于CyclicBarrier的其他用法,读者可以自行查阅API,不再累赘。

    五、总结

      有了AQS与ReentrantLock的基础,分析CyclicBarrier就会非常简单,因为其底层就是由两者支撑的,关于CycylicBarrier的源码就分析到此,有疑问的读者,欢迎交流,谢谢各位园友的观看~

      

  • 相关阅读:
    5.3公理系统
    5.2逻辑语言vs逻辑演算
    Spike-in 对照(Spike-in control)
    R 语言学习笔记(4)—— 数值&字符处理
    R 语言学习笔记(3)—— 基础绘图
    R语言学习笔记(2)——数据结构与数据集
    R 语言学习笔记(1)——R 工作空间与输入输出
    单核苷酸多态性SNP(single nucleotide polymorphism)
    你真的遵守编码规范了吗
    论牧羊犬如何混迹于Scrum圈
  • 原文地址:https://www.cnblogs.com/leesf456/p/5392816.html
Copyright © 2011-2022 走看看