JUC并发—7.AQS源码分析三

大纲
1.等待多线程完成的CountDownLatch先容
2.CountDownLatch.await()方法源码
3.CountDownLatch.coutDown()方法源码
4.CountDownLatch总结
5.控制并发线程数的Semaphore先容
6.Semaphore的令牌获取过程
7.Semaphore的令牌释放过程
8.同步屏障CyclicBarrier先容
9.CyclicBarrier的await()方法源码
10.使用CountDownLatch等待注册的完成
11.使用CyclicBarrier将工作任务多线程分而治之
12.使用CyclicBarrier聚合服务接口的返回效果
13.使用Semaphore等待指定数量线程完成任务
 
volatile、synchronized、CAS、AQS、读写锁、锁优化和锁故障、并发集合、线程池、同步组件
 
1.等待多线程完成的CountDownLatch
(1)CountDownLatch的简介
(2)CountDownLatch的应用
(3)CountDownLatch的例子
 
(1)CountDownLatch的简介
CountDownLatch答应一个或多个线程等待其他线程完成操纵。CountDownLatch提供了两个焦点方法，分别是await()方法和countDown()方法。CountDownLatch.await()方法让调用线程举行阻塞进入等待状态，CountDownLatch.countDown()方法用于对计数器举行递减。
 
CountDownLatch在构造时需要传入一个正整数作为计数器初始值。线程每调用一次countDown()方法，都会对该计数器减一。当计数器为0时，会唤醒所有执行await()方法时被阻塞的线程。
 
(2)CountDownLatch的应用
应用一：
使用多线程去解析一个Excel里多个sheet的数据，每个线程解析一个sheet里的数据，等所有sheet解析完再提示处理完成。此时便可以使用CountDownLatch来实现，当然可以使用Thread.join()方法。
 
留意：Thread.join()方法是基于wait()和notify()来实现的。在main线程里开启一个线程A，main线程假如执行了线程A的join()方法，那么就会导致main线程被阻塞，main线程会等待线程A执行完毕才会继续往下执行。
 
应用二：
微服务注册中心的register-client，为了在注册线程执行成功后，才发送心跳。可以使用CountDownLatch，当然也可以使用Thread.join()方法。
 
应用三：
可以通过CountDownLatch实现类似并发的效果。把CountDownLatch的计数器设置为1，然后让1000个线程调用await()方法。当1000个线程初始化完成后，在main线程调用countDown()让计数器归零。这样这1000个线程就会在一个for()循环中，依次被唤醒。
 
(3)CountDownLatch的例子

1. public class CountDownLatchDemo {
2.     public static void main(String[] args) throws Exception {
3.         final CountDownLatch latch = new CountDownLatch(2);
4.         new Thread() {
5.             public void run() {
6.                 try {
7.                     Thread.sleep(1000);
8.                     System.out.println("线程1开始执行，休眠2秒...");
9.                     Thread.sleep(1000);
11.                     System.out.println("线程1准备执行countDown操作...");
12.                     latch.countDown();
14.                     System.out.println("线程1完成执行countDown操作...");
15.                 } catch (Exception e) {
16.                     e.printStackTrace();
17.                 }
18.             }
19.         }.start();
21.         new Thread() {
22.             public void run() {
23.                 try {
24.                     Thread.sleep(1000);
25.                     System.out.println("线程2开始执行，休眠2秒...");
26.                     Thread.sleep(1000);
28.                     System.out.println("线程2准备执行countDown操作...");
29.                     latch.countDown();
31.                     System.out.println("线程2完成执行countDown操作...");
32.                 } catch (Exception e) {
33.                     e.printStackTrace();
34.                 }
35.             }
36.         }.start();
38.         System.out.println("main线程准备执行countDownLatch的await操作，将会同步阻塞等待...");
39.         latch.await();
40.         System.out.println("所有线程都完成countDown操作，结束同步阻塞等待...");
41.     }
42. }

