zoukankan      html  css  js  c++  java
  • CountDownLatch 源码解析—— await()

    上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await() 的原理。

    我们已经知道await 能够让当前线程处于阻塞状态,直到锁存器计数为零(或者线程中断)。

    下面是它的源码。

    end.await();  
        ↓
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    sync 是CountDownLatch的内部类。下面是它的定义。

    private static final class Sync extends AbstractQueuedSynchronizer {
      ...
    }

    它继承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 这个类在java线程中属于一个非常重要的类。

    它提供了一个框架来实现阻塞锁,以及依赖FIFO等待队列的相关同步器(比如信号、事件等)。

    继续走下去,就跳到 AbstractQueuedSynchronizer 这个类中。

    sync.acquireSharedInterruptibly(1);  
        ↓
    public final void acquireSharedInterruptibly(int arg)  //AbstractQueuedSynchronizer
                throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

    这里有两个判断,首先判断线程是否中断,然后再进行下一个判断,这里我们主要看看第二个判断。

    protected int tryAcquireShared(int acquires) {
        return (getState() == 0) ? 1 : -1;
    }

    需要注意的是 tryAcquireShared 这个方法是在Sync 中实现的。

    AbstractQueuedSynchronizer 中虽然也有对它的实现,但是默认的实现是抛一个异常。

    tryAcquireShared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。

    我们可以看到Sync 中是通过判断state 是否为0 来返回对应的 int 值的。

    那么 state 又代表什么?  

    /**
     * The synchronization state.
     */
       private volatile int state;

    上面代码很清楚的表明 state 是表示同步的状态 。

    需要注意的是 state 使用 volatile 关键字修饰。

    volatile 关键字能够保证 state 的修改立即被更新到主存,当有其他线程需要读取时,会去内存中读取新值。

    也就是保证了state的可见性。是最新的数据。

    走到这里 state 是多少呢?

    这里我们就需要看一看CountDownLatch 的 构造函数了。

    CountDownLatch end = new CountDownLatch(2);
        ↓
    public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }
        ↓
    Sync(int count) {
        setState(count);
    }

    原来构造函数中的数字就是这个作用啊,用来set state 。

    所以我们这里state == 2 了。tryAcquireShared 就返回 -1。进入到下面

    doAcquireSharedInterruptibly(arg);
        ↓
    private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head) {
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

    OK,这段代码有点长,里面还调用了几个函数。我们一行一行的看。

    第一行 出现了一个新的类 Node。

    Node 是AQS(AbstractQueuedSynchronizer)类中的内部类,定义了一种链式结构。如下所示。

         +------+  prev +-----+       +-----+
    head |      | <---- |     | <---- |     |  tail
         +------+       +-----+       +-----+

    千万记住这个结构。

    第一行代码中还有一个方法 addWaiter(Node.SHARED) 。

    addWaiter(Node.SHARED)  //Node.SHARED  表示该结点处于共享模式
    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail; // private transient volatile Node tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

    首先是构造了一个Node,将当前的线程存进去了,模式是共享模式。

    tail 表示 这个等待队列的队尾,此刻是null. 所以 pred == null ,进入到enq(node) ;

    enq(node)
        ↓
    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

    同样tail 为 null , 进入到 compareAndSetHead 。

    compareAndSetHead(new Node())
        ↓
    /**
     * CAS head field. Used only by enq.
     */
    private final boolean compareAndSetHead(Node update) {
        return unsafe.compareAndSwapObject(this, headOffset, null, update);
    }

    这是一个CAS操作,如果head 是 null 的话,等待队列的 head 就会被设置为 update 的值,也就是一个新的结点。

     tail = head;  那么此时 tail 也不再是null了。进入下一次的循环。

    这次首先将node 的 prev 指针指向 tail ,然后通过一个CAS 操作将node 设置为尾部,并返回了队列的 tail ,也就是 node 。

    等待队列的模型变化如下

               +------+  prev      +----------------+
    head(tail) |      | <---- node | currentThread  |
               +------+            +----------------++------+  prev            +----------------+
    head   |      | <---- node(tail) | currentThread  |
           +------+                  +----------------+

    ok,到了这里await 方法 就返回了,是一个 thread 等于当前线程的Node。

    返回到 doAcquireSharedInterruptibly(int arg) 中,进入下面循环。

    for (;;) {
        final Node p = node.predecessor();
        if (p == head) {
            int r = tryAcquireShared(arg);
            if (r >= 0) {
                setHeadAndPropagate(node, r);
                p.next = null; // help GC
                failed = false;
                return;
            }
        }
        if (shouldParkAfterFailedAcquire(p, node) &&
            parkAndCheckInterrupt())
            throw new InterruptedException();
    }

    这个时候假设state 仍然大于0,那么此时 r < 0,所以进入到 shouldParkAfterFailedAcquire 这个方法 。

    shouldParkAfterFailedAcquire(p, node)
        ↓
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)  //static final int SIGNAL    = -1;
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
        ↓
    /**
     * CAS waitStatus field of a node.
     */
    private static final boolean compareAndSetWaitStatus(Node node,
                                                         int expect,
                                                         int update) {
        return unsafe.compareAndSwapInt(node, waitStatusOffset,
                                        expect, update);
    }
        
        

    可以看到 shouldParkAfterFailedAcquire  也是一路走,走到 compareAndSetWaitStatus。

    compareAndSetWaitStatus 将 prev 的 waitStatus 设置为 Node.SIGNAL 。

    Node.SIGNAL 表示后续结点中的线程需要被unparking(类似被唤醒的意思)。该方法返回false。

    经过这轮循环,队列模型变成下面状态

           +--------------------------+   prev           +------------------+
    head   | waitStatus = Node.SIGNAL | <---- node(tail) | currentThread    |
           +--------------------------+                  +------------------+

    因为shouldParkAfterFailedAcquire返回的是false,所以后面这个条件就不再看了。继续 for (;;)  中的循环。

    如果state仍然大于0,再次进入到 shouldParkAfterFailedAcquire。

    这次因为head 中的waitStatus 为 Node.SIGNAL ,所以 shouldParkAfterFailedAcquire 返回true。

    这次就需要看parkAndCheckInterrupt 这个方法了。

     private final boolean parkAndCheckInterrupt() {
            LockSupport.park(this);
            return Thread.interrupted();
        }

    ok,线程没有被中断,所以,返回false。继续 for (;;)  中的循环。

    如果state 一直大于0,并且线程一直未被中断,那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。

    那么什么情况下跳出循环呢?也就是什么情况下state 会 小于0呢? 下一篇文章 我将说明。

    总结一下,await()  方法 其实就是初始化一个队列,将需要等待的线程(state > 0)加入一个队列中,并用waitStatus 标记后继结点的线程状态。

  • 相关阅读:
    文件修改的两种方式
    人工智能 01. 语音合成,语音识别,相似度,图灵机器人,智能对话
    flask --- 04 . 偏函数, 线程安全,栈堆,
    flask --- 03 .特殊装饰器, CBV , redis ,三方组件
    flask --- 02. 路由, 初始化配置,蓝图
    linux --- 10.常见命令
    flask --- 01 .初始
    linux --- 9. docker 容器 和 rabbitmq 队列
    linux --- 8. mysql数据库,redis 数据库
    linux --- 7. 路飞学城部署
  • 原文地址:https://www.cnblogs.com/cuglkb/p/8572609.html
Copyright © 2011-2022 走看看