Java 多線程學習（6）synchronized 的成神之路

時間  2020-02-12
標籤 java 多線程 學習 synchronized 之路 欄目 Java 简体版
原文   原文鏈接

轉載請註明原創出處，謝謝！html

HappyFeet的博客java

前段時間把 Object.await() 和 Object.notify()、LockSupport.park() 和 LockSupport.unpark() 差很少理解了，不過在跟蹤 C++ 源碼的時候看到了 ObjectMonitor.cpp 這個類（在 await() 釋放鎖和被 notify() 後從新獲取鎖的時候會調用這部分代碼）；在看這裏面的實現的時候，發現，咦？怎麼它內部的邏輯和 synchronized 的特色如此之像！後來發現原來這裏面就有 synchronized 底層實現。node

之前經過文檔學習了 synchronized 的用法及特色，可是並無瞭解它的原理，現在終於知道了它爲何有這些特色了！c++

不只如此，在這個過程當中還把以前學到的一些知識點（對象的內存佈局中的 Mark Word 那一部分）串了起來，有種恍然大悟的感受！web

下面就來對 synchronized 關鍵字作一個整理，加深記憶。編程

1、基礎篇

一、synchronized 怎麼用？

/** * @author happyfeet * @since Jan 19, 2020 */
public class SynchronizedUsageExample {

    private static int staticCount;
    private final Object lock = new Object();
    private int count;

    public static synchronized int synchronizedOnStaticMethod() {
        return staticCount++;
    }

    public synchronized void synchronizedOnInstanceMethod() {
        count++;
    }

    public int synchronizedOnCodeBlock() {
        synchronized (lock) {
            return count++;
        }
    }
}
複製代碼

很簡單的一個例子，不過涵蓋了 synchronized 的全部用法：安全

  • （1）做用於靜態方法
  • （2）做用於實例方法
  • （3）做用於代碼塊

二、synchronized 有哪些特色？

  • 老是須要一個對象做爲鎖

    • 做用於靜態方法時鎖是 class 對象bash

    • 做用於實例方法時鎖是當前實例對象，即 thisoracle

    • 做用於代碼塊時鎖是括號裏的對象，即上例中的 lock 對象app

  • 同一把鎖同一時刻只能被一個線程所持有，不一樣鎖之間互不影響

當多個線程訪問同步代碼塊時，須要先獲取鎖，有且僅有一個線程能夠獲取到鎖，沒有獲取到鎖的線程會被阻塞，直到獲取到鎖；不一樣鎖之間互不影響。

咱們來看個例子：

/** * @author happyfeet * @since Feb 03, 2020 */
public class SynchronizedLockExample {

    private final Object lock = new Object();

    public static void main(String[] args) {

        SynchronizedLockExample lockExample = new SynchronizedLockExample();

        Thread thread1 = new Thread(() -> lockExample.synchronizedOnCodeBlock(), "thread-1");

        Thread thread2 = new Thread(() -> lockExample.synchronizedOnCodeBlock(), "thread-2");

        Thread thread3 = new Thread(() -> lockExample.synchronizedOnInstanceMethod(), "thread-3");

        Thread thread4 = new Thread(() -> lockExample.synchronizedOnInstanceMethod(), "thread-4");

        sleepOneSecond();
        thread1.start();
        sleepOneSecond();
        thread2.start();
        sleepOneSecond();
        thread3.start();
        sleepOneSecond();
        thread4.start();

        while (true) {

        }
    }

    private synchronized void synchronizedOnInstanceMethod() {
        println("I'm in synchronizedOnInstanceMethod, thread name is {}.", Thread.currentThread().getName());
        while (true) {
            // do something
        }
    }

    private void synchronizedOnCodeBlock() {
        synchronized (lock) {
            println("I'm in synchronizedOnCodeBlock, thread name is {}.", Thread.currentThread().getName());
            while (true) {
                // do something
            }
        }
    }

    private static void sleepOneSecond() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
複製代碼

執行 main 函數，程序輸出以下：

2020-02-03T10:19:12.993 thread thread-1 : I'm in synchronizedOnCodeBlock, thread name is thread-1.
2020-02-03T10:19:14.984 thread thread-3 : I'm in synchronizedOnInstanceMethod, thread name is thread-3.
複製代碼

Dump Threads，查看各個線程的運行狀態，如圖所示：

synchronized 學習 - 圖一

代碼簡介

很是簡單的一個例子，兩個方法：synchronizedOnInstanceMethod 和 synchronizedOnCodeBlock（兩個方法內部均爲死循環，模擬正在處理業務邏輯，線程會一直持有鎖）；

其中，synchronizedOnInstanceMethod 方法以 this 對象做爲鎖，synchronizedOnCodeBlock 以 lock 對象做爲鎖，而後啓動了 4 個線程調用這兩個方法：thread-1 和 thread-2 調用 synchronizedOnCodeBlock 方法，thread-3 和 thread-4 調用 synchronizedOnInstanceMethod 方法。

結果分析

從程序輸出能夠看到，thread-1 和 thread-3 都進入到了同步代碼塊內，處於 RUNNABLE 狀態；因爲 lock 和 this 鎖分別被 thread-1 和 thread-3 所持有，thread-2 和 thread-4 沒法獲取到相應的鎖，因而都處於 BLOCKED 狀態。仔細看上面的截圖能夠發現，thread-4 正在等待獲取 this 對象鎖（一樣的，thread-2 正在等待獲取 lock 對象鎖，不過截圖中沒有體現出來）：

- waiting to lock <0x000000076acf00f0> (a com.yhh.example.concurrency.SynchronizedLockExample)
複製代碼

一些拓展

同一個類中，全部被 synchronized 修飾的靜態方法使用的鎖都是 class 對象；全部被 synchronized 修飾的實例方法使用的鎖都是 this 對象；因此，對於這種：

private synchronized void method1() {
        // do something
    }

