深入分析同步工具类之CountDownLatch

概览：

CountDownLatch又称闭锁，其作用是让一个或者多个线程挂起，直到其他的线程执行完后恢复挂起的线程，使其继续执行。内部维护着一个静态内部类Sync，该类继承AbstractQueuedSynchronizer（这个类之前分析过了，参见深入分析同步工具类之AbstractQueuedSynchronizer），Sync实例维护着state属性，调用await()方法，使当前线程挂起，当一个线程执行完后，调用countDown()方法，state-1，直到state变为0，被挂起的线程恢复执行。

常用的方法：

public CountDownLatch(int count) //构造函数初始化state值，可以理解为需要优先执行的线程数量

public void await() //调用后，所在的线程挂起

public void countDown() //优先执行的线程执行完调用，state-1，当state=0，执行阻塞队列中的线程

使用实例：

主线程和线程池中的一个线程会被挂起，等线程池中的另外5个线程执行完才会被执行

 CountDownLatch latch=new CountDownLatch(5);

        ExecutorService service= Executors.newFixedThreadPool(6);

        for (int i=0;i<5;i++){

            service.submit(new Runnable() {

                @Override

                public void run() {

                    try {

                        Thread.sleep(5000);

                        System.out.println(Thread.currentThread().getName());

                    } catch (InterruptedException e) {

                        e.printStackTrace();

                    }

                    latch.countDown();

                }

            });

        }

        service.submit(new Runnable() {

            @Override

            public void run() {

                try {

                    latch.await();

                    System.out.println(Thread.currentThread().getName());

                } catch (InterruptedException e) {

                    e.printStackTrace();

                }

            }

        });

        latch.await();

        System.out.println("主线程");

运行结果：

代码分析：

1.await()

//线程等待

public void await() throws InterruptedException {

        //AQS的实现

        sync.acquireSharedInterruptibly(1);

    }

public final void acquireSharedInterruptibly(int arg)

            throws InterruptedException {

        if (Thread.interrupted())

            throw new InterruptedException();

       //尝试获取状态，<0优先执行的线程未执行完，>0已执行完

       //Sync子类自己实现

        if (tryAcquireShared(arg) < 0)

            doAcquireSharedInterruptibly(arg);

    }

//获取AQS的state值，我们调用countDown会改变这个值

protected int tryAcquireShared(int acquires) {

            return (getState() == 0) ? 1 : -1;

        }

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);

        }

    }

addWaiter

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;

        if (pred != null) {

            node.prev = pred;

            if (compareAndSetTail(pred, node)) {

                pred.next = node;

                return node;

            }

        }

        //第一个节点添加队列时，头节点和尾节点都为空，需要初始化

        enq(node);

        return node;

    }

 private Node enq(final Node node) {

        for (;;) {

            Node t = tail;

            if (t == null) { // Must initialize

               //CAS设置头节点

                if (compareAndSetHead(new Node()))

               //头节点赋值给tail，此时head和tail指向同一个对象，如果对任何一个对象中的属性做修改，那么2个引用的属性也会跟着变(后面挂起线程的时候会修改waitStatus属性)

                    tail = head;

            } else {

                node.prev = t;

                //设置尾节点为当前结点，将2个节点串起来，即：node.prev = t;t.next = node;

                if (compareAndSetTail(t, node)) {

                    t.next = node;

                    return t;//退出循环

                }

            }

        }

    }

//注意：这里新生成的head节点并没有后继节点 head.next==null,并且head==node.prev(该node是第一次插入的节点) 这个特性在countDown的时候会使用到

假设线程A和线程B先后调用await()方法，并且tryAcquireShared(int acquires)<0,那么此时线程A、B分别被挂起，线程A和B在挂起时先后调用shouldParkAfterFailedAcquire方法，这样各自前驱节点的waitStatus就会被设置为-1，代表该线程需要执行；同时因为线程A的前驱和head引用同一个对象，

所以head==Node并且其waitStatus都为-1

此时队列的节点如下图：

调用挂起线程的方法：

//挂起线程前先将该节点的前驱节点的waitStatus设为-1，即表示其后继节点代表的线程需要执行，这样上图Node B的前驱Node的waitStatus==-1，

因为Node在初始化的时候和head同引用一个对象，所以head 的waitStatus也为-1

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {

        int ws = pred.waitStatus;

        if (ws == Node.SIGNAL)

            /*

             * 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;

    }

//挂起前程

private final boolean parkAndCheckInterrupt() {

        LockSupport.park(this);

        return Thread.interrupted();

    }

此时，若优先执行的线程执行完毕，调用countDown方法，更新state的值，也就是state==0的时候，就会恢复头节点的下一个节点所代表的线程

public void countDown() {

        sync.releaseShared(1);

    }

public final boolean releaseShared(int arg) {

        if (tryReleaseShared(arg)) {

            doReleaseShared();

            return true;

        }

        return false;

    }

protected boolean tryReleaseShared(int releases) {

            // Decrement count; signal when transition to zero

            for (;;) {

                int c = getState();

                if (c == 0)

                    return false;

                int nextc = c-1;

                if (compareAndSetState(c, nextc))

                   //最后一个线程执行完，返回true

                    return nextc == 0;

            }

        }

private void doReleaseShared() {

        for (;;) {

            Node h = head;

            if (h != null && h != tail) {

                int ws = h.waitStatus;

                if (ws == Node.SIGNAL) {//true

                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))

                        continue;            // loop to recheck cases

                    //恢复后继节点代表的线程

                    unparkSuccessor(h);

                }

                else if (ws == 0 &&

                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))

                    continue;                // loop on failed CAS

            }

            if (h == head)                   // loop if head changed

                break;

        }

    }

private void unparkSuccessor(Node node) {

        int ws = node.waitStatus;

        if (ws < 0)

            compareAndSetWaitStatus(node, ws, 0);

      //此时s==null，从尾节点开始找到最前面的节点(Node A),将其恢复

        Node s = node.next;

        if (s == null || s.waitStatus > 0) {

            s = null;

            for (Node t = tail; t != null && t != node; t = t.prev)

                if (t.waitStatus <= 0)

                    s = t;

        }

        if (s != null)

            LockSupport.unpark(s.thread);

    }

这个时候线程A开始运行，若未发生线程中断，则继续执行循环内的代码，p==head成立，此时获取状态成功，恢复后继节点代表的线程，并退出循环：

        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;

                    }

                }

              ...

            }

private void setHeadAndPropagate(Node node, int propagate) {

        Node h = head; // Record old head for check below

        //设置当前结点为头节点

        setHead(node);

        if (propagate > 0 || h == null || h.waitStatus < 0 ||

            (h = head) == null || h.waitStatus < 0) {

            Node s = node.next;

            if (s == null || s.isShared())

                //恢复后继节点的线程

                doReleaseShared();

        }

    }

最后，线程B运行获取状态，退出循环，程序结束

巴特西

深入分析同步工具类之CountDownLatch

最新文章

热门文章