学习自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资源。
CountDownLatch的-1方法
public void countDown() { sync.releaseShared(1); }
AQS中
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false; }
sync
cas当前值,减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; } }
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表示有效状态。
aqs-do release
private void doReleaseShared() { for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) {//如果是等待唤醒状态 if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))//期望把它更新到初始化状态 continue; // loop to recheck cases unparkSuccessor(h);//唤醒等待队列里的下一个线程 } else if (ws == 0 && !h.compareAndSetWaitStatus(0, Node.PROPAGATE)) continue;//如果是初始化状态,就把它改成可运行状态 // loop on failed CAS } if (h == head) // loop if head changed break; } }
releaseShared整个方法理解起来就是先把state的值降1,如果降到0了,执行doReleaseShared()方法。
回顾这个图
所以这个方法就是把前面的state个数的线程执行完了以后,再执行我们的这些等待通知的线程。显然图中的线程3就是处于等待唤醒状态。然后把它更新成初始化状态,然后再唤醒等待队列里的下一个线程。然后再把自身转化成可执行转化成可运行状态。
未完待续