    private synchronized void method2() {
        // do something
    }

    private synchronized void method3() {
        // do something
    }
複製代碼

其實都是公用 this 對象這一把鎖，若是 this 鎖被線程 A 經過某個方法好比 method1() 將鎖持有，那麼，其餘線程調用 method1()、method2()、method3() 時都會被阻塞；靜態方法也是如此。

這裏的核心點是：同一把鎖同一時刻只能被一個線程所持有。

  • 可重入

synchronized 是可重入鎖，可重入鎖指的是在一個線程中能夠屢次獲取同一把鎖。

可重入鎖最大的做用就是避免死鎖。能夠參考這裏面的回答：java的可重入鎖用在哪些場合？

2、高級篇

三、從字節碼的角度看 synchronized

以 SynchronizedUsageExample 爲例，反編譯獲得字節碼以下（省略了一些不重要的字節碼）：

Classfile /Users/happyfeet/projects/java/sampleJava/src/main/java/com/yhh/example/concurrency/SynchronizedUsageExample.class
  ... 
{
  public static synchronized int synchronizedOnStaticMethod();
    descriptor: ()I
    flags: ACC_PUBLIC, ACC_STATIC, ACC_SYNCHRONIZED // 同步標記
    Code:
      stack=3, locals=0, args_size=0
         0: getstatic     #4                  // Field staticCount:I
         3: dup
         4: iconst_1
         5: iadd
         6: putstatic     #4                  // Field staticCount:I
         9: ireturn
      LineNumberTable:
        line 14: 0

  public synchronized void synchronizedOnInstanceMethod();
    descriptor: ()V
    flags: ACC_PUBLIC, ACC_SYNCHRONIZED // 同步標記
    Code:
      stack=3, locals=1, args_size=1
         0: aload_0
         1: dup
         2: getfield      #5                  // Field count:I
         5: iconst_1
         6: iadd
         7: putfield      #5                  // Field count:I
        10: return
      LineNumberTable:
        line 18: 0
        line 19: 10

  public int synchronizedOnCodeBlock();
    descriptor: ()I
    flags: ACC_PUBLIC
    Code:
      stack=4, locals=3, args_size=1
         0: aload_0
         1: getfield      #3                  // Field lock:Ljava/lang/Object;
         4: dup
         5: astore_1
         6: monitorenter // enter the monitor
         7: aload_0
         8: dup
         9: getfield      #5                  // Field count:I
        12: dup_x1
        13: iconst_1
        14: iadd
        15: putfield      #5                  // Field count:I
        18: aload_1
        19: monitorexit // exit the monitor（正常退出）
        20: ireturn
        21: astore_2
        22: aload_1
        23: monitorexit // exit the monitor（出現異常時走的邏輯）
        24: aload_2
        25: athrow
      Exception table:
         from    to  target type
             7    20    21   any
            21    24    21   any
      LineNumberTable:
        line 22: 0
					...
}
SourceFile: "SynchronizedUsageExample.java"
複製代碼

將有 synchronized 修飾的方法與普通方法對比，能夠獲得：

  • 有 synchronized 修飾的方法的 flags 中多了一個同步標記：ACC_SYNCHRONIZED

  • 有 synchronized 修飾的代碼塊中出現了成對的 monitorenter 和 monitorexit（注意：上面字節碼中第二個 monitorexit 是處理異常，說明在執行第一個 monitorexit 以前程序出現了異常）

四、ACC_SYNCHRONIZED、monitorenter 和 monitorexit

這幾個的做用，須要從 JVM 規範中去找，主要是這幾個地方：

JVM 的同步機制是經過 monitor （監視器鎖）的進入和退出實現的。這裏是 JVM 所定義的規範，不只僅是 Java 語言，其餘能夠運行於 JVM 上的語言一樣適用，例如：Scala。

對於 Java 語言來講，比較經常使用的同步實現就是使用 synchronized 關鍵字，做用於方法和代碼塊上。

（1）同步標記：ACC_SYNCHRONIZED

當 synchronized 做用於方法上時，反編譯成字節碼以後會發現方法的 flags 中會有一個同步標記：ACC_SYNCHRONIZED。

對於有 ACC_SYNCHRONIZED 標記的方法，線程在執行方法時的步驟是這樣的：

