一文讓你讀懂Synchronized底層實現(xiàn)，秒殺面試官

本文為死磕Synchronized底層實現(xiàn)第三篇文章，內(nèi)容為輕量級鎖實現(xiàn)。

輕量級鎖并不復(fù)雜，其中很多內(nèi)容在偏向鎖一文中已提及過，與本文內(nèi)容會有部分重疊。

另外輕量級鎖的背景和基本流程在概論中已有講解。強(qiáng)烈建議在看過兩篇文章的基礎(chǔ)下閱讀本文。

本系列文章將對HotSpot的synchronized鎖實現(xiàn)進(jìn)行全面分析，內(nèi)容包括偏向鎖、輕量級鎖、重量級鎖的加鎖、解鎖、鎖升級流程的原理及源碼分析，希望給在研究synchronized路上的同學(xué)一些幫助。

本文分為兩個部分：

1.輕量級鎖獲取流程

2.輕量級鎖釋放流程

本人看的JVM版本是jdk8u。

?

輕量級鎖獲取流程

下面開始輕量級鎖獲取流程分析，代碼在bytecodeInterpreter.cpp#1816。

CASE(_monitorenter):?{
??oop?lockee?=?STACK_OBJECT(-1);
??...
??if?(entry?!=?NULL)?{
???...
???//?上面省略的代碼中如果CAS操作失敗也會調(diào)用到InterpreterRuntime::monitorenter

????//?traditional?lightweight?locking
????if?(!success)?{
??????//?構(gòu)建一個無鎖狀態(tài)的Displaced?Mark?Word
??????markOop?displaced?=?lockee->mark()->set_unlocked();
??????//?設(shè)置到Lock?Record中去
??????entry->lock()->set_displaced_header(displaced);
??????bool?call_vm?=?UseHeavyMonitors;
??????if?(call_vm?||?Atomic::cmpxchg_ptr(entry,?lockee->mark_addr(),?displaced)?!=?displaced)?{
????????//?如果CAS替換不成功，代表鎖對象不是無鎖狀態(tài)，這時候判斷下是不是鎖重入
????????//?Is?it?simple?recursive?case?
????????if?(!call_vm?&&?THREAD->is_lock_owned((address)?displaced->clear_lock_bits()))?{
??????????entry->lock()->set_displaced_header(NULL);
????????}?else?{
??????????//?CAS操作失敗則調(diào)用monitorenter
??????????CALL_VM(InterpreterRuntime::monitorenter(THREAD,?entry),?handle_exception);
????????}
??????}
????}
????UPDATE_PC_AND_TOS_AND_CONTINUE(1,?-1);
??}?else?{
????istate->set_msg(more_monitors);
????UPDATE_PC_AND_RETURN(0);?//?Re-execute
??}
}

如果鎖對象不是偏向模式或已經(jīng)偏向其他線程，則success為false。這時候會構(gòu)建一個無鎖狀態(tài)的mark word設(shè)置到Lock Record中去，我們稱Lock Record中存儲對象mark word的字段叫Displaced Mark Word。

如果當(dāng)前鎖的狀態(tài)不是無鎖狀態(tài)，則CAS失敗。如果這是一次鎖重入，那直接將Lock Record的?Displaced Mark Word設(shè)置為null。

我們看個demo，在該demo中重復(fù)3次獲得鎖，

synchronized(obj){
????synchronized(obj){
????	synchronized(obj){
????	}
????}
}

假設(shè)鎖的狀態(tài)是輕量級鎖，下圖反應(yīng)了mark word和線程棧中Lock Record的狀態(tài)，可以看到右邊線程棧中包含3個指向當(dāng)前鎖對象的Lock Record。其中棧中最高位的Lock Record為第一次獲取鎖時分配的。其Displaced Mark word的值為鎖對象的加鎖前的mark word，之后的鎖重入會在線程棧中分配一個Displaced Mark word為null的Lock Record。

為什么JVM選擇在線程棧中添加Displaced Mark word為null的Lock Record來表示重入計數(shù)呢？首先鎖重入次數(shù)是一定要記錄下來的，因為每次解鎖都需要對應(yīng)一次加鎖，解鎖次數(shù)等于加鎖次數(shù)時，該鎖才真正的被釋放，也就是在解鎖時需要用到說鎖重入次數(shù)的。一個簡單的方案是將鎖重入次數(shù)記錄在對象頭的mark word中，但mark word的大小是有限的，已經(jīng)存放不下該信息了。另一個方案是只創(chuàng)建一個Lock Record并在其中記錄重入次數(shù)，Hotspot沒有這樣做的原因我猜是考慮到效率有影響：每次重入獲得鎖都需要遍歷該線程的棧找到對應(yīng)的Lock Record，然后修改它的值。