复制代码

 
2.CountDownLatch.await()方法源码
(1)CountDownLatch.await()方法的阻塞流程
(2)CountDownLatch.await()方法的唤醒流程
(3)CountDownLatch.await()方法的阻塞总结
 
(1)CountDownLatch.await()方法的阻塞流程
CountDownLatch是基于AQS中的共享锁来实现的。从CountDownLatch的构造方法可知，CountDownLatch的count就是AQS的state。
 
调用CountDownLatch的await()方法时，会先调用AQS的acquireSharedInterruptibly()模版方法，然后会调用CountDownLatch的内部类Sync实现的tryAcquireShared()方法。tryAcquireShared()方法会判断state的值是否为0，假如为0，才返回1，否则返回-1。
 
当调用CountDownLatch内部类Sync的tryAcquireShared()方法得到的返回值是-1时，才会调用AQS的doAcquireSharedInterruptibly()方法，将当火线程封装成Node结点加入等待队列，然后挂起当火线程举行阻塞。

1. //A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
2. public class CountDownLatch {
3.     private final Sync sync;
4.    
5.     public CountDownLatch(int count) {
6.         if (count < 0) {
7.             throw new IllegalArgumentException("count < 0");
8.         }
9.         this.sync = new Sync(count);
10.     }
11.    
12.     //Synchronization control For CountDownLatch.
13.     //Uses AQS state to represent count.
14.     private static final class Sync extends AbstractQueuedSynchronizer {
15.         Sync(int count) {
16.             setState(count);
17.         }
18.         
19.         int getCount() {
20.             return getState();
21.         }
22.         
23.         protected int tryAcquireShared(int acquires) {
24.             return (getState() == 0) ? 1 : -1;
25.         }
26.         
27.         protected boolean tryReleaseShared(int releases) {
28.             //Decrement count; signal when transition to zero
29.             for (;;) {
30.                 int c = getState();
31.                 if (c == 0) {
32.                     return false;
33.                 }
34.                 int nextc = c-1;
35.                 if (compareAndSetState(c, nextc)) {
36.                     return nextc == 0;
37.                 }
38.             }
39.         }
40.     }
41.    
42.     //Causes the current thread to wait until the latch has counted down to zero,
43.     //unless the thread is Thread#interrupt interrupted.
44.     public void await() throws InterruptedException {
45.         //执行AQS的acquireSharedInterruptibly()方法
46.         sync.acquireSharedInterruptibly(1);
47.     }
48.     ...
49. }
51. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
52.     ...
53.     //Acquires in shared mode, aborting if interrupted.
54.     //Implemented by first checking interrupt status, then invoking at least once #tryAcquireShared, returning on success.
55.     //Otherwise the thread is queued, possibly repeatedly blocking and unblocking,
56.     //invoking #tryAcquireShared until success or the thread is interrupted.
57.     public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
58.         if (Thread.interrupted()) {
59.             throw new InterruptedException();
60.         }
61.         //执行CountDownLatch的内部类Sync实现的tryAcquireShared()方法，抢占共享锁
62.         if (tryAcquireShared(arg) < 0) {
63.             //执行AQS的doAcquireSharedInterruptibly()方法
64.             doAcquireSharedInterruptibly(arg);
65.         }
66.     }
67.    
68.     //Acquires in shared interruptible mode.
69.     private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
70.         final Node node = addWaiter(Node.SHARED);//封装当前线程为Shared类型的Node结点
71.         boolean failed = true;
72.         try {
73.             //第一次循环r = -1，所以会执行AQS的shouldParkAfterFailedAcquire()方法
74.             //将node结点的有效前驱结点的状态设置为SIGNAL
75.             for (;;) {
76.                 final Node p = node.predecessor();//node结点的前驱结点
77.                 if (p == head) {
78.                     int r = tryAcquireShared(arg);
79.                     if (r >= 0) {
80.                         setHeadAndPropagate(node, r);
81.                         p.next = null; // help GC
82.                         failed = false;
83.                         return;
84.                     }
85.                 }
86.                 //执行shouldParkAfterFailedAcquire()方法设置node结点的前驱结点的状态为SIGNAL
87.                 //执行parkAndCheckInterrupt()方法挂起当前线程
88.                 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {
89.                     throw new InterruptedException();
90.                 }
91.             }
92.         } finally {
93.             if (failed) {
94.                 cancelAcquire(node);
95.             }
96.         }
97.     }
98.    
99.     //Checks and updates status for a node that failed to acquire.
100.     //Returns true if thread should block. This is the main signal control in all acquire loops.
101.     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
102.         int ws = pred.waitStatus;
103.         if (ws == Node.SIGNAL) {
104.             //This node has already set status asking a release to signal it, so it can safely park.
105.             return true;
106.         }
107.         if (ws > 0) {
108.             //Predecessor was cancelled. Skip over predecessors and indicate retry.
109.             do {
110.                 node.prev = pred = pred.prev;
111.             } while (pred.waitStatus > 0);
112.             pred.next = node;
113.         } else {
114.             //waitStatus must be 0 or PROPAGATE.  
115.             //Indicate that we need a signal, but don't park yet.  
116.             //Caller will need to retry to make sure it cannot acquire before parking.
117.             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
118.         }
119.         return false;
120.     }
121.    
122.     //设置头结点和唤醒后续线程
123.     //Sets head of queue, and checks if successor may be waiting in shared mode,
124.     //if so propagating if either propagate > 0 or PROPAGATE status was set.
125.     private void setHeadAndPropagate(Node node, int propagate) {
126.         Node h = head;
127.         setHead(node);//将node结点设置为头结点
128.         if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) {
129.             Node s = node.next;
130.             if (s == null || s.isShared()) {
131.                 doReleaseShared();
132.             }
133.         }
134.     }
135.    
136.     private void setHead(Node node) {
137.         head = node;
138.         node.thread = null;
139.         node.prev = null;
140.     }
141.     ...
142. }