  • enters the monitor
  • invokes the method itself
  • exits the monitor whether the method invocation completes normally or abruptly
（2）monitorenter 和 monitorexit 指令

當 synchronized 做用於代碼塊時，能夠發現字節碼中出現了成對的 monitorenter 和 monitorexit 指令。

咱們來看看 JVM 規範中是怎麼描述這兩個指令的：

做用

獲取對象的 monitor

參數

objectref，必須是對象引用，若是爲 null 則拋出 NPE

描述

每一個對象與一個 monitor 關聯，monitor 只有在擁有 owner 的狀況下才被鎖定。當一個線程執行到 monitorenter 指令時會嘗試獲取對象鎖所關聯的 monitor，獲取規則以下：

  • 若是 monitor 的 entry count 爲0，則該線程能夠進入 monitor，並將 monitor 的 entry count 的值設爲 1，該線程成爲 monitor 的 owner；
  • 若是當前線程已經擁有該 monitor，只是從新進入（reenter），則將 monitor 的 entry count 的值加 1；
  • 若是 monitor 的 owner 是其餘線程，則當前線程進入阻塞狀態，直到 monitor 的 entry count 爲 0 以後再次嘗試獲取 monitor。

做用

釋放對象的 monitor

參數

objectref，必須是對象引用，若是爲 null 則拋出 NPE

描述

執行 monitorexit 的線程必須是 objectref 關聯的 monitor 的 owner，若是不是則拋出 IllegalMonitorStateException；

執行 monitorexit 的效果就是將 objectref 關聯的 monitor 的 entry count 的值減 1，當 entry count 的值變爲 0 時，當前線程將釋放對象的 monitor，再也不是該 monitor 的 owner。

到了這，對於 synchronized 的使用基本上是沒有什麼問題了；不過對於 synchronized 的執行過程尚未一個完整的認識，下面咱們來看一看 synchronized 在 HotSpot 中是如何執行的？

3、成神篇

五、偏向鎖、輕量級鎖和重量級鎖？

咱們先來看看什麼是偏向鎖、輕量級鎖和重量級鎖？後續講 synchronized 在 HotSpot 中的執行過程當中會講到這些鎖是如何獲取的。

（1）偏向鎖

偏向鎖是 JDK 1.6 加入的，其目的是消除數據在無競爭狀況下的同步原語，進一步提升程序的性能。

若是開啓了偏向鎖，鎖會偏向於第一個得到它的線程，若是後續的執行過程該鎖沒有被其餘線程獲取，則持有偏向鎖的線程將永遠不須要再進行同步；

而一旦出現多個線程競爭同一把鎖，就必須撤銷偏向鎖，撤銷完以後再執行輕量級鎖的同步操做，此時就會帶來額外的鎖撤銷的消耗。

偏向鎖能夠提升帶有同步但無競爭的程序性能，但若是程序中大多數鎖老是被多個不一樣的線程訪問，那偏向模式就是多餘的；

JDK 1.6 默認開啓，能夠經過參數 -XX:-UseBiasedLocking 禁用偏向鎖。

偏向鎖是直接修改 Mark Word 中的 lock（標誌位）和 biased_lock（是否偏向鎖），偏向線程 ID 也直接存儲於鎖對象的 Mark Word 中。

（2）輕量級鎖

輕量級鎖也是 JDK 1.6 加入的，其目的是在線程交替執行的狀況下，減小傳統的重量級鎖使用操做系統互斥量產生的性能消耗。

輕量級鎖會在當前線程的棧幀中創建一個名爲 Lock Record 的空間，用於存儲鎖對象目前的 Mark Word 的拷貝，而後經過 CAS 嘗試將對象的 Mark Word 更新爲指向 Lock Record 的指針。若是更新成功，說明這個線程擁有了該對象鎖，同時對象 Mark Word 的鎖標誌位轉變爲 「00」，代表此對象處於輕量級鎖定狀態。

當兩個以上線程競爭同一個鎖，那麼輕量級鎖再也不有效，須要膨脹爲重量級鎖。此時除了互斥量的開銷外，還有額外的輕量級鎖的加鎖和解鎖操做，所以，在有競爭的狀況下，輕量級鎖會比傳統的重量級鎖更慢。

既然如此，那爲何還須要有輕量級鎖呢？由於 「對於絕大部分鎖，在整個同步週期內都是不存在競爭的」，這是一個經驗數據。

（3）重量級鎖

重量級鎖是經過對象內部的監視器（monitor）實現，鎖對象的 Mark Word 指向一個類型爲 ObjectMonitor 的對象指針；monitor 的本質是依賴於底層操做系統的 Mutex Lock 實現，操做系統實現線程同步時，須要進行系統調用，須要從用戶態（User Mode）到內核態（Kernel Mode）的切換，代價相對較高。

三種鎖的適用場景以下：

