码迷,mamicode.com
首页 > 其他好文 > 详细

深入分析同步工具类之CountDownLatch

时间:2017-03-22 00:21:37      阅读:230      评论:0      收藏:0      [点我收藏+]

标签:sleep   主线程   ssh   代码   sign   turn   font   return   submit   

概览:

   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的时候会使用到

当第一个节点插入队列的时候,可以用下图来描绘生成的队列:

技术分享

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

private void doReleaseShared() {
     
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                
                if (ws == Node.SIGNAL) {//表明后继节点需要执行
                    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) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        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);
    }

挂起线程:

 

//挂起线程前先将该节点的前驱节点的waitStatus设为-1,即表示其后继节点代表的线程需要执行,这样上图Node B的前驱Node A的waitStatus==-1,
因为Node A 在初始化的时候和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(); }

  2、countDown()

public void countDown() {
        //AQS中的方法
        sync.releaseShared(1);
    }

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {//获取状态
            doReleaseShared();//恢复线程,见上面分析
            return true;
        }
        return false;
    }

 

深入分析同步工具类之CountDownLatch

标签:sleep   主线程   ssh   代码   sign   turn   font   return   submit   

原文地址:http://www.cnblogs.com/Non-Tecnology/p/6596034.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!