CountDownLatch 源码的角度简析为什么线程能等待

た入场券 2022-05-28 22:57 171阅读 0赞

Demo

public static void main(String[] args) {
        Worker worker1 = new Worker("worker1",1000l);
        Worker worker2 = new Worker("worker2",3000l);
        worker1.start();
        worker2.start();
        try {
                //3.worker1 和 worker2完成了，才往下走，是怎么做到的，看await方法
                cdl.await();
        } catch (InterruptedException e) {
                e.printStackTrace();
        }
        System.out.println("main thread done!!!");
    }
    //1.了解CountDownLatch的构造方法
    final static CountDownLatch cdl = new CountDownLatch(2);
    static class Worker extends Thread{
            private long time ;
            private String name;
            public Worker(String name ,long time){
                this.name = name;
                this.time = time;
            }
            public void run() {
                dowork();
                //模拟多长时间
                try {
                    TimeUnit.SECONDS.sleep(time);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //2.调用countdown,了解countdown原理
                cdl.countDown();
                System.out.println(name + " work done");
            }
            public void dowork(){
                System.out.println(name + " begin work");
            }

上面的例子比较简单，worker1 和 worker2 要同时完成了，才能打印出”main thread done!!!”  
了解了干什么用后，再来看源码是怎么实现的。 我标出了1,2,3的地方是重点，下面依次分析

1.构造方法

public class CountDownLatch { 
     private final Sync sync;
     public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            //构造函数里new了个Sync。 Sync 何许物也? 
            this.sync = new Sync(count);
     }
     //和ReentrantLock里的Sync很相似,也是extends AbstractQueuedSynchronizer
     private static final class Sync extends AbstractQueuedSynchronizer { 
            //AQS里的state,该例子中设为2
            Sync(int count) {
                setState(count);
            }
            ...
        }
    
    }

2.countDown

public class CountDownLatch { 
        public void countDown() {
            //调用了AQS的releaseShared方法,方法中调用内部类Syn中定义的的tryReleaseShared
            sync.releaseShared(1);
        }
    
        private static final class Sync extends AbstractQueuedSynchronizer { 
             int getCount() {
                return getState();
            }
            protected int tryAcquireShared(int acquires) {
                return (getState() == 0) ? 1 : -1;
            }
            protected boolean tryReleaseShared(int releases) {
                // CAS 将AQS中的state设为state-1，如果设置成功了看看state是否=0
                for (;;) {
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
        }
    }
    
    public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { 
        //AQS的releaseShared方法
        public final boolean releaseShared(int arg) {
                     if (tryReleaseShared(arg)) {
                          //如果状态是0就释放共享锁
                          doReleaseShared();
                       return true;
                         }
                      return false;
        }
    
    
    }

3.await

public class CountDownLatch { 
     public void await() throws InterruptedException {
            //调用AQS中的acquireSharedInterruptibly
            sync.acquireSharedInterruptibly(1);
        }
    }
    
    
    public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { 
             public final void acquireSharedInterruptibly(int arg)
                        throws InterruptedException {
                    if (Thread.interrupted())
                        throw new InterruptedException();
                    if (tryAcquireShared(arg) < 0)
                        doAcquireSharedInterruptibly(arg);
                }
    
    
           private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {
            //main线程成为一个等待的node节点，也是唯一的节点
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                for (;;) {
                   //main线程封装成为的node节点
                   //在一个循环中不断的询问是否轮到自己了，轮到了的话就是state==0,就return 了
                   //也就是countdown减到0了
                   //这里也就回答了线程为什么会等待
                    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);
            }
        }
    }