Java并发编程之CountDownLatch源码分析

原创
2019/01/06 17:18
阅读数 252

CountDownLatch介绍

CountDownLatch是JDK1.5提供用来多线程并发同步的工具,可以让一个或多个线程等待另一个线程执行完再执行。

例子

private static CountDownLatch countDownLatch = new CountDownLatch(1);
	
	static class ThreadRunnable1 implements Runnable{
		
		@Override
		public void run() {
			countDownLatch.countDown();		
		}
		
	}
	public static void main(String[] args) throws InterruptedException {
		Thread thread1 = new Thread(new ThreadRunnable1());
		thread1.setName("线程1");
		thread1.start();
		countDownLatch.await();
	}

如上例子,先创建了计数器为1的CountDownLatch,运行线程1调用countDown()方法,主线程调用await()方法并阻塞。调用countDown()会将计数器减1当计数器到达0,会唤醒await()的阻塞并继续执行下面的逻辑。

CountDownLatch原理分析

CountDownLatch定义了一个成员变量

private final Sync sync;

Sync继承至AbstractQueuedSynchronizer是CountDownLatch的内部类,提供了阻塞队列的支持。CountDownLatch的UML关系图如下

CountDownLatch主要利用AbstractQueuedSynchronizer(下文称AQS)中的队列和队列的通知传播的技术实现。创建CountDownLatch时先初始化设置的计数器,调用countDown()可以减计数器,当计数器到0时会通知去唤醒调用了await()的线程,而调用了await()线程会被加入到AQS同步阻塞队列,所以先唤醒AQS的同步阻塞队列的头节点线程,头节点线程被唤醒后会同样唤醒AQS同步阻塞队列的下个节点,以此传播唤醒所有调用了await()的线程。

下面来分析下CountDownLatch的2个主要方法countDown()和await()

countDown()源码分析

countDown()源码如下

 public void countDown() {
        sync.releaseShared(1);
 }

countDown()是直接调用了AQS里的releaseShared(1)方法我们来看下releaseShared(1)方法

releaseShared(1)源码如下

 public final boolean releaseShared(int arg) {
 		//计数器减一并返回剩余计数器是否等于0
        if (tryReleaseShared(arg)) {
		//唤醒同步阻塞队列中的头节点的下个节点线程
            doReleaseShared();
            return true;
        }
        return false;
    }

先调用tryReleaseShared(arg)将计数器减一并返回剩余计数器是否等于0,如果是调用doReleaseShared()唤醒同步阻塞队列中的头节点的下个节点线程。

tryReleaseShared(arg)方法是CountDownLatch内部类Sync的实现。我们来看下tryReleaseShared(arg)的源码。

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
		//获取当前计时器
                int c = getState();
		//如果已经为0则返回false,说明被之前线程减成0了
                if (c == 0)
                    return false;
		//计时器减一
                int nextc = c-1;
		//CSA更新值,如果更新失败自旋更新直至成功
                if (compareAndSetState(c, nextc))
		//更新成功后剩余的计时器是否为0
                    return nextc == 0;
            }
        }

tryReleaseShared(arg)实现很简单,获取当前计时器减一并使用CAS方式更新,如果更新失败自旋更新直至成功,最后返回剩余的计时器是否为0。

doReleaseShared()这个方法比较重要await()中也会调用,这里先不做分析,下面会在await()方法中一起分析。

await()源码分析

await()源码如下

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

await()是直接调用了AQS里的acquireSharedInterruptibly(1)方法我们来看下acquireSharedInterruptibly(1)方法

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
	//判断当前线程是否被中断,中断的话抛出异常
        if (Thread.interrupted())
            throw new InterruptedException();
	//判断当前计数器是否为0是的话返回1否则返回-1
        if (tryAcquireShared(arg) < 0)
	//加入到同步阻塞队列等待被唤醒
            doAcquireSharedInterruptibly(arg);
    }

先判断当前线程是否被中断,中断的话抛出异常。然后调用tryAcquireShared(arg)判断当前计数器是否为0是的话返回1否则返回-1,这里如果计数器为0了就表示计数器已经被countDown()减完了无需进行阻塞。如果计数器未被减完则调用 doAcquireSharedInterruptibly(arg)加入到同步阻塞队列等待被唤醒。