  • 偏向鎖：適用於只有一個線程獲取鎖
  • 輕量級鎖：適用於多個線程交替獲取鎖，不存在競爭
  • 重量級鎖：適用於多個線程競爭同一個鎖

六、synchronized 在 HotSpot 中的執行過程

這裏主要目的是對 synchronized 的執行過程有一個大概的認識，因此省略了不少代碼及註釋（註釋其實很重要，不過爲了篇幅起見，都去掉了），只保留了一些關鍵的 C++ 源碼，如需查看完整 C++ 源碼能夠點擊代碼片斷後面的連接查看。

synchronized 的 HotSpot 實現依賴於對象頭的 Mark Word，關於 Mark Word 能夠參考：深刻理解Java虛擬機之----對象的內存佈局 中的 Mark Word 模塊，這是其中一張關於 64 位系統的 Mark Word 示例圖片：

HotSpot 虛擬機對象頭 Mark Word

對於 synchronized 方法會多一個 ACC_SYNCHRONIZED 同步標記，線程在執行方法時的步驟爲：

enters the monitor => invoke method => exits the monitor

這裏對於 enters the monitor 具體作了啥不是很清楚，不過我猜測應該是和 monitorenter 指令的行爲是一致的；一樣的，exits the monitor 與 monitorexit 指令一致。

能夠參考：Java synchronized是否有對應的字節碼指令實現？ 和 R 大的讀書筆記 第146頁 Threads and Synchronization - The Java bytecode implementation

如下爲本身的一些理解（可能會有不對的地方），僅做參考。

（1）templateTable_x86_64.cpp#monitorenter：openjdk/hotspot/src/cpu/x86/vm/templateTable_x86_64.cpp

synchronized 代碼塊的字節碼會出現成對的 monitorenter 和 monitorexit 指令，那麼這兩個指令是如何發揮做用的呢？

參考 R 大讀書筆記 第 232 頁 7.2.1 Interpreter 模塊 及 JVM 之模板解釋器

JVM 在執行 monitorenter 指令的入口爲：

void TemplateTable::monitorenter() {
  
	...
    
  // store object
  __ movptr(Address(c_rarg1, BasicObjectLock::obj_offset_in_bytes()), rax);
  __ lock_object(c_rarg1);
  
  ...
    
  // The bcp has already been incremented. Just need to dispatch to
  // next instruction.
  __ dispatch_next(vtos);
}
複製代碼

點擊查看源碼

這一段主要是彙編代碼，很晦澀，太難懂。

實話實說，我是沒看懂這一段代碼在幹嗎，一臉懵逼，直接進入下一個方法：lock_object。

（2）interp_masm_x86_64.cpp#lock_object：openjdk/hotspot/src/cpu/x86/vm/interp_masm_x86_64.cpp
void InterpreterMacroAssembler::lock_object(Register lock_reg) {
 assert(lock_reg == c_rarg1, "The argument is only for looks. It must be c_rarg1");

 if (UseHeavyMonitors) {
   call_VM(noreg,
           CAST_FROM_FN_PTR(address, InterpreterRuntime::monitorenter),
           lock_reg);
 } else {
   Label done;

   ...

   if (UseBiasedLocking) {
     biased_locking_enter(lock_reg, obj_reg, swap_reg, rscratch1, false, done, &slow_case);
   }

   ...

   // Call the runtime routine for slow case
   call_VM(noreg,
           CAST_FROM_FN_PTR(address, InterpreterRuntime::monitorenter),
           lock_reg);

   bind(done);
 }
}
複製代碼

點擊查看源碼

這裏也一樣基本上沒看懂，有幾個關鍵字眼能夠推敲一番：UseHeavyMonitors、UseBiasedLocking，進入下一個方法：InterpreterRuntime::monitorenter

（3）interpreterRuntime.cpp#monitorenter：openjdk/hotspot/src/share/vm/interpreter/interpreterRuntime.cpp
IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorenter(JavaThread* thread, BasicObjectLock* elem))
	...
  Handle h_obj(thread, elem->obj());
  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
複製代碼

點擊查看源碼

BasicObjectLock 類型的 elem 對象包含一個 BasicLock 類型的 _lock 對象和一個指向 Object 對象的指針 _obj；

class BasicObjectLock VALUE_OBJ_CLASS_SPEC {
  friend class VMStructs;
 private:
  BasicLock _lock;                                    // the lock, must be double word aligned
  oop       _obj;                                     // object holds the lock;

 public:
  // Manipulation
  oop obj() const { return _obj;  }
  void set_obj(oop obj) { _obj = obj; }
  BasicLock* lock() { return &_lock; }

  ...
};
複製代碼

點擊查看源碼

  • 這裏的 Object 對象就是做爲鎖的對象；
  • BasicLock 類型 _lock 對象主要用來保存 _obj 指向的 Object 對象的對象頭數據；
class BasicLock VALUE_OBJ_CLASS_SPEC {
  friend class VMStructs;
 private:
  volatile markOop _displaced_header;
 public:
  markOop displaced_header() const { return _displaced_header; }
  void set_displaced_header(markOop header) { _displaced_header = header; }

  ...
};
複製代碼

點擊查看源碼

UseBiasedLocking：標識 JVM 是否開啓偏向鎖功能，若是開啓則執行 fast_enter 邏輯，不然執行 slow_enter；