复制代码

(2)CountDownLatch.await()方法的唤醒流程
调用await()方法时，起首会将当火线程封装成Node结点并添加到等待队列中，然后在执行第一次for循环时会设置该Node结点的前驱结点状态为SIGNAL，接着在执行第二次for循环时才会将当火线程举行挂起阻塞。
 
当该线程后续被唤醒时，该线程又会进入下一次for循环。假如该线程对应的node结点的前驱结点是等待队列的头结点且state值已为0，那么就执行AQS的setHeadAndPropagate()方法设置头结点 + 唤醒后续线程。
 
其中setHeadAndPropagate()方法有两个工作(设置头结点 + 唤醒通报)：
工作一：设置当前被唤醒线程对应的结点为头结点
工作二：当满意如下这两个条件的时候需要调用doReleaseShared()方法唤醒后续的线程
条件一：propagate > 0，表现当前是共享锁，需要举行唤醒通报
条件二：s.isShared()判断当前结点为共享模式
 
CountDownLatch的实现中会在以下两个场景调用doReleaseShared()方法：
场景一：state为1时调用的countDown()方法会调用doReleaseShared()方法
场景二：当阻塞的线程被唤醒时，会调用setHeadAndPropagate()方法，进而调用doReleaseShared()方法，这样可以提升唤醒共享结点的速率
 
(3)CountDownLatch.await()方法的阻塞总结
只要state != 0，就会举行如下处理：
一.将当火线程封装成一个Node结点，然后添加到AQS的等待队列中
二.调用LockSupport.park()方法，挂起当火线程
 
3.CountDownLatch.coutDown()方法源码
(1)CountDownLatch.coutDown()的唤醒流程
(2)CountDownLatch.tryReleaseShared()
(3)AQS的doReleaseShared()方法
 
