java基础-并发编程-CountDownLatch（JDK1.8）源码学习

CountDownLatch方法调用与类关系图

一、初始化：public CountDownLatch(int count)

public CountDownLatch(int count) {
     if (count < 0) throw new IllegalArgumentException("count < 0");
     this.sync = new Sync(count);
 }
 
Sync(int count) {
	// 将参数**count**值赋值给AQS类中**private volatile int state**变量
    setState(count);
}
1
2
3
4
5
6
7
8
9

二、获取共享锁并将当前线程挂起（获取到共享锁时不挂起）：public void await()

public void await() throws InterruptedException {
        // 此处获取可中断的共享锁
        sync.acquireSharedInterruptibly(1);
    }

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        // 如果线程已被中断，则抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
        // 尝试获取共享锁，此钩子方法在java.util.concurrent.CountDownLatch.Sync类中实现
        if (tryAcquireShared(arg) < 0)
        	// 未获取到共享锁，则执行doAcquireSharedInterruptibly方法
            doAcquireSharedInterruptibly(arg);
    }   
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

protected int tryAcquireShared(int acquires) {
			// 当初始化时设置的state属性值为0（countDown方法中修改此值）时才认为可以直接获取到锁
            return (getState() == 0) ? 1 : -1;
        }
1
2
3
4

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 添加一个共享节点到同步队列，此共享节点有以下特征
        // 1、该节点为同步队列尾节点
        // 2、该节点中封装了当前线程信息
        // 3、该节点此时属性waitStatus赋值为0
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                // 如果当前线程为第一次获取锁，则node节点时同步队列中第二个元素，因此p == head
                // 如果当前线程刚被唤醒，则node节点时同步队列中第二个元素，因此p == head
                if (p == head) {
                	// 再次尝试获取共享锁
                    int r = tryAcquireShared(arg);
                    // 当所有多线程都完成任务，调用countDown方法将state属性值设置为0时，r值为1
                    // 当一部分线程完成任务，调用countDown方法后state属性值不为0，r值为-1
                    if (r >= 0) {
                    	// 将当前节点node设置为同步队列头节点，并传播
                        setHeadAndPropagate(node, r);
                        // 将前头节点head从当前前node上移除（上面已将前头节点head从当前前node上移除,此时前头节点完全从同步队列上移除）
                        p.next = null; // help GC
                        // 完成线程唤醒，failed 重新赋值false,不会在此方法最后执行cancelAcquire(node);
                        failed = false;
                        return;
                    }
                }
                // 如果同步队列中node前面还有其他等待节点，则将node节点中属性waitStatus赋值为-1,此时返回false,不会将该线程挂起
                // shouldParkAfterFailedAcquirenode一般执行两遍，那么很有可能第二遍的时候，发现自己的前驱突然变成head了并且获取共享锁成功，又或者本来第一遍的前驱就是head但第二遍获取共享锁成功了
                if (shouldParkAfterFailedAcquire(p, node) &&
                	// 执行完前一个判断后再循环一次，如果同步队列中node前面还有其他等待节点，则挂起该线程
                	// 线程调用两次shouldParkAfterFailedAcquire，和一次parkAndCheckInterrupt后，便阻塞了。之后就只能等待别人unpark自己了
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
        	// 线程被中断，抛出异常，此时failed为初始值true,取消获取锁
            if (failed)
                cancelAcquire(node);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

    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前节点
            node.prev = pred;
            // 将node节点设置为尾节点
            if (compareAndSetTail(pred, node)) {
            	// 设置成功，则将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
            	// 尾节点为null，头节点肯定也为空，此时使用空节点重置头节点
                if (compareAndSetHead(new Node()))
                	// 尾节点和头节点指向同一个节点
                    tail = head;
            } else {
            	// 将尾节点设置为node前节点
                node.prev = t;
                // 将node节点设置为尾节点
                if (compareAndSetTail(t, node)) {
                	// 设置成功，则将node设置为前尾节点的后继节点
                    t.next = node;
                    return t;
                }
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

AQS深入理解 setHeadAndPropagate源码分析 JDK8

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        // h保存了旧的head，但现在head已经变成node了
        setHead(node);
        // 如果同时有两个线程进入方法【doReleaseShared】，同时执行到【compareAndSetWaitStatus(h, Node.SIGNAL, 0)】
        // 其中只有一个执行unparkSuccessor，两一个将head节点waitStatus值修改为了-3，然后退出，此种情况在当前判断中传播唤醒下一节点
        // TODO
        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 setHead(Node node) {
		// 将当前节点设置为头节点head
        head = node;
        // 头节点中线程为空，因为头节点中的线程已经被唤醒且执行完，使用空头节点存储待唤醒节点的状态
        node.thread = null;
        // 将前头节点从当前节点上移除
        node.prev = null;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

private void doReleaseShared() {
        
        for (;;) {
            Node h = head;
            // 同步队列中存在待唤醒的线程
            // h != null: 头节点为空节点
            // h != tail：同步队列刚初始化，还没有待唤醒的线程
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                // head节点的后继节点待唤醒
                if (ws == Node.SIGNAL) {
                	// CAS修改头节点waitStatus值为0，修改失败再次循环 ——①
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // 修改成功，唤醒后继节点
                    unparkSuccessor(h);
                }
               // 若①处执行成功后线程被打断，此时ws为0，此时将修改头节点waitStatus值为-3
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            // 如果此时head节点已被修改，则可以继续唤醒下一个节点
            // 如果此时head节点未被修改，则跳出循环
            if (h == head)                   // loop if head changed
                break;
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

private void unparkSuccessor(Node node) {
		// TODO
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        Node s = node.next;
        // 如果下一个节点不存在或下一个节点线程被中断（s.waitStatus == 1）
        if (s == null || s.waitStatus > 0) {
            s = null;
            // 跳过该节点，从尾节点向前遍历
            for (Node t = tail; t != null && t != node; t = t.prev)
                // 找到不是node节点（头节点）的最前面一个待唤醒节点
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

三、递减闩锁的计数：public void countDown()

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

public final boolean releaseShared(int arg) {
		// 尝试释放共享锁
        if (tryReleaseShared(arg)) {
        	// 尝试失败，释放共享锁（注解在上面）
            doReleaseShared();
            return true;
        }
        return false;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13

protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        for (;;) {
            int c = getState();
            // state==0时门闩已打开，此时调用countDown无意义
            if (c == 0)
                return false;
            // state!=0, state-1,CAS设置给state
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13