（4）synchronizer.cpp#fast_enter：openjdk/hotspot/src/share/vm/runtime/synchronizer.cpp
void ObjectSynchronizer::fast_enter(Handle obj, BasicLock* lock, bool attempt_rebias, TRAPS) {
 if (UseBiasedLocking) {
    if (!SafepointSynchronize::is_at_safepoint()) {
      BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);
      if (cond == BiasedLocking::BIAS_REVOKED_AND_REBIASED) {
        return;
      }
    } else {
      assert(!attempt_rebias, "can not rebias toward VM thread");
      BiasedLocking::revoke_at_safepoint(obj);
    }
    assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");
 }

 slow_enter (obj, lock, THREAD) ;
}
複製代碼

點擊查看源碼

偏向鎖的獲取

BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);

這個方法作的事情就是獲取偏向鎖。點進去一看，又是一大塊代碼。。。並且還不是很容易理解。。。

參考 《深刻理解 JVM 虛擬機》這本書和戰小狼的這篇博客 JVM源碼分析之synchronized實現 加以理解，大體邏輯爲：

（a）獲取鎖對象的對象頭中的 Mark Word，判斷是否是可偏向狀態；

假設當前虛擬機啓用了偏向鎖（啓用參數：-XX:+UseBiasedLocking，這是 JDK 1.6 的默認值），那麼，當鎖對象第一次被線程獲取的時候，虛擬機就會把對象頭中的標誌位設爲 「01」，便可偏向狀態。

（b）判斷 Mark Word 中的偏向線程 ID：若是爲空，則進入步驟（c）；若是指向當前線程，說明當前線程是偏向鎖的持有者，能夠不須要同步，直接執行同步塊代碼；若是指向其它線程，進入步驟（d）；

（c）經過 CAS 設置 Mark Word 中的偏向線程 ID 爲當前線程 ID，若是設置成功，當前線程成爲了偏向鎖的持有者，能夠直接執行同步塊代碼；不然進入步驟（d）；

（d）若是是從第（b）步直接進入步驟（d），說明當前線程嘗試獲取一個已經處於偏向模式的鎖；而從第（c）步進入的步驟（d），說明當前還有其餘線程在競爭鎖，且當前線程競爭失敗了；這兩種狀況都會使得鎖的偏向模式宣告結束。當到達全局安全點（safepoint），得到偏向鎖的線程被掛起，並撤銷偏向鎖，執行 slow_enter 將鎖升級；升級完成後被阻塞在安全點的線程嘗試獲取升級以後的鎖。

這裏撤銷偏向鎖並將鎖升級，會根據鎖對象目前是否處於被鎖定的狀態分爲兩種狀況：未鎖定（標誌位爲 「01」）和輕量級鎖定（標誌位爲 「00」）。

（5）synchronizer.cpp#slow_enter：openjdk/hotspot/src/share/vm/runtime/synchronizer.cpp

fast_enter 獲取偏向鎖失敗，就會進入到 slow_enter 方法中，而這裏包含了獲取輕量級鎖的實現。

void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {
  markOop mark = obj->mark();

  if (mark->is_neutral()) {
    // Anticipate successful CAS -- the ST of the displaced mark must
    // be visible <= the ST performed by the CAS.
    lock->set_displaced_header(mark);
    if (mark == (markOop) Atomic::cmpxchg_ptr(lock, obj()->mark_addr(), mark)) {
      TEVENT (slow_enter: release stacklock) ;
      return ;
    }
    // Fall through to inflate() ...
  } else
  if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {
    lock->set_displaced_header(NULL);
    return;
  }

	...
  lock->set_displaced_header(markOopDesc::unused_mark());
  ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
}
複製代碼

點擊查看源碼

輕量級鎖的獲取

（a）mark->is_neutral() 判斷 Mark Word 是否處於未鎖定狀態，若是是，進入第（c）步，不然進入第（b）步；

（b）若是當前線程是輕量級鎖的持有者，能夠不須要同步，直接執行同步塊代碼；不然，進入步驟（d）；

（c）經過 CAS 設置鎖對象的 Mark Word 爲 lock，若是設置成功，當前線程成爲了輕量級鎖的持有者，能夠直接執行同步塊代碼；不然進入步驟（d）；

（d）若是是從第（b）步直接進入步驟（d），說明當前線程與一個已經處於輕量級鎖的線程爭用同一個鎖；而從第（c）步進入的步驟（d），說明當前還有其餘線程在競爭鎖，且當前線程競爭失敗了；若是有兩個以上的線程爭用同一個鎖，那輕量級鎖也再也不有效，須要執行 inflate 將鎖膨脹爲重量級鎖，同時將鎖標誌的狀態值變爲 「10」。

