学习自http://www.importnew.com/21889.html
https://blog.csdn.net/yanyan19880509/article/details/52349056
https://www.cnblogs.com/waterystone/p/4920797.html
|
1
2
3
|
public void await() throws InterruptedException { }; //调用await()方法的线程会被挂起，它会等待直到count值为0才继续执行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException { }; //和await()类似，只不过等待一定的时间后count值还没变为0的话就会继续执行
public void countDown() { }; //将count值减1
|
下面看一个例子大家就清楚CountDownLatch的用法了：
|
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
|
public class Test {
public static void main(String[] args) {
final CountDownLatch latch = new CountDownLatch(2);
new Thread(){
public void run() {
try {
System.out.println("子线程"+Thread.currentThread().getName()+"正在执行");
Thread.sleep(3000);
System.out.println("子线程"+Thread.currentThread().getName()+"执行完毕");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
};
}.start();
new Thread(){
public void run() {
try {
System.out.println("子线程"+Thread.currentThread().getName()+"正在执行");
Thread.sleep(3000);
System.out.println("子线程"+Thread.currentThread().getName()+"执行完毕");
latch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
};
}.start();
try {
System.out.println("等待2个子线程执行完毕...");
latch.await();
System.out.println("2个子线程已经执行完毕");
System.out.println("继续执行主线程");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
} |
执行结果：
|
1
2
3
4
5
6
7
|
线程Thread-0正在执行
线程Thread-1正在执行
等待2个子线程执行完毕...
线程Thread-0执行完毕
线程Thread-1执行完毕
2个子线程已经执行完毕
继续执行主线程 |
在new CountDownLatch(2)的时候
对于 CountDownLatch 来说，state=2表示所有调用await方法的线程都应该阻塞，等到同一个latch被调用两次countDown后才能唤醒沉睡的线程
为了充分了解AQS里的链表，这里假设上面挂起等待的线程数为2个
当latch被成功减到0后，AQS的state就成了0。那个成功减到0的那个线程。然后节点3被唤醒了。当节点3醒来后，发现自己是通知状态，然后删除自己，唤醒节点4。
上面的流程，如果落实到代码，把 state置为0的那个线程，会判断head指向节点的状态，如果为通知状态，则唤醒后续节点，即线程3节点，然后head指向线程3节点，head指向的旧节点会被删除掉。当线程3恢复执行后，发现自身为通知状态，又会把head指向线程4节点，然后删除自身节点，并唤醒线程4。
线程节点的状态是什么时候设置上去的？其实，一个线程在阻塞之前，就会把它前面的节点设置为通知状态，这样便可以实现链式唤醒机制了。
AQS是一些同步的抽象，简单介绍一下
两大元素
1.volatile int state->共享资源
2.FIFO线程等待队列，多线程争用资源被阻塞时会进入此队列
AQS定义两种资源占用方式：
1.Exclusive（独占，只有一个线程能执行，如ReentrantLock）
2.Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）
自定义同步器在实现时只需要实现state的增减即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS已经在底层实现好了
自定义同步器的时候，主要实现以下几种方法：
1.isHeldExclusively()，该线程是否正在独占资源，只有用到condition的时候才需要去实现这个方法
2.tryAcquire(int)，独占。尝试获取资源，成功返回true，失败返回fasle
3.tryRelease(int)
4.tryAcquireShared(int)，共享。尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
5.tryReleaseShared(int)
以ReentrantLock为例
state初始化为0，表示未锁定状态。A线程lock()时，会调用tryAcquire()独占该锁并将state+1。此后，其他线程再tryAcquire()时就会失败，直到A线程unlock()到state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A线程自己是可以重复获取此锁的（state会累加），这就是可重入的概念。但要注意，获取多少次就要释放多么次，这样才能保证state是能回到零态的。
以CountDownLatch以例
任务分为N个子线程去执行，state也初始化为N（注意N要与线程个数一致）。这N个子线程是并行执行的，每个子线程执行完后countDown()一次，state会CAS减1。等到所有子线程都执行完后(即state=0)，会unpark()主调用线程，然后主调用线程就会从await()函数返回，继续后余动作。
深入看一下CountDownLatch
首先是他的内部类
private static final class Sync extends AbstractQueuedSynchronizer { private static final long serialVersionUID = 4982264981922014374L; Sync(int count) { setState(count); } int getCount() { return getState(); } protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c - 1; if (compareAndSetState(c, nextc)) return nextc == 0; } } } private final Sync sync;
tryAcquireShared不知道有啥用，理论上应该是来获取资源的，但是对于CountDownLatch不需要了，因为他只需要一开始设定state即可。不过CountDownLatch为什么是共享锁我理解了，他允许多个线程同时使用CPU资源。
Node.waitStatus
-CANCELLED：值为1，在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该Node的结点，其结点的waitStatus为CANCELLED，即结束状态，进入该状态后的结点将不会再变化。
-SIGNAL：值为-1，被标识为该等待唤醒状态的后继结点，当其前继结点的线程释放了同步锁或被取消，将会通知该后继结点的线程执行。说白了，就是处于唤醒状态，只要前继结点释放锁，就会通知标识为SIGNAL状态的后继结点的线程执行。
-CONDITION：值为-2，与Condition相关，该标识的结点处于等待队列中，结点的线程等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
-PROPAGATE：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。
0状态：值为0，代表初始化状态。
AQS在判断状态时，通过用waitStatus>0表示取消状态，而waitStatus<0表示有效状态。
CountDownLatch的-1方法
public void countDown() { sync.releaseShared(1); }
AQS中
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }
前面讲到：releaseShared就是让state-1，当你减到0的时候，减到0的那个线程开始唤醒线程。
这里和代码逻辑对应一下。
tryReleaseShared就是让state-1。如果减到0，就返回true了，开始执行doReleaseShared，唤醒线程。
private void doReleaseShared() { for (;;) { Node h = head;//头结点 if (h != null && h != tail) {//如果不为空，且不是尾节点 int ws = h.waitStatus; if (ws == Node.SIGNAL) {//如果是待唤醒，waitStatus改为初始化状态 if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h);//找到他后面的第一个节点，也就是允许使用cpu，同时解锁了 } else if (ws == 0 &&//如果是初始化状态 希望把他变成可运行状态 !h.compareAndSetWaitStatus(0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head)//如果head变了，就继续loop // loop if head changed break; } }
所以doReleaseShared的整体意思是：唤醒下一个线程
unparkSuccessor
private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) node.compareAndSetWaitStatus(ws, 0);//如果是可运行、可唤醒，那就置为初始化状态 Node s = node.next; if (s == null || s.waitStatus > 0) { s = null;//如果为空或者cancel for (Node p = tail; p != node && p != null; p = p.prev)//从后往前找 if (p.waitStatus <= 0)//目的是为了找到一个可运行、可唤醒的 s = p; } if (s != null)//如果找到了 LockSupport.unpark(s.thread);//解锁这个线程 }
整体的意思就是，找到他后面的第一个节点，也就是允许使用cpu，同时解锁了
那么被唤醒的线程是在哪里唤醒下一个线程的呢？
需要从await入手
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0)//如果state!=0，之前初始化已经让state=2了 doAcquireSharedInterruptibly(arg); }
private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED);//尾部加一个node try { for (;;) { final Node p = node.predecessor();//前一个node if (p == head) {//是head int r = tryAcquireShared(arg); if (r >= 0) {//state==0 setHeadAndPropagate(node, r);//那就开始，并唤醒下一个线程 p.next = null; // help GC return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())//锁住咯，并让上一个线程置为通知状态 throw new InterruptedException(); } } catch (Throwable t) { cancelAcquire(node); throw t; } }
CountDownLatch就分析的差不多了
为何CountDownLatch被称为共享锁？
因为CountDownLatch latch = new CountDownLatch(2)，他允许了多个线程同时使用CPU资源，而ReentrantLock是单行道。在这个实例化中，锁可以认为有2道（state），每个线程执行完后，通过cas方式正确地减了一道。
再讲一下CyclicBarrier、 Semaphore，不过就不扯源码了
CyclicBarrier
可以实现让一组线程等待至某个状态之后再全部同时执行
例子
续
|
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
|
public class Test {
public static void main(String[] args) {
int N = 4;
CyclicBarrier barrier = new CyclicBarrier(N);
for(int i=0;i<N;i++)
new Writer(barrier).start();
}
static class Writer extends Thread{
private CyclicBarrier cyclicBarrier;
public Writer(CyclicBarrier cyclicBarrier) {
this.cyclicBarrier = cyclicBarrier;
}
@Override
public void run() {
System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
try {
Thread.sleep(5000); //以睡眠来模拟写入数据操作
System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕，等待其他线程写入完毕");
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
}catch(BrokenBarrierException e){
e.printStackTrace();
}
System.out.println("所有线程写入完毕，继续处理其他任务...");
}
}
} |
执行结果：
|
1
2
3
4
5
6
7
8
9
10
11
12
|
线程Thread-0正在写入数据...
线程Thread-3正在写入数据...
线程Thread-2正在写入数据...
线程Thread-1正在写入数据...
线程Thread-2写入数据完毕，等待其他线程写入完毕
线程Thread-0写入数据完毕，等待其他线程写入完毕
线程Thread-3写入数据完毕，等待其他线程写入完毕
线程Thread-1写入数据完毕，等待其他线程写入完毕
所有线程写入完毕，继续处理其他任务...所有线程写入完毕，继续处理其他任务...所有线程写入完毕，继续处理其他任务...所有线程写入完毕，继续处理其他任务... |
总结CountDownLatch、CyclicBarrie的区别
前者是手动-1，另外线程先阻塞着。state减到0后，另外线程可以跑了
后者是若干线程先阻塞着，等到阻塞到若干数量后，所有线程就可以跑了
Semaphore
若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。那么我们就可以通过Semaphore来实现：
|
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
|
public class Test {
public static void main(String[] args) {
int N = 8; //工人数
Semaphore semaphore = new Semaphore(5); //机器数目
for(int i=0;i<N;i++)
new Worker(i,semaphore).start();
}
static class Worker extends Thread{
private int num;
private Semaphore semaphore;
public Worker(int num,Semaphore semaphore){
this.num = num;
this.semaphore = semaphore;
}
@Override
public void run() {
try {
semaphore.acquire();
System.out.println("工人"+this.num+"占用一个机器在生产...");
Thread.sleep(2000);
System.out.println("工人"+this.num+"释放出机器");
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
} |
执行结果：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
工人0占用一个机器在生产...
工人1占用一个机器在生产...
工人2占用一个机器在生产...
工人4占用一个机器在生产...
工人5占用一个机器在生产...
工人0释放出机器
工人2释放出机器
工人3占用一个机器在生产...
工人7占用一个机器在生产...
工人4释放出机器
工人5释放出机器
工人1释放出机器
工人6占用一个机器在生产...
工人3释放出机器
工人7释放出机器
工人6释放出机器
|
其实到这里都明白了，state就代表了共享资源池，在不同的并发器里有不同的意义。Semaphore的意义是资源，我可以acquire1个，也可以acquire多个。
state在CountDownLatch中的意义是，当前资源有几个人在使用，直到0个人使用，锁才能解除；
state在CyclicBarrier中的意义是当前资源几个人已经用完了，直到x个人用完了，锁才能解除。
AQS就不继续深究了，本文叙述其共享实现就够了