前言：大部分多線程同步場(chǎng)景以躯，在功能和性能層面蟆盐，synchronized可以滿足移必，少部分場(chǎng)景Lock可以滿足室谚，dubbo的源碼也符合這個(gè)比例，需要使用到Condition的場(chǎng)景極少避凝，整個(gè)dubbo源碼中只在啟動(dòng)函數(shù)中舞萄，服務(wù)關(guān)閉這一處使用到了Lock+Condition機(jī)制眨补。
1.Lock+Condition用法
生產(chǎn)者管削，消費(fèi)者模式在面試coding中出場(chǎng)率很高，可以用synchronized+wait+ notify來(lái)實(shí)現(xiàn)撑螺，也可以使用Lock+Condition實(shí)現(xiàn)含思。直接上代碼
public class LockConditionTest {
private LinkedList<String> queue=new LinkedList<String>();
private Lock lock = new ReentrantLock();
private int maxSize = 5;
private Condition providerCondition = lock.newCondition();
private Condition consumerCondition = lock.newCondition();
public void provide(String value){
try {
lock.lock();
while (queue.size() == maxSize) {
providerCondition.await();
}
queue.add(value);
consumerCondition.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
public String consume(){
String result = null;
try {
lock.lock();
while (queue.size() == 0) {
consumerCondition.await();
}
result = queue.poll();
providerCondition.signal();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
return result;
}
public static void main(String[] args) {
LockConditionTest t = new LockConditionTest();
new Thread(new Provider(t)).start();
new Thread(new Consumer(t)).start();
}
}
以兩個(gè)問(wèn)題驅(qū)動(dòng)
1.隊(duì)列滿了，生產(chǎn)者線程怎么停下來(lái)的？隊(duì)列從滿又變?yōu)椴粷M的時(shí)候含潘，怎么重新激活饲做。
2.隊(duì)列空了，消費(fèi)者線程如何停下來(lái)遏弱，又如何重新開(kāi)始消費(fèi)盆均。
一步一步解答這兩個(gè)問(wèn)題
2.ReentrantLock
ReentrantLock初始化的時(shí)候，默認(rèn)是初始化一個(gè)NonfairSync漱逸。
public ReentrantLock() {
sync = new NonfairSync();
}
核心代碼在AbstractQueuedSynchronizer中,只看數(shù)據(jù)結(jié)構(gòu)的話泪姨，這是一個(gè)基于Node,帶頭指針和尾指針的雙向鏈表，每一個(gè)Node里面存一個(gè)線程饰抒，以及該線程的等待狀態(tài)
static final class Node {
volatile int waitStatus;
volatile Node prev;
volatile Node next;
volatile Thread thread;
Node nextWaiter;
}
private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;
那么肮砾，ReentrantLock在lock和unlock的時(shí)候，操作的就是這個(gè)雙向鏈表sync queue袋坑。
先看獲取鎖的過(guò)程
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c + acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}
1.如果這個(gè)鎖沒(méi)有任何線程持有仗处，那么當(dāng)前線程直接可以獲得。（這是非公平鎖的設(shè)計(jì)枣宫，如果是公平鎖婆誓，需要檢查是否有線程在排隊(duì)，如果有也颤，當(dāng)前線程不能直接搶占旷档，也要加入排隊(duì)。）
2.如果這個(gè)鎖被占用了歇拆，占用線程是當(dāng)前線程鞋屈，那么state+1，這也是可重入鎖之所以可以重入的由來(lái)故觅。
3.都不滿足厂庇，暫時(shí)獲取鎖失敗，返回false
那么如果在3這一步獲取鎖失敗输吏，后續(xù)如何處理呢权旷？
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
1.addWaiter：在等待隊(duì)列sync queue中添加一個(gè)節(jié)點(diǎn)
2.acquireQueued：節(jié)點(diǎn)自旋獲取鎖或者進(jìn)入阻塞
addWaiter比較簡(jiǎn)單，即把當(dāng)前等待線程加入sync queue的尾節(jié)點(diǎn)贯溅。
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.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
acquireQueued具體做了什么呢拄氯？
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
failed = false;
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
if (failed)
cancelAcquire(node);
}
}
1.自旋
2.如果當(dāng)前就一個(gè)線程在等待，那么嘗試獲取鎖它浅。（判斷條件：當(dāng)前節(jié)點(diǎn)的前驅(qū)為head译柏，即head.next = 當(dāng)前節(jié)點(diǎn)）
3.不滿足2，如果滿足進(jìn)入阻塞的條件姐霍，調(diào)用LockSupport.park(this)進(jìn)入阻塞鄙麦。
一句話總結(jié)lock的過(guò)程：當(dāng)前線程直接去嘗試獲取鎖典唇，不成功，則加入sync queue尾節(jié)點(diǎn)進(jìn)行阻塞等待（非公平）胯府。
在看unlock的過(guò)程
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
1.先釋放當(dāng)前線程占有的鎖介衔，核心就是維護(hù)state的值。加一次鎖骂因，state+1炎咖，釋放反之。
2.unparkSuccessor ：之前提到寒波，lock的時(shí)候塘装，會(huì)維護(hù)一個(gè)排隊(duì)的雙向隊(duì)列sync queue，此時(shí)影所，會(huì)unpark喚醒其中的head.next蹦肴，使其進(jìn)入鎖競(jìng)爭(zhēng)狀態(tài)。
注：公平鎖猴娩，非公平鎖加鎖的過(guò)程小有區(qū)別阴幌，一個(gè)是搶占式的，一個(gè)是排隊(duì)式的卷中，釋放鎖的過(guò)程則是完全相同的矛双。
3.Condition
先看類圖
用過(guò)Object的wait，notify的對(duì)這些方法應(yīng)該不陌生蟆豫，對(duì)應(yīng)這里的await和signal
先看await
public final void await() throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}
1.構(gòu)造一個(gè)Node议忽，形成一個(gè)單向鏈表condition queue，存儲(chǔ)用于await在某一個(gè)condition上的線程十减。
2.釋放當(dāng)前Node持有的鎖栈幸。這個(gè)釋放過(guò)程跟之前提到的unlock過(guò)程類似。
3.LockSupport.park進(jìn)行阻塞帮辟，等待signal的喚醒速址。
4.如果被喚醒，繼續(xù)加入鎖的競(jìng)爭(zhēng)中去由驹。
在看signal
public final void signal() {
if (!isHeldExclusively())
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if (first != null)
doSignal(first);
}
在某個(gè)condition進(jìn)行await的時(shí)候芍锚，形成了一個(gè)單向鏈表condition queue。
那么在signal的時(shí)候蔓榄，先對(duì)頭結(jié)點(diǎn)firstWaiter進(jìn)行喚醒并炮。
喚醒的過(guò)程見(jiàn)下
final boolean transferForSignal(Node node) {
/*
* If cannot change waitStatus, the node has been cancelled.
*/
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;
/*
* Splice onto queue and try to set waitStatus of predecessor to
* indicate that thread is (probably) waiting. If cancelled or
* attempt to set waitStatus fails, wake up to resync (in which
* case the waitStatus can be transiently and harmlessly wrong).
*/
Node p = enq(node);
int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
LockSupport.unpark(node.thread);
return true;
}
1.將這個(gè)頭結(jié)點(diǎn)，從condition queue中移到之前提到的sync queue中甥郑。
2.LockSupport.unpark喚醒這個(gè)節(jié)點(diǎn)中的線程逃魄，進(jìn)行鎖爭(zhēng)奪。
4 總結(jié)
ReentrantLock lock依賴的是一個(gè)雙向鏈表sync queue
condition依賴的是一個(gè)單項(xiàng)鏈表condition queue
二者的阻塞和喚醒依賴的都是LockSupport的park和unpark方法壹若。
公平鎖與非公平鎖的區(qū)別就在于獲取鎖的方式不同嗅钻，公平鎖獲取，當(dāng)前線程必須檢查sync queue里面是否已經(jīng)有排隊(duì)線程店展。而非公平鎖則不用考慮這一點(diǎn)养篓，當(dāng)前線程可以直接去獲取鎖。
開(kāi)篇實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模型的時(shí)候赂蕴，有兩個(gè)問(wèn)題柳弄，現(xiàn)在有答案了
1.隊(duì)列滿了，生產(chǎn)者線程怎么停下來(lái)的概说？隊(duì)列從滿又變?yōu)椴粷M的時(shí)候碧注，怎么重新激活。
---通過(guò)lock機(jī)制糖赔，保證同一時(shí)刻萍丐，只有一個(gè)線程獲取到鎖，要么生產(chǎn)放典，要么消費(fèi)逝变，隊(duì)列滿了之后，生產(chǎn)者線程調(diào)用providerCondition.await()奋构，進(jìn)入阻塞等待狀態(tài)壳影，使得生產(chǎn)者線程停下來(lái)。當(dāng)消費(fèi)線程消費(fèi)的時(shí)候弥臼，調(diào)用 providerCondition.signal()宴咧，重新激活生產(chǎn)者線程。
2.隊(duì)列空了径缅，消費(fèi)者線程如何停下來(lái)掺栅，又如何重新開(kāi)始消費(fèi)。
---與第一個(gè)問(wèn)題同理纳猪。