所以最終Hotspot選擇每次獲得鎖都添加一個Lock Record來表示鎖的重入。

接下來看看InterpreterRuntime::monitorenter方法

IRT_ENTRY_NO_ASYNC(void,?InterpreterRuntime::monitorenter(JavaThread*?thread,?BasicObjectLock*?elem))
??...
??Handle?h_obj(thread,?elem->obj());
??assert(Universe::heap()->is_in_reserved_or_null(h_obj()),
?????????"must?be?NULL?or?an?object");
??if?(UseBiasedLocking)?{
????//?Retry?fast?entry?if?bias?is?revoked?to?avoid?unnecessary?inflation
????ObjectSynchronizer::fast_enter(h_obj,?elem->lock(),?true,?CHECK);
??}?else?{
????ObjectSynchronizer::slow_enter(h_obj,?elem->lock(),?CHECK);
??}
??...
IRT_END

fast_enter的流程在偏向鎖一文已經(jīng)分析過，如果當(dāng)前是偏向模式且偏向的線程還在使用鎖，那會將鎖的mark word改為輕量級鎖的狀態(tài)，同時會將偏向的線程棧中的Lock Record修改為輕量級鎖對應(yīng)的形式。代碼位置在biasedLocking.cpp#212。

//?線程還存活則遍歷線程棧中所有的Lock?Record
??GrowableArray<MonitorInfo*>*?cached_monitor_info?=?get_or_compute_monitor_info(biased_thread);
??BasicLock*?highest_lock?=?NULL;
??for?(int?i?=?0;?i?<?cached_monitor_info->length();?i++)?{
????MonitorInfo*?mon_info?=?cached_monitor_info->at(i);
????//?如果能找到對應(yīng)的Lock?Record說明偏向的線程還在執(zhí)行同步代碼塊中的代碼
????if?(mon_info->owner()?==?obj)?{
??????...
??????//?需要升級為輕量級鎖，直接修改偏向線程棧中的Lock?Record。為了處理鎖重入的case，在這里將Lock?Record的Displaced?Mark?Word設(shè)置為null，第一個Lock?Record會在下面的代碼中再處理
??????markOop?mark?=?markOopDesc::encode((BasicLock*)?NULL);
??????highest_lock?=?mon_info->lock();
??????highest_lock->set_displaced_header(mark);
????}?else?{
??????...
????}
??}
??if?(highest_lock?!=?NULL)?{
????//?修改第一個Lock?Record為無鎖狀態(tài)，然后將obj的mark?word設(shè)置為指向該Lock?Record的指針
????highest_lock->set_displaced_header(unbiased_prototype);
????obj->release_set_mark(markOopDesc::encode(highest_lock));
????...
??}?else?{
????...
??}

我們看slow_enter的流程。

void?ObjectSynchronizer::slow_enter(Handle?obj,?BasicLock*?lock,?TRAPS)?{
??markOop?mark?=?obj->mark();
??assert(!mark->has_bias_pattern(),?"should?not?see?bias?pattern?here");
??//?如果是無鎖狀態(tài)
??if?(mark->is_neutral())?{
????//設(shè)置Displaced?Mark?Word并替換對象頭的mark?word
????lock->set_displaced_header(mark);
????if?(mark?==?(markOop)?Atomic::cmpxchg_ptr(lock,?obj()->mark_addr(),?mark))?{
??????TEVENT?(slow_enter:?release?stacklock)?;
??????return?;
????}
??}?else
??if?(mark->has_locker()?&&?THREAD->is_lock_owned((address)mark->locker()))?{
????assert(lock?!=?mark->locker(),?"must?not?re-lock?the?same?lock");
????assert(lock?!=?(BasicLock*)obj->mark(),?"don't?relock?with?same?BasicLock");
????//?如果是重入，則設(shè)置Displaced?Mark?Word為null
????lock->set_displaced_header(NULL);
????return;
??}

??...
??//?走到這一步說明已經(jīng)是存在多個線程競爭鎖了?需要膨脹為重量級鎖
??lock->set_displaced_header(markOopDesc::unused_mark());
??ObjectSynchronizer::inflate(THREAD,?obj())->enter(THREAD);
}