（6）synchronizer.cpp#inflate：openjdk/hotspot/src/share/vm/runtime/synchronizer.cpp
ObjectMonitor * ATTR ObjectSynchronizer::inflate (Thread * Self, oop object) {
			...
  for (;;) {
      const markOop mark = object->mark() ;

      // The mark can be in one of the following states:
      // * Inflated - just return
      // * Stack-locked - coerce it to inflated
      // * INFLATING - busy wait for conversion to complete
      // * Neutral - aggressively inflate the object.
      // * BIASED - Illegal. We should never see this

      // CASE: inflated
      if (mark->has_monitor()) {
          ObjectMonitor * inf = mark->monitor() ;
        	...
          return inf ;
      }

      // CASE: inflation in progress - inflating over a stack-lock.
      // Some other thread is converting from stack-locked to inflated.
				...
      if (mark == markOopDesc::INFLATING()) {
         TEVENT (Inflate: spin while INFLATING) ;
         ReadStableMark(object) ;
         continue ;
      }

      // CASE: stack-locked
				...
      if (mark->has_locker()) {
          ObjectMonitor * m = omAlloc (Self) ;
          m->Recycle();
          m->_Responsible  = NULL ;
          m->OwnerIsThread = 0 ;
          m->_recursions   = 0 ;
          m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;   // Consider: maintain by type/class

          markOop cmp = (markOop) Atomic::cmpxchg_ptr (markOopDesc::INFLATING(), object->mark_addr(), mark) ;
          if (cmp != mark) {
             omRelease (Self, m, true) ;
             continue ;       // Interference -- just retry
          }

          ...
          markOop dmw = mark->displaced_mark_helper() ;
          assert (dmw->is_neutral(), "invariant") ;
          m->set_header(dmw) ;
          m->set_owner(mark->locker());
          m->set_object(object);

          object->release_set_mark(markOopDesc::encode(m));

          // Hopefully the performance counters are allocated on distinct cache lines
          // to avoid false sharing on MP systems ...
          if (ObjectMonitor::_sync_Inflations != NULL) ObjectMonitor::_sync_Inflations->inc() ;
          TEVENT(Inflate: overwrite stacklock) ;
          if (TraceMonitorInflation) {
            if (object->is_instance()) {
              ResourceMark rm;
              tty->print_cr("Inflating object " INTPTR_FORMAT " , mark " INTPTR_FORMAT " , type %s",
                (void *) object, (intptr_t) object->mark(),
                object->klass()->external_name());
            }
          }
          return m ;
      }

      // CASE: neutral
      ObjectMonitor * m = omAlloc (Self) ;
      // prepare m for installation - set monitor to initial state
      m->Recycle();
      m->set_header(mark);
      m->set_owner(NULL);
      m->set_object(object);
      m->OwnerIsThread = 1 ;
      m->_recursions   = 0 ;
      m->_Responsible  = NULL ;
      m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;       // consider: keep metastats by type/class

      if (Atomic::cmpxchg_ptr (markOopDesc::encode(m), object->mark_addr(), mark) != mark) {
          m->set_object (NULL) ;
          m->set_owner  (NULL) ;
          m->OwnerIsThread = 0 ;
          m->Recycle() ;
          omRelease (Self, m, true) ;
          m = NULL ;
          continue ;
          // interference - the markword changed - just retry.
          // The state-transitions are one-way, so there's no chance of
          // live-lock -- "Inflated" is an absorbing state.
      }

      // Hopefully the performance counters are allocated on distinct
      // cache lines to avoid false sharing on MP systems ...
      if (ObjectMonitor::_sync_Inflations != NULL) ObjectMonitor::_sync_Inflations->inc() ;
      TEVENT(Inflate: overwrite neutral) ;
      if (TraceMonitorInflation) {
        if (object->is_instance()) {
          ResourceMark rm;
          tty->print_cr("Inflating object " INTPTR_FORMAT " , mark " INTPTR_FORMAT " , type %s",
            (void *) object, (intptr_t) object->mark(),
            object->klass()->external_name());
        }
      }
      return m ;
  }
}
複製代碼

點擊查看源碼

鎖的膨脹過程

這一段代碼乍一看很長，真的很長，不過上面已是刪減了不少代碼以後的版本了。細看以後發現其實這一段代碼很容易理解：

（a）首先，這個代碼最終是要獲得一個 ObjectMonitor 對象，後續會調用其 enter 方法。

（b）而且它是一個自旋方法： for (;;) {}

（c）接着它列舉了 mark 的五種狀態以及在這五種狀態下的操做（實際上第五種是不可達狀態）：

// The mark can be in one of the following states:
      // *  Inflated     - just return
      // *  Stack-locked - coerce it to inflated
      // *  INFLATING    - busy wait for conversion to complete
      // *  Neutral      - aggressively inflate the object.
      // *  BIASED       - Illegal.  We should never see this
