示例小demo
public class reentrantlockTest {
private ReentrantLock lock=new ReentrantLock();
public void test1(){
try {
System.out.println("test1開始等待獲取鎖");
lock.lock();
System.out.println("test1已經(jīng)獲取鎖");
Thread.sleep(4000);
}catch (InterruptedException e) {
Thread.interrupted();
} finally {
lock.unlock();
}
}
public void test2(){
try {
System.out.println("test2開始等待獲取鎖");
lock.lock();
System.out.println("test2開始已經(jīng)獲取鎖");
}finally {
lock.unlock();
}
}
public static void main(String[] args) {
final reentrantlockTest test= new reentrantlockTest();
new Thread(new Runnable() {
@Override
public void run() {
test.test1();
}
},"t1").start();
new Thread(new Runnable() {
@Override
public void run() {
test.test2();
}
},"t2").start();
}
test1開始等待獲取鎖
test1已經(jīng)獲取鎖
test2開始等待獲取鎖
test2開始已經(jīng)獲取鎖
用起來(lái)很簡(jiǎn)單lock,unlock就可以了歌粥。當(dāng)多個(gè)線程同時(shí)要獲取這個(gè)鎖時(shí)候到底發(fā)生了什么?
簡(jiǎn)單描述
在具體分析源碼之前,先用語(yǔ)言簡(jiǎn)單描述一下.
aqs隊(duì)列簡(jiǎn)單圖示,源自網(wǎng)絡(luò)
有一位足療店技師活特別好,大家去消費(fèi)都想點(diǎn)他服務(wù),但是他只是一個(gè)人,所有來(lái)的人得排隊(duì)等他服務(wù),開始服務(wù)就相當(dāng)于加鎖,服務(wù)結(jié)束相當(dāng)于釋放鎖資源监右。
之后第一個(gè)人去叫第二個(gè)人進(jìn)來(lái)服務(wù),不斷循環(huán),直到都所有人都消費(fèi)完成.
還有個(gè)概念
公平鎖:大家都老老實(shí)實(shí)排隊(duì),先到的先被服務(wù).
非公平鎖：在第一個(gè)人叫第二個(gè)人的這段時(shí)間內(nèi),來(lái)消費(fèi)就先進(jìn)屋看看,還沒人進(jìn)來(lái),就直接讓技師服務(wù)(就是搶個(gè)時(shí)間差,直接插隊(duì),不過這樣就節(jié)省了第一個(gè)人叫第二人的時(shí)間,效率高些).
那ReentrantLock和Synchronize鎖有什么區(qū)別
先說相同點(diǎn)：兩者都是可以非公平鎖.在最新的jdk版本下效率差不多
不同點(diǎn)：
Synchronize:使用簡(jiǎn)單一些,非特定場(chǎng)景可以用
ReentrantLock：可以實(shí)現(xiàn)公平鎖,可以在等資源時(shí)候可以中斷.
那么AQS在里面起什么作用？
隊(duì)列怎么排,怎么入隊(duì),出隊(duì),資源消息是怎么傳遞的.
開始源碼分析
從lock開始分析
public void lock() {
sync.lock();
}
final void lock() {
state是一個(gè)表示鎖資源狀態(tài)的參數(shù)
1.上來(lái)就直接插隊(duì),用CAS算法看能不能獲取鎖資源
if (compareAndSetState(0, 1))
獲取成功后,當(dāng)前線程獲得資源
setExclusiveOwnerThread(Thread.currentThread());
else
2.如果獲取不成功,正常排隊(duì)等待
acquire(1);
}
lock默認(rèn)采用非公平鎖,上來(lái)嘗試獲取鎖,獲取不到進(jìn)入aqs隊(duì)列排隊(duì)
acquire()
public final void acquire(int arg) {
1.嘗試獲取鎖資源,直接拋出異常,必須由子類定義什么才是獲取鎖
2.如果獲取不到鎖,創(chuàng)建一個(gè)新的隊(duì)列節(jié)點(diǎn),將節(jié)點(diǎn)放到隊(duì)列里面
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
3.拿到鎖之后,如果該線程是中斷狀態(tài),直接中斷線程
selfInterrupt();
}
入隊(duì)前在嘗試一次獲取鎖,如果獲取不到,添加一個(gè)節(jié)點(diǎn),把節(jié)點(diǎn)放到隊(duì)列里排隊(duì),阻塞知道被前節(jié)點(diǎn)unpark通知,恢復(fù)執(zhí)行后檢查是否需要中斷
tryAcquire()
protected final boolean tryAcquire(int acquires) {
return nonfairTryAcquire(acquires);
}
nonfairTryAcquire(acquires);
================================
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
1.獲取鎖的狀態(tài)
int c = getState();
2.如果沒有線程持有鎖
if (c == 0) {
3.嘗試自己獲取鎖
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
4.如果是當(dāng)前已經(jīng)是該線程持有鎖,那么將state計(jì)數(shù)器+1.這里就體現(xiàn)了 可重入的特性
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,表示鎖已經(jīng)被占有了,只有cas（0到1）是才獲取成功,這樣保證了獨(dú)占模式。如果當(dāng)前是該線程占有這個(gè)鎖,那么state+1,unlock時(shí)-1,重入性就是這么來(lái)的.
可以看到這里繼承了aqs之后,自由的實(shí)現(xiàn)鎖的獲取方式,concurrent包里面不少類通過實(shí)現(xiàn)不同的鎖的獲取,來(lái)實(shí)現(xiàn)不同的特性.后續(xù)博文中會(huì)陸續(xù)介紹
addWaiter(Node.EXCLUSIVE)
這里添加了一個(gè)獨(dú)占模式的節(jié)點(diǎn)
先看addWaiter(Node.EXCLUSIVE)
添加一個(gè)等待節(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
1.如果存在尾部節(jié)點(diǎn),將新節(jié)點(diǎn)鏈接在其后,并且將新節(jié)點(diǎn)設(shè)置為尾部節(jié)點(diǎn)
Node pred = tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
2.有可能隊(duì)列為空,則直接入隊(duì)
enq(node);
return node;
}
=====================
private Node enq(final Node node) {
for (;;) {
1.如果尾部為空
Node t = tail;
if (t == null) { // Must initialize
2.先初始化一個(gè)空節(jié)點(diǎn),將其設(shè)置為 頭，尾 然后for循環(huán)
if (compareAndSetHead(new Node()))
tail = head;
} else {
3.到這里肯定已經(jīng)有尾部節(jié)點(diǎn)了,將我們的節(jié)點(diǎn)加在尾部
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
4.最后將節(jié)點(diǎn)返回
return t;
}
}
}
}
這里就是初始化隊(duì)列(實(shí)際是雙向鏈表)熟吏，頭尾都是傀儡節(jié)點(diǎn),將節(jié)點(diǎn)鏈接到鏈表尾部
現(xiàn)在有了一個(gè)節(jié)點(diǎn),那么就開始入隊(duì)了
acquireQueued()
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
1.獲取當(dāng)前節(jié)點(diǎn)的前一個(gè)節(jié)點(diǎn)
final Node p = node.predecessor();
2.如果前節(jié)點(diǎn)是頭節(jié)點(diǎn),嘗試一次獲取鎖
if (p == head && tryAcquire(arg)) {
3.如果獲得了鎖,把該節(jié)點(diǎn)設(shè)置為頭節(jié)點(diǎn),將其設(shè)置為傀儡節(jié)點(diǎn)
setHead(node);
4.原頭節(jié)點(diǎn)后面的鏈表置為空,// help GC
p.next = null; // help GC
failed = false;
return interrupted;
}
5.如果獲取鎖失敗,那么就需要掛起該線程,等待通知
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
interrupted = true;
}
} finally {
6.如果失敗,做失敗處理
if (failed)
7.后面進(jìn)行分析
cancelAcquire(node);
}
}
shouldParkAfterFailedAcquire()
=============
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
1.這里又多了一個(gè)狀態(tài),waitStatus表示節(jié)點(diǎn)的等待狀態(tài)
2.先獲取前節(jié)點(diǎn)狀態(tài)
int ws = pred.waitStatus;
3.如果前節(jié)點(diǎn)處于通知狀態(tài),意味當(dāng)前節(jié)點(diǎn)可以嘗試去獲取鎖了
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
4.如果前節(jié)點(diǎn)是cancel狀態(tài)
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
5.那就不斷遍歷找到不是取消狀態(tài)的節(jié)點(diǎn)
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.
*/
6.如果頭節(jié)點(diǎn)是新創(chuàng)建的狀態(tài)是0(PROPAGATE以后在討論),（這里設(shè)置為0是因?yàn)閁nlock時(shí)候處理的,具體細(xì)節(jié)看后面）那么將他設(shè)置為通知狀態(tài)
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
parkAndCheckInterrupt()
===============
private final boolean
1.走到了這里說明,該線程是可以掛起的狀態(tài)了
parkAndCheckInterrupt() {
2.這里用了park(park(),unpark(),這一對(duì)方法很有意思。wait,notify都應(yīng)該知道,先wait,在sign才可以生效要不然就卡死了.這里park,unPark,作用是類似的,但是完全不需要順序,可以先unpark,在Park.具體寫個(gè)小demo就可以了解了)
所以就避免了這種情況,head已經(jīng)執(zhí)行完了,也unpark了,但是當(dāng)前線程還沒有執(zhí)行到掛起的地方,造成卡死
最后線程阻塞到這里,等到前節(jié)點(diǎn)通知
LockSupport.park(this);
3.返回線程是否中斷了（被自己或其他中斷了）
只有當(dāng)節(jié)點(diǎn)被喚起后才能設(shè)置中斷狀態(tài)
return Thread.interrupted();
}
上面入隊(duì)之后進(jìn)行阻塞,直到接到前節(jié)點(diǎn)發(fā)到的信號(hào).
以上就是lock的實(shí)現(xiàn)流程,下面在看看unlock的操作
public void unlock() {
sync.release(1);
}
public final boolean release(int arg) {
1.嘗試釋放鎖資源
if (tryRelease(arg)) {
Node h = head;
if (h != null && h.waitStatus != 0)
unparkSuccessor(h);
return true;
}
return false;
}
tryRelease(arg) 同tryAcquire一樣需要子類自定義獲取鎖的方式
====================================
protected boolean tryRelease(int arg) {
throw new UnsupportedOperationException();
}
子類的具體實(shí)現(xiàn)
protected final boolean tryRelease(int releases) {
1.鎖當(dāng)前狀態(tài)-1
int c = getState() - releases;
if (Thread.currentThread() != getExclusiveOwnerThread())
throw new IllegalMonitorStateException();
boolean free = false;
2.如果狀態(tài)等于0了,那么這個(gè)鎖就被成功的釋放了
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
3.如果減了一次還不為0,那么當(dāng)前這個(gè)線程多了lock了這個(gè)鎖,這里也是可重入導(dǎo)致的
setState(c);
return free;
}
==========================================
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;
1.將頭狀態(tài)設(shè)置為0
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.
*/
2.找到后繼節(jié)點(diǎn),并且沒有取消,通知該節(jié)點(diǎn)恢復(fù) 這也是為什么必須要在finaly里unlock
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);
}
以上是源碼解析
最后留一些思考
1.ReentrantLock獨(dú)占性,可重入性是怎么實(shí)現(xiàn)的
2.AQS在里面起什么作用玄窝？
3.公平性和非公平性是怎么實(shí)現(xiàn)的
4.park(),unPark()