(1)CountDownLatch.coutDown()的唤醒流程
调用CountDownLatch的countDown()方法时，会先调用AQS的releaseShared()模版方法，然后会执行CountDownLatch的内部类Sync实现的tryReleaseShared()方法。
 
假如tryReleaseShared()方法返回true，则执行AQS的doReleaseShared()方法，通过AQS的doReleaseShared()方法唤醒共享锁模式下的等待队列中的线程。

1. //A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
2. public class CountDownLatch {
3.     private final Sync sync;
4.    
5.     public CountDownLatch(int count) {
6.         if (count < 0) {
7.             throw new IllegalArgumentException("count < 0");
8.         }
9.         this.sync = new Sync(count);
10.     }
11.    
12.     //Synchronization control For CountDownLatch.
13.     //Uses AQS state to represent count.
14.     private static final class Sync extends AbstractQueuedSynchronizer {
15.         Sync(int count) {
16.             setState(count);
17.         }
18.         
19.         int getCount() {
20.             return getState();
21.         }
22.         
23.         protected int tryAcquireShared(int acquires) {
24.             return (getState() == 0) ? 1 : -1;
25.         }
26.         
27.         protected boolean tryReleaseShared(int releases) {
28.             //Decrement count; signal when transition to zero
29.             for (;;) {
30.                 int c = getState();
31.                 if (c == 0) {
32.                     return false;
33.                 }
34.                 int nextc = c-1;
35.                 if (compareAndSetState(c, nextc)) {
36.                     return nextc == 0;
37.                 }
38.             }
39.         }
40.     }
41.            
42.     //Decrements the count of the latch, releasing all waiting threads if the count reaches zero.
43.     public void countDown() {
44.         //执行AQS的releaseShared()方法
45.         sync.releaseShared(1);
46.     }
47.     ...
48. }
50. public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {
51.     ...
52.     //Releases in shared mode.  
53.     //Implemented by unblocking one or more threads if #tryReleaseShared returns true.
54.     public final boolean releaseShared(int arg) {
55.         //执行CountDownLatch的内部类Sync实现的tryReleaseShared()方法，释放共享锁
56.         if (tryReleaseShared(arg)) {
57.             //执行AQS的doReleaseShared()方法
58.             doReleaseShared();
59.             return true;
60.         }
61.         return false;
62.     }
63.    
64.     //Release action for shared mode -- signals successor and ensures propagation.
65.     //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal.
66.     private void doReleaseShared() {
67.         for (;;) {
68.             //每次循环时头结点都会发生变化
69.             //因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程
70.             //然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点
71.             Node h = head;//获取最新的头结点
72.             if (h != null && h != tail) {//等待队列中存在挂起线程的结点
73.                 int ws = h.waitStatus;
74.                 if (ws == Node.SIGNAL) {//头结点的状态正常，表示对应的线程可以被唤醒
75.                     if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) {
76.                         continue;//loop to recheck cases
77.                     }
78.                     //唤醒头结点的后继结点
79.                     //唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点
80.                     unparkSuccessor(h);
81.                 } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {
82.                     //如果ws = 0表示初始状态，则修改结点为PROPAGATE状态
83.                     continue;//loop on failed CAS
84.                 }
85.             }
86.             if (h == head) {//判断头结点是否有变化
87.                 break;//loop if head changed
88.             }
89.         }
90.     }
91.    
92.     //Wakes up node's successor, if one exists.
93.     private void unparkSuccessor(Node node) {
94.         int ws = node.waitStatus;
95.         if (ws < 0) {
96.             compareAndSetWaitStatus(node, ws, 0);
97.         }
99.         Node s = node.next;
100.         if (s == null || s.waitStatus > 0) {
101.             s = null;
102.             for (Node t = tail; t != null && t != node; t = t.prev) {
103.                 if (t.waitStatus <= 0) {
104.                     s = t;
105.                 }
106.             }
107.         }
108.         if (s != null) {
109.             LockSupport.unpark(s.thread);
110.         }
111.     }
112.     ...
113. }