複製代碼
  • Inflated：說明已經膨脹完成了，直接返回。
  • Stack-locked：前面在講輕量級鎖的獲取時講到了當兩個以上的線程爭用同一個鎖，要膨脹爲重量級鎖。此時的 mark 是處於輕量級鎖定的狀態，即 Stack-locked，須要執行膨脹操做膨脹爲重量級鎖。這裏須要注意的是每一個競爭鎖的線程都會經過 CAS 嘗試膨脹鎖（CAS 設置 mark 狀態爲 INFLATING），CAS 成功的線程將繼續執行膨脹操做，CAS 失敗的線程則撤銷以前的準備操做，並進入自旋等待膨脹完成。
  • INFLATING：說明其餘線程正在進行膨脹過程，當前線程進入自旋等待膨脹完成；這裏的自旋不會一直佔用 CPU 資源，它每隔一段時間會經過 os::NakedYield() 放棄 CPU 資源，或者調用 park() 掛起本身（點擊查看源碼）；當其餘線程完成鎖的膨脹操做，退出自旋並返回。
  • Neutral：這個狀態說明 mark 尚處於未鎖定狀態，因此這個分支是將 mark 從未鎖定狀態直接膨脹成爲重量級鎖；咱們回退到 slow_enter 方法，查看其調用過程，能夠分析獲得 mark 是不會在 Netural 狀態進行膨脹操做的；因此若是上游方法是 slow_enter 是不會走這個分支的，這個分支應該是爲了支持其餘地方的調用；大體看了一下，這裏的思路和 Stack-locked 相似，都是經過 CAS + 自旋執行膨脹操做，有興趣能夠了解一下。
  • BIASED：不可達狀態。

（d）當鎖膨脹完成，返回對應的 ObjectMonitor 對象以後，並不表示該線程競爭到了鎖，真正的鎖競爭發生在 ObjectMonitor::enter 方法中。

（7）objectMonitor.cpp#enter：openjdk/hotspot/src/share/vm/runtime/objectMonitor.cpp
void ATTR ObjectMonitor::enter(TRAPS) {
  Thread * const Self = THREAD ;
  void * cur ;

  cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
  if (cur == NULL) {
     // Either ASSERT _recursions == 0 or explicitly set _recursions = 0.
     assert (_recursions == 0   , "invariant") ;
     assert (_owner      == Self, "invariant") ;
     // CONSIDER: set or assert OwnerIsThread == 1
     return ;
  }

  if (cur == Self) {
     // TODO-FIXME: check for integer overflow! BUGID 6557169.
     _recursions ++ ;
     return ;
  }

  if (Self->is_lock_owned ((address)cur)) {
    assert (_recursions == 0, "internal state error");
    _recursions = 1 ;
    // Commute owner from a thread-specific on-stack BasicLockObject address to
    // a full-fledged "Thread *".
    _owner = Self ;
    OwnerIsThread = 1 ;
    return ;
  }

		...
    // TODO-FIXME: change the following for(;;) loop to straight-line code.
    for (;;) {
      jt->set_suspend_equivalent();
      // cleared by handle_special_suspend_equivalent_condition()
      // or java_suspend_self()

      EnterI (THREAD) ;

      if (!ExitSuspendEquivalent(jt)) break ;

      //
      // We have acquired the contended monitor, but while we were
      // waiting another thread suspended us. We don't want to enter
      // the monitor while suspended because that would surprise the
      // thread that suspended us.
      //
          _recursions = 0 ;
      _succ = NULL ;
      exit (false, Self) ;

      jt->java_suspend_self();
    }
    ...
  }
		...
}
複製代碼

點擊查看源碼

cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;

首先咱們要明白，這裏返回的 cur 是 &_owner 的實際原值；

（a）若是 cur == NULL，說明該線程執行 cmpxchg_ptr 成功，成功獲取到鎖，直接返回；

（b）若是 cur == Self，即以前持有鎖的線程就是本身，說明這是一個重入操做，只須要將 _recursions 加 1 以後返回便可；

（c）若是 Self->is_lock_owned ((address)cur)，這裏，我以爲能夠這樣理解：首先明白一點，這也是一個重入操做，可是它是一個特殊的重入操做；由於以前持有鎖的 &_owner 其實也是當前線程，可是這個 cur 和 Self 沒辦法經過 == 判斷出來，須要用另外一種方式才能夠判斷，即 Self->is_lock_owned ((address)cur)；這種方式判斷出來若是爲真，那麼 _recursions 的值也必定爲 0，而後將 Self 賦給 _owner，同時， _recursions 設置爲 1，而後返回；

（d）若是上面三種狀況都不知足，那就說明真正遇到了競爭；緊接着一個自旋操做競爭鎖：先經過 EnterI 方法競爭鎖，而後判斷在獲取鎖的這段時間內，若是沒有其餘線程掛起本身，那麼獲取鎖成功，返回；不然將鎖釋放，同時執行 jt->java_suspend_self() 等待掛起本身的線程將本身喚醒。

咱們接下來看 EnterI 這個方法，EnterI 方法纔是真正實現競爭鎖的地方。