tryAcquireShared(arg)方法也是CountDownLatch内部类Sync的实现,下面看下tryAcquireShared(arg)的源码,源码如下

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

实现很简单取计数器判断是否为0是的话返回1否则返回-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) {
		//判断当前计数器是否为0是的话返回1否则返回-1
                    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);
        }
    }

doAcquireSharedInterruptibly里有个循环,这个循环的主要作用就是在线程唤醒后重试获取锁直到获取锁。node.predecessor()获取当前线程节点的前一个节点,如果是头节点,则当前线程尝试获取锁,获取锁成功调用setHeadAndPropagate(node, r)设置头节点并传播通知。如果获取失败或者非头节点则调用shouldParkAfterFailedAcquire(p, node)判断是否需要阻塞等待,如果需要阻塞等待则调用parkAndCheckInterrupt()阻塞当前线程并让出cup资源资质被前一个节点唤醒,如果线程被中断则抛出InterruptedException()异常。

doAcquireSharedInterruptibly中的addWaiter(Node.SHARED),shouldParkAfterFailedAcquire(p, node)和parkAndCheckInterrupt()方法之前的文章已经描述过(链接)这里就不再赘述。

下面来看下setHeadAndPropagate(node, r)的实现

 private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head;
	//更新当前节点为头节点
        setHead(node);
	//这里propagate必大于0
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
	//如果下个节点为空或者是共享模式,就要传播唤醒下个节点
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

主要逻辑就是设置当前节点为头节点,如果当前节点的下个节点不为空就要传播唤醒下个节点。

下面主要看下doReleaseShared()方法

  private void doReleaseShared() {
        for (;;) {
		//1.获取头节点
            Node h = head;
		//2.头节点不会空,而且头节点有下个节点
            if (h != null && h != tail) {
                int ws = h.waitStatus;
		//3.如果头节点状态是SIGNAL,则唤醒
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);
                }
		//4.如果节点为初始状态则设置节点为PROPAGATE状态
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
		//5.如果头节点改变,继续循环
            if (h == head)                   // loop if head changed
                break;
        }
    }

整个循环就是利用乐观锁的方式改变节点状态并唤醒下个节点线程。5个步骤如下

  • (1)获取头节点

  • (2)头节点不会空,而且头节点有下个节点

  • (3)步骤就是判断头节点状态是不是SIGNAL状态,这里如果头节点状态是SIGNAL说明此节点已经被它的下个节点设置为SIGNAL状态,需要唤醒此节点的下个节点, 这里调用unparkSuccessor(h)唤醒,unparkSuccessor(h)源码方法之前的文章已经描述过(链接)这里就不再赘述。

  • (4)如果头节点为初始状态0这设置状态为PROPAGATE。这里不会唤醒他的下个节点,那么下个节点会不会一直阻塞。其实下个节点线程不会一直阻塞,如果头节点为初始状态0说明它的下个节点还没调用shouldParkAfterFailedAcquire(p, node)方法改变头节点的状态为SIGNAL。如果在步骤4前下个节点线程调用shouldParkAfterFailedAcquire(p, node)改变头节点的状态为SIGNAL,shouldParkAfterFailedAcquire(p, node)会返回false,继续循环这时下个节点线程不会阻塞直接进入setHeadAndPropagate(node, r)方法。如果当前线程执行步骤4设置失败,说明下个节点线程在调用此方法之前先调用shouldParkAfterFailedAcquire(p, node)改变头节点的状态,这样也不影响流程,这时重新循环调用unparkSuccessor(h)即可。不过本人没看出步骤4在CountDownLatch有什么实质用处。

  • (5)步骤5是循环退出的条件如果头节点未变退出循环,否则继续循环,头节点什么时候会改变,一个是唤醒下个节点后下个节点在步骤5前调用setHeadAndPropagate(node, r)方法更新头节点,另外一个就是另一个线程添加节点后没有阻塞直接setHeadAndPropagate(node, r)方法更新头节点。

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部