說(shuō)起 Java 的并發(fā)就不得不提到 ReentrantLock,說(shuō)起 ReentrantLock 就不得不說(shuō)到 AQS。下面,我將為大家簡(jiǎn)單地聊聊 Java 并發(fā)中的 ReentrantLock和AQS,剖析一下此二者的源碼。
一.前言
首先在聊ReentrantLock之前,我們需要知道整個(gè)JUC的并發(fā)同步的基石,currrent里面所有的共享變量都是由volatile修飾的,我們知道volatile的語(yǔ)義有2大特點(diǎn),可見(jiàn)性以及防止重排序(內(nèi)存屏障,volatie寫(xiě)與volatile讀)
1、當(dāng)?shù)诙€(gè)操作為volatile寫(xiě)操做時(shí),不管第一個(gè)操作是什么(普通讀寫(xiě)或者volatile讀寫(xiě)),都不能進(jìn)行重排序。這個(gè)規(guī)則確保volatile寫(xiě)之前的所有操作都不會(huì)被重排序到volatile之后;
2、當(dāng)?shù)谝粋€(gè)操作為volatile讀操作時(shí),不管第二個(gè)操作是什么,都不能進(jìn)行重排序。這個(gè)規(guī)則確保volatile讀之后的所有操作都不會(huì)被重排序到volatile之前;
3、當(dāng)?shù)谝粋€(gè)操作是volatile寫(xiě)操作時(shí),第二個(gè)操作是volatile讀操作,不能進(jìn)行重排序。
而cas操作同時(shí)包含了volatile寫(xiě)/讀語(yǔ)義,這二者的完美結(jié)合就組成了current的基石
二.ReentrantLock的基礎(chǔ)用法
1.
public class ReentrantLockText {
public static void main(String[] args) {
Lock lock = new ReentrantLock();
Thread t1 = new Thread(()->{
try {
lock.lock();
System.out.println("t1 start");
TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
System.out.println("t1 end");
} catch (InterruptedException e) {
System.out.println("interrupted!");
} finally {
lock.unlock();
}
});
t1.start();
Thread t2 = new Thread(()->{
try {
//lock.lock();
lock.lockInterruptibly(); //可以對(duì)interrupt()方法做出響應(yīng)
System.out.println("t2 start");
TimeUnit.SECONDS.sleep(5);
System.out.println("t2 end");
} catch (InterruptedException e) {
System.out.println("interrupted!");
} finally {
lock.unlock();
}
});
t2.start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.interrupt(); //打斷線程2的等待
}
}
運(yùn)行結(jié)果
reentrantlock用于替代synchronized
- 需要注意的是,必須要必須要必須要手動(dòng)釋放鎖(重要的事情說(shuō)三遍)
- 使用syn鎖定的話如果遇到異常,jvm會(huì)自動(dòng)釋放鎖,但是lock必須手動(dòng)釋放鎖,因此經(jīng)常在finally中進(jìn)行鎖的釋放
- 使用reentrantlock可以進(jìn)行“嘗試鎖定”tryLock,這樣無(wú)法鎖定,或者在指定時(shí)間內(nèi)無(wú)法鎖定,線程可以決定是否繼續(xù)等待
- 使用ReentrantLock還可以調(diào)用lockInterruptibly方法,可以對(duì)線程interrupt方法做出響應(yīng)
- 在一個(gè)線程等待鎖的過(guò)程中,可以被打斷
2.ReentrantLock還有一個(gè)tryLock(time),可以指定時(shí)間,如果指定時(shí)間內(nèi)沒(méi)有獲得鎖,則放棄,可以通過(guò)其返回值來(lái)決定是否繼續(xù)等待
3.還有就是Condition了(我個(gè)人覺(jué)得這是最靈活的一個(gè)地方)
public class Lock_condition {
public static void main(String[] args) {
char[] aI = "1234567".toCharArray();
char[] aC = "ABCDEFG".toCharArray();
Lock lock = new ReentrantLock();
Condition conditionT1 = lock.newCondition();
Condition conditionT2 = lock.newCondition();
new Thread(()->{
try {
lock.lock();
for(char c : aI) {
System.out.print(c);
conditionT2.signal();
conditionT1.await();
}
conditionT2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}, "t1").start();
new Thread(()->{
try {
lock.lock();
for(char c : aC) {
System.out.print(c);
conditionT1.signal();
conditionT2.await();
}
conditionT1.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}, "t2").start();
}
}
這是condition結(jié)合lock(獨(dú)占鎖)的用法
condition目前只實(shí)現(xiàn)了獨(dú)占鎖,關(guān)于condition的源碼理解,后續(xù)也會(huì)繼續(xù)更新,暫時(shí)我們只需要知道類(lèi)似object的wait與notifiy
三.原理+源碼
我們現(xiàn)在知道了基本用法,那么我們就可以開(kāi)始探究源碼了
1.AQS
我們知道JUC里面的核心類(lèi)就是AQS,那么AQS究竟是個(gè)啥東西呢?
1)先上內(nèi)部類(lèi)NODE源碼
static final class Node {
static final Node SHARED = new Node();
static final Node EXCLUSIVE = null;
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int PROPAGATE = -3;
volatile int waitStatus;
volatile Node next;
volatile Node prev;
volatile Thread thread;
Node nextWaiter;
不知道大家在看到這個(gè) Node next;Node prev;的時(shí)候是啥感覺(jué),反正我當(dāng)時(shí)是激動(dòng)壞了,這不就是一個(gè)雙向鏈表嘛,
再看volatile Thread thread;這個(gè)屬性,這是一個(gè)管理線程的雙向鏈表,換句話說(shuō)就是將線程打包成立節(jié)點(diǎn)放入AQS的鏈表中
基礎(chǔ)的結(jié)構(gòu)清楚之后。
SHARED 與EXCLUSIVE 代表是獨(dú)占節(jié)點(diǎn)還是共享節(jié)點(diǎn)
2)再上AQS屬性源碼
private transient volatile Node head;
private transient volatile Node tail;
private volatile int state;
Node0 head與tail不用說(shuō),這是來(lái)管理節(jié)點(diǎn)的,
這里我們要核心介紹一個(gè)屬性是state,這也是AQS這個(gè)類(lèi)的靈魂,
1.再獨(dú)占鎖中這個(gè)state是1或者0,(如果大于1則表示鎖重入,這個(gè)稍后會(huì)有源碼分析)
2在共享鎖中代表還有多少共享鎖資源,
3.在讀寫(xiě)鎖中,高16位代表寫(xiě)鎖是否被占用,低16位代表有多少讀鎖,
4.在CountDownLatch中,通過(guò)構(gòu)造參數(shù)代碼門(mén)閘剩余個(gè)數(shù)
5.在Semaphore中,同樣通過(guò)構(gòu)造參數(shù)代表信號(hào)燈個(gè)數(shù)
2.ReentrantLock獲取鎖源碼(獨(dú)占鎖)
首先公平鎖與非公平鎖,分別繼承與Sync
FairSync NoFairSync ,默認(rèn)是非公平鎖,可以在構(gòu)造方法上指定
ReentrantLock lock = new ReentrantLock(true);//ture則是公平鎖
公平鎖故名思意,在AQS中管理著一個(gè)線程隊(duì)列,如果這時(shí)候 有一個(gè)線程過(guò)來(lái)?yè)屵@把鎖,如果是公平鎖,那么會(huì)判斷隊(duì)列是不是存在不同與當(dāng)前線程的等待隊(duì)列(FIFO),如果存在則去排隊(duì),非公平鎖則是直接去排隊(duì)
(2.1.非公平鎖獲取鎖)
final void lock() {
if (compareAndSetState(0, 1))//cas原子操作嘗試去修改值,如果修改成功說(shuō)明成功獲取到了鎖
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
首先cas原子操作嘗試去修改值,如果修改成功說(shuō)明成功獲取到了鎖,進(jìn)入setExclusiveOwnerThread()方法
protected final void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread = thread;
}
將當(dāng)前線程記錄,實(shí)現(xiàn)偏向鎖,一行代碼便完美實(shí)現(xiàn)了偏向鎖!!
如果失敗則,調(diào)用acquire();這個(gè)方法調(diào)用的實(shí)際上是子類(lèi)的nonfairTryAcquire方法
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;
}
首先,獲取state的值,判斷是否為0,如果為0,則說(shuō)明鎖沒(méi)有被占有,(可能是剛剛被釋放)那么cas操作開(kāi)始嘗試獲取鎖,
**(注意注意注意)**重要的事情說(shuō)三遍,這里僅僅嘗試獲取一次,沒(méi)有自旋!!這是獨(dú)占鎖與共享鎖的區(qū)別之一,因?yàn)槿绻鹲tate>=0(對(duì)于共享鎖來(lái)說(shuō)state代表剩余數(shù)量),那么共享鎖會(huì)不斷嘗試自旋獲取鎖,之道state<0,因?yàn)橹灰?那么就可能共享到鎖
接下來(lái)的else if就是重入鎖的操作了,判斷當(dāng)先線程是不是記錄的線程,如果是,每次重入state+1,如果不是就返回flase,直接拜拜
(2.2公平鎖獲取鎖)
剛剛介紹了公平鎖的意義,所有直接上源碼,公平鎖比非公平鎖多了一個(gè)公平判斷
public final boolean hasQueuedPredecessors() {
// The correctness of this depends on head being initialized
// before tail and on head.next being accurate if the current
// thread is first in queue.
Node t = tail; // Read fields in reverse initialization order
Node h = head;
Node s;
return h != t &&
((s = h.next) == null || s.thread != Thread.currentThread());
}
我們可以看到hasQueuedPredecessors是用來(lái)判斷隊(duì)列是否有不同于當(dāng)前線程的節(jié)點(diǎn)等待,這里重點(diǎn)討論一個(gè)情況,
h != t返回true,(s = h.next) == null返回true
首先可以知道隊(duì)列中有2個(gè)節(jié)點(diǎn),但是頭節(jié)點(diǎn)沒(méi)有后繼結(jié)點(diǎn),在這里列舉一種情況,有另一個(gè)線程已經(jīng)執(zhí)行到初始化隊(duì)列的操作了,介于compareAndSetHead(new Node())與tail = head之間,也就是之后說(shuō)的enq自旋方法,請(qǐng)繼續(xù)往下看
繼續(xù)如果非公平鎖沒(méi)有獲取到鎖,那么會(huì)調(diào)用acquireQueued和addwaiter方法
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
首先來(lái)看addwaiter方法
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;
}
首先將,當(dāng)前線程封裝成一個(gè),Node,然后通過(guò)判斷尾結(jié)點(diǎn)是不是空的方式,判斷隊(duì)列是不是空的,如果存在尾結(jié)點(diǎn),那么直接進(jìn)先驅(qū)后繼的改造,放入雙向鏈表,完成鏈表結(jié)構(gòu),那么如果是空的呢?這么調(diào)用了一個(gè)enq的自旋方法
private Node enq(final Node node) {
for (;;) {
Node t = tail;
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
我們可以看到這個(gè)一個(gè)自旋方法,第一次循環(huán):t為null,那么cas就new出來(lái)一個(gè)新結(jié)點(diǎn),頭尾都指向這個(gè)新結(jié)點(diǎn),
第二次循環(huán),將傳進(jìn)來(lái)的這個(gè)線程結(jié)點(diǎn)的前驅(qū)指向剛剛new的這個(gè)結(jié)點(diǎn),然后cas操作進(jìn)行,將這個(gè)線程結(jié)點(diǎn)替換為尾部結(jié)點(diǎn),然后head后繼指向線程結(jié)點(diǎn),返回head
經(jīng)過(guò)二次循環(huán),得到了一個(gè)由2個(gè)節(jié)點(diǎn)組成的隊(duì)列,head-》node,head是假節(jié)點(diǎn)(里面不包括線程為null),node才是真正的線程結(jié)點(diǎn)(addwrite封裝好的傳進(jìn)來(lái)的線程結(jié)點(diǎn))
問(wèn)題1.為什么一定要用cas操作,因?yàn)榉乐箘e的線程修改了該隊(duì)列
好,現(xiàn)在我們繼續(xù),看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);
}
}
這是一個(gè)最最核心的方法,從隊(duì)列中取線程
首先這是一個(gè)自旋,判斷該節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)是不是head,因?yàn)椋‵IFO)先進(jìn)先出隊(duì)列。
如果不是直接拜拜進(jìn)入shouldParkAfterFailedAcquire,繼續(xù)上源碼
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
static final int CANCELLED = 1;
static final int SIGNAL = -1;
static final int CONDITION = -2;
static final int PROPAGATE = -3;
- CANCELLED:因?yàn)槌瑫r(shí)或者中斷,結(jié)點(diǎn)會(huì)被設(shè)置為取消狀態(tài),被取消狀態(tài)的結(jié)點(diǎn)不應(yīng)該去競(jìng)爭(zhēng)鎖,只能保持取消狀態(tài)不變,不能轉(zhuǎn)換為其他狀態(tài)。處于這種狀態(tài)的結(jié)點(diǎn)會(huì)被踢出隊(duì)列,被GC回收;
- SIGNAL:表示這個(gè)結(jié)點(diǎn)的繼任結(jié)點(diǎn)被阻塞了,到時(shí)需要通知它;
- CONDITION:表示這個(gè)結(jié)點(diǎn)在條件隊(duì)列中,因?yàn)榈却硞€(gè)條件而被阻塞;
- PROPAGATE:使用在共享模式頭結(jié)點(diǎn)有可能牌處于這種狀態(tài),表示鎖的下一次獲取可以無(wú)條件傳播;
- 0:None of the above,新結(jié)點(diǎn)會(huì)處于這種狀態(tài)。
首先說(shuō)明一下waitStatus這個(gè)屬性,為什么之前不提呢,因?yàn)橹皼](méi)有對(duì)waitStatus進(jìn)行操作,我們?cè)趎ew節(jié)點(diǎn)與封裝結(jié)點(diǎn)的時(shí)候沒(méi)有考慮這個(gè)屬性,所以我們現(xiàn)在當(dāng)成一個(gè)新屬性,值為0來(lái)看待,
shouldParkAfterFailedAcquire這個(gè)代碼的邏輯意義是說(shuō)明呢?
如果是SIGNAL那么,直接ture,
如果大于0,則是CANCELLED,被取消了,直接剔除隊(duì)列
如果都不是,那么將其前驅(qū)結(jié)點(diǎn)設(shè)為SIGNAL,也就是可以安心睡覺(jué)了,定好鬧鐘了,可以被等著喚醒了,
我們現(xiàn)在很明顯是第三種,因?yàn)橹吧抖紱](méi)干,就是0,
所以本來(lái)狀態(tài)
進(jìn)行shouldParkAfterFailedAcquire之后,
那么現(xiàn)在鏈表中,對(duì)線程1的前驅(qū)設(shè)鬧鐘,0變成-1
假設(shè)這時(shí)候又來(lái)了個(gè)線程2,那么同理,對(duì)線程而的先驅(qū)設(shè)鬧鐘0變成-1
在調(diào)用了shouldParkAfterFailedAcquire()之后,調(diào)用parkAndCheckInterrupt方法用于阻塞,
這里提一下,關(guān)于parkAndCheckInterrupt,lock里面用于阻塞都是基于lockSupper.park()與lockSupper.unpark()完成了,而lockSupper調(diào)用的又是unsafe這個(gè)類(lèi),我們知道java是基于jvm實(shí)現(xiàn)的,并不能和c++一樣直接對(duì)os進(jìn)行操作,所以jvm給我們提供了一個(gè)梯子,這個(gè)梯子就是unsafe這個(gè)類(lèi),直接將線程的的交個(gè)操作系統(tǒng)阻塞
四,釋放鎖
終于到了釋放鎖,獨(dú)占鎖的釋放鎖的邏輯相對(duì)與共享鎖來(lái)說(shuō)比較簡(jiǎn)單,后續(xù)我也會(huì)繼續(xù)更新共享鎖的源碼
public final boolean release(int arg) {
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
首先我們來(lái)看tryRelease方法,
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
首先獲取,int c = getState() - releases;這里可能c是>0的,因?yàn)楠?dú)占鎖的重入鎖(上面以及說(shuō)明了獨(dú)占鎖的重入的源碼操作),所以有可能是需要進(jìn)行多次解鎖的,繼續(xù)
判斷當(dāng)前線程是不是獨(dú)占線程,如果不是則報(bào)IllegalMonitorStateException異常
一直解鎖到c=0的時(shí)候,那么線程已經(jīng)解鎖,則設(shè)setExclusiveOwnerThread=null
設(shè)置當(dāng)前獨(dú)占線程為null,然后設(shè)置state為0
繼續(xù)回到release,如果頭節(jié)點(diǎn)不是null而且h.waitStatus != 0,說(shuō)明是-1,說(shuō)明設(shè)置鬧鐘了,需要喚醒a(bǔ)qs隊(duì)列中的阻塞結(jié)點(diǎn),調(diào)用的是unparkSuccessor方法,繼續(xù)看源碼
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
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 <= 0)
s = t;
}
if (s != null)
LockSupport.unpark(s.thread);
}
這里首先明確,這里傳進(jìn)來(lái)的node是啥?是頭節(jié)點(diǎn)?。。?!,一定要明確這個(gè),博主就是因?yàn)閯傞_(kāi)始沒(méi)明確這個(gè),半天沒(méi)明白,因?yàn)閱拘岩欢ㄊ菃拘杨^節(jié)點(diǎn)之后的waitStatus不為1的結(jié)點(diǎn)
首先判斷頭節(jié)點(diǎn),如果是-1則設(shè)置為0,這個(gè)中間狀態(tài),表示有結(jié)點(diǎn)被喚醒了,
然后拿到,head的后繼節(jié)點(diǎn),進(jìn)行判斷,如何是null或者waitStatus >0,(就是1,CANCELLED,代表被取消了),這個(gè)時(shí)候從尾部開(kāi)始遍歷,剔除waitStatus >0的節(jié)點(diǎn),找到第一個(gè)waitStatus <=0的節(jié)點(diǎn),用LockSupport.unpark(s.thread);將其喚醒,喚醒之后的線程回到acquireQueued方法中,
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
//被阻塞的線程,被喚醒后在進(jìn)行循環(huán),
//然后通過(guò)return interrupted;
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);
}
}
首先,將該結(jié)點(diǎn)設(shè)置為head,然后將老head指向null,幫助gc回收,然后return返回,至此線程自由
那么此時(shí)該隊(duì)列為
五,總結(jié)
寫(xiě)了很久,如果有啥不對(duì)的地方,歡迎大家指正
以上就是 Java 并發(fā)中的 ReentrantLock 和 AQS 源碼的詳細(xì)內(nèi)容,想要了解更多關(guān)于 Java 并發(fā)的 ReentrantLock 和 AQS 的其他資料請(qǐng)關(guān)注W3Cschool其它相關(guān)文章!