复制代码

 
7.Semaphore的令牌释放过程
(1)Semaphore的令牌释放过程
(2)Semaphore的令牌释放本质
 
(1)Semaphore的令牌释放过程
在调用Semaphore的release()方法去释放令牌时：起首会执行AQS的模版方法releaseShared()，然后执行Sync实现的tryReleaseShared()方法来释放锁(累加state值)。假如释放锁成功，则执行AQS的doReleaseShared()方法去唤醒线程。
 
(2)Semaphore的令牌释放本质
Semaphore的release()方法释放令牌的本质就是对state字段举行累加，然后唤醒等待队列头结点的后继结点 + 唤醒通报来唤醒等待的线程。
 
留意：并非一定要执行acquire()方法的线程才能调用release()方法，任意一个线程都可以调用release()方法，也可以通过reducePermits()方法来减少令牌数。
[code]public class Semaphore implements java.io.Serializable {    private final Sync sync;        //Creates a Semaphore with the given number of permits and nonfair fairness setting.    public Semaphore(int permits) {        sync = new NonfairSync(permits);    }        //Releases a permit, returning it to the semaphore.    public void release() {        //执行AQS的模版方法releaseShared()        sync.releaseShared(1);    }        //Synchronization implementation for semaphore.      //Uses AQS state to represent permits. Subclassed into fair and nonfair versions.    abstract static class Sync extends AbstractQueuedSynchronizer {        Sync(int permits) {            //设置state的值为传入的令牌数            setState(permits);        }                //实验释放锁，也就是对state值举行累加        protected final boolean tryReleaseShared(int releases) {            for (;;) {                int current = getState();                int next = current + releases;                if (next < current) {                    throw new Error("Maximum permit count exceeded");                }                if (compareAndSetState(current, next)) {                    return true;                }            }        }        ...    }    ...}public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable {    ...        //Releases in shared mode.      //Implemented by unblocking one or more threads if #tryReleaseShared returns true.    public final boolean releaseShared(int arg) {        //执行Semaphore的内部类Sync实现的tryReleaseShared()方法，释放共享锁        if (tryReleaseShared(arg)) {            //执行AQS的doReleaseShared()方法，唤醒等待队列中的线程            doReleaseShared();            return true;        }        return false;    }        //Release action for shared mode -- signals successor and ensures propagation.     //Note: For exclusive mode, release just amounts to calling unparkSuccessor of head if it needs signal.    private void doReleaseShared() {        for (;;) {            //每次循环时头结点都会发生变化            //因为调用unparkSuccessor()方法会唤醒doAcquireSharedInterruptibly()方法中阻塞的线程            //然后阻塞的线程会在执行setHeadAndPropagate()方法时通过setHead()修改头结点            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                    }                    //唤醒头结点的后继结点                    //唤醒的线程会在doAcquireSharedInterruptibly()方法中执行setHeadAndPropagate()方法修改头结点                    unparkSuccessor(h);                } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) {                    //假如ws = 0表现初始状态，则修改结点为PROPAGATE状态                    continue;//loop on failed CAS                }            }            if (h == head) {//判断头结点是否有变化                break;//loop if head changed            }        }    }        //Wakes up node's successor, if one exists.    private void unparkSuccessor(Node node) {        int ws = node.waitStatus;        if (ws < 0) {            compareAndSetWaitStatus(node, ws, 0);        }        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  0L) {                        nanos = trip.awaitNanos(nanos);                    }                } catch (InterruptedException ie) {                    if (g == generation && ! g.broken) {                        breakBarrier();                        throw ie;                    } else {                        Thread.currentThread().interrupt();                    }                }                if (g.broken) {                    throw new BrokenBarrierException();                }                if (g != generation) {                    return index;                }                if (timed && nanos