輕量級鎖釋放流程

CASE(_monitorexit):?{
??oop?lockee?=?STACK_OBJECT(-1);
??CHECK_NULL(lockee);
??//?derefing's?lockee?ought?to?provoke?implicit?null?check
??//?find?our?monitor?slot
??BasicObjectLock*?limit?=?istate->monitor_base();
??BasicObjectLock*?most_recent?=?(BasicObjectLock*)?istate->stack_base();
??//?從低往高遍歷棧的Lock?Record
??while?(most_recent?!=?limit?)?{
????//?如果Lock?Record關(guān)聯(lián)的是該鎖對象
????if?((most_recent)->obj()?==?lockee)?{
??????BasicLock*?lock?=?most_recent->lock();
??????markOop?header?=?lock->displaced_header();
??????//?釋放Lock?Record
??????most_recent->set_obj(NULL);
??????//?如果是偏向模式，僅僅釋放Lock?Record就好了。否則要走輕量級鎖or重量級鎖的釋放流程
??????if?(!lockee->mark()->has_bias_pattern())?{
????????bool?call_vm?=?UseHeavyMonitors;
????????//?header!=NULL說明不是重入，則需要將Displaced?Mark?Word?CAS到對象頭的Mark?Word
????????if?(header?!=?NULL?||?call_vm)?{
??????????if?(call_vm?||?Atomic::cmpxchg_ptr(header,?lockee->mark_addr(),?lock)?!=?lock)?{
????????????//?CAS失敗或者是重量級鎖則會走到這里，先將obj還原，然后調(diào)用monitorexit方法
????????????most_recent->set_obj(lockee);
????????????CALL_VM(InterpreterRuntime::monitorexit(THREAD,?most_recent),?handle_exception);
??????????}
????????}
??????}
??????//執(zhí)行下一條命令
??????UPDATE_PC_AND_TOS_AND_CONTINUE(1,?-1);
????}
????//處理下一條Lock?Record
????most_recent++;
??}
??//?Need?to?throw?illegal?monitor?state?exception
??CALL_VM(InterpreterRuntime::throw_illegal_monitor_state_exception(THREAD),?handle_exception);
??ShouldNotReachHere();
}

輕量級鎖釋放時需要將Displaced Mark Word替換到對象頭的mark word中。如果CAS失敗或者是重量級鎖則進(jìn)入到InterpreterRuntime::monitorexit方法中。

//%note?monitor_1
IRT_ENTRY_NO_ASYNC(void,?InterpreterRuntime::monitorexit(JavaThread*?thread,?BasicObjectLock*?elem))
?
??Handle?h_obj(thread,?elem->obj());
??...
??ObjectSynchronizer::slow_exit(h_obj(),?elem->lock(),?thread);
??//?Free?entry.?This?must?be?done?here,?since?a?pending?exception?might?be?installed?on
??//釋放Lock?Record
??elem->set_obj(NULL);
??...
IRT_END

monitorexit調(diào)用完slow_exit方法后,就釋放Lock Record。

void?ObjectSynchronizer::slow_exit(oop?object,?BasicLock*?lock,?TRAPS)?{
??fast_exit?(object,?lock,?THREAD)?;
}
void?ObjectSynchronizer::fast_exit(oop?object,?BasicLock*?lock,?TRAPS)?{
??...
??markOop?dhw?=?lock->displaced_header();
??markOop?mark?;
??if?(dhw?==?NULL)?{
?????//?重入鎖，什么也不做
???	?...
?????return?;
??}

??mark?=?object->mark()?;

??//?如果是mark?word==Displaced?Mark?Word即輕量級鎖，CAS替換對象頭的mark?word
??if?(mark?==?(markOop)?lock)?{
?????assert?(dhw->is_neutral(),?"invariant")?;
?????if?((markOop)?Atomic::cmpxchg_ptr?(dhw,?object->mark_addr(),?mark)?==?mark)?{
????????TEVENT?(fast_exit:?release?stacklock)?;
????????return;
?????}
??}
??//走到這里說明是重量級鎖或者解鎖時發(fā)生了競爭，膨脹后調(diào)用重量級鎖的exit方法。
??ObjectSynchronizer::inflate(THREAD,?object)->exit?(true,?THREAD)?;
}

該方法中先判斷是不是輕量級鎖，如果是輕量級鎖則將替換mark word，否則膨脹為重量級鎖并調(diào)用exit方法，相關(guān)邏輯將在重量級鎖的文章中講解。

?