（8）objectMonitor.cpp#EnterI：openjdk/hotspot/src/share/vm/runtime/objectMonitor.cpp
void ATTR ObjectMonitor::EnterI (TRAPS) {
    Thread * Self = THREAD ;

    // Try the lock - TATAS
    if (TryLock (Self) > 0) {
        assert (_succ != Self              , "invariant") ;
        assert (_owner == Self             , "invariant") ;
        assert (_Responsible != Self       , "invariant") ;
        return ;
    }
		...
    if (TrySpin (Self) > 0) {
        assert (_owner == Self        , "invariant") ;
        assert (_succ != Self         , "invariant") ;
        assert (_Responsible != Self  , "invariant") ;
        return ;
    }

    // The Spin failed -- Enqueue and park the thread ...
    ObjectWaiter node(Self) ;
    Self->_ParkEvent->reset() ;
    node._prev   = (ObjectWaiter *) 0xBAD ;
    node.TState  = ObjectWaiter::TS_CXQ ;

    // Push "Self" onto the front of the _cxq.
    // Once on cxq/EntryList, Self stays on-queue until it acquires the lock.
    // Note that spinning tends to reduce the rate at which threads
    // enqueue and dequeue on EntryList|cxq.
    ObjectWaiter * nxt ;
    for (;;) {
        node._next = nxt = _cxq ;
        if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;

        // Interference - the CAS failed because _cxq changed. Just retry.
        // As an optional optimization we retry the lock.
        if (TryLock (Self) > 0) {
            assert (_succ != Self         , "invariant") ;
            assert (_owner == Self        , "invariant") ;
            assert (_Responsible != Self  , "invariant") ;
            return ;
        }
    }
		...
    for (;;) {

        if (TryLock (Self) > 0) break ;

        // park self
        if (_Responsible == Self || (SyncFlags & 1)) {
            TEVENT (Inflated enter - park TIMED) ;
            Self->_ParkEvent->park ((jlong) RecheckInterval) ;
            // Increase the RecheckInterval, but clamp the value.
            RecheckInterval *= 8 ;
            if (RecheckInterval > 1000) RecheckInterval = 1000 ;
        } else {
            TEVENT (Inflated enter - park UNTIMED) ;
            Self->_ParkEvent->park() ;
        }

        if (TryLock(Self) > 0) break ;
				...
    }

    UnlinkAfterAcquire (Self, &node) ;
    if (_succ == Self) _succ = NULL ;

    assert (_succ != Self, "invariant") ;
    if (_Responsible == Self) {
        _Responsible = NULL ;
        OrderAccess::fence(); // Dekker pivot-point
    }

    if (SyncFlags & 8) {
       OrderAccess::fence() ;
    }
    return ;
}
複製代碼

點擊查看源碼

若是看過 AQS 的源碼的人來看這一段代碼，會發現這段代碼和 AQS 的 ConditionObject 的行爲很類似。下面咱們來看 EnterI 是怎麼去競爭鎖的：

（a）TryLock (Self) 嘗試一下獲取鎖，若是返回值 > 0，說明獲取鎖成功，返回；

（b）TrySpin (Self) 在入隊以前自旋一小段時間嘗試獲取鎖，若是返回值 > 0，說明獲取鎖成功，返回；

（c）若是前面兩個操做都沒成功，就建立一個 ObjectWaiter 類型的 node，加入到 cxq/EntryList 的隊首；入隊是一個自旋操做，若是入隊失敗，再次嘗試獲取鎖，若是獲取成功，直接返回；不然繼續執行入隊操做；

（d）執行到這，說明 Self 已經在隊列裏面了，這裏又是一個自旋操做：TryLock + park()；嘗試獲取鎖，若是獲取成功，進入（e）；不然掛起本身，等待被喚醒（當持有鎖的線程釋放鎖時，會喚醒最先進入等待隊列的節點）。當該線程喚醒以後，會從掛起的點繼續執行。

（e）獲取到鎖，將本身從隊列中移除，返回。

至此，關於 monitorenter 的執行過程基本上就結束了；而 monitorexit 的執行過程和 monitorenter 基本相似，能夠參考 monitorenter 的執行過程自行理解，這裏再也不進行講述。這裏有一點須要注意，執行 monitorexit 時，會在代碼的最後最後調用 unpark 用於喚醒阻塞於 EnterI 方法中的線程。

遺留的問題

  • 偏向鎖和輕量級鎖是如何記錄重入次數的？

4、結語

看過了 synchronized 底層源碼以後再來回看《深刻理解 Java 虛擬機》中關於 synchronized 的部分，發現原來本身的這些理解書本上原本就有，並且詳細的很，難免想到以前看到的口天師兄的這個回答：你的編程能力從何時開始日新月異？，真的很是精闢。

寫這篇文章花了挺長時間的，不過收穫也確實蠻多的，說說本身的感覺吧：

  • 最開始看的時候實際上是最難受的，基本上遇到的全部東西都是問題，不過慢慢的看多了以後就會發現有不少地方是類似的（例如 EnterI 方法和 AQS 的 ConditionObject）
  • Java Language and Virtual Machine Specifications 中的內容很是好，必定要找時間把它看完
  • 《深刻理解 Java 虛擬機》第二版： 這本書也寫的特別好，值得反覆閱讀，深刻閱讀；這本書最近出第三版了，正在考慮要不要入手一本
  • 書上總結的內容確實是精華，不過若是可以本身實踐一番，效果更佳

參考資料：

（1）17.1. Synchronization

（2）3.14. Synchronization

（3）2.11.10. Synchronization

（4）monitorenter

（5）monitorexit

（6）《深刻理解 Java 虛擬機》第二版 周志明 著.

（7）JVM源碼分析之synchronized實現

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程