java併發筆記四之synchronized 鎖的膨脹過程（鎖的升級過程）深入剖析

警告⚠️：本文耗時很長，先做好心理準備，建議PC端瀏覽器瀏覽效果更佳。

• 當沒有競爭出現時，預設會使用偏向鎖。JVM 會利用 CAS 操作（compare and swap），在對象頭上的 Mark Word 部分設置線程 ID，以表示這個對象偏向於當前線程，所以並不涉及真正的互斥鎖。這樣做的假設是基於在很多應用場景中，大部分對象生命周期中最多會被一個線程鎖定，使用偏向鎖可以降低無競爭開銷。
• 如果有另外的線程試圖鎖定某個已經被偏向過的對象，JVM 就需要撤銷（revoke）偏向鎖，並切換到輕量級鎖實現。輕量級鎖依賴 CAS 操作 Mark Word 來試圖獲取鎖，如果重試成功，就使用輕量級鎖；否則，進一步升級為重量級鎖

public class TestDemo {
}
public class DemoExample1 {
    static TestDemo testDemo;
    public static void main(String[] args) throws Exception {
        testDemo= new TestDemo();
        synchronized (testDemo){
            System.out.println("lock ing");
            testDemo.hashCode();
            System.out.println(ClassLayout.parseInstance(testDemo).toPrintable());
        }
    }
}

javap -c DemoExample1.class

• 當執行 monitorenter 時，如果目標鎖對象的計數器為 0，那麼說明它沒有被其他線程所持有。在這個情況下，Java 虛擬機會將該鎖對象的持有線程設置為當前線程，並且將其計數器加 1。

• 在目標鎖對象的計數器不為 0 的情況下，如果鎖對象的持有線程是當前線程，那麼 Java 虛擬機可以將其計數器加 1，否則需要等待，直至持有線程釋放該鎖。當執行 monitorexit 時，Java 虛擬機則需將鎖對象的計數器減 1。當計數器減為 0 時，那便代表該鎖已經被釋放掉了。

• 之所以採用這種計數器的方式，是為了允許同一個線程重覆獲取同一把鎖。舉個例子，如果一個 Java 類中擁有多個 synchronized 方法，那麼這些方法之間的相互調用，不管是直接的還是間接的，都會涉及對同一把鎖的重覆加鎖操作。因此，我們需要設計這麼一個可重入的特性，來避免編程里的隱式約束。

public class DemoExample3 {
    public int sharedState;

    public void nonSafeAction() {
        while (sharedState < 100000) {
            int former = sharedState++;
            int latter = sharedState;
            if (former != latter - 1) {
                System.out.println("Observed data race, former is " +
                        former + ", " + "latter is " + latter);
            }
        }
    }
 
    public static void main(String[] args) throws InterruptedException {
        final DemoExample3 demoExample3 = new DemoExample3();
        Thread thread1 = new Thread() {
            @Override
            public void run() {
                demoExample3.nonSafeAction();
            }
        };
  
        Thread thread2 = new Thread() {
            @Override
            public void run() {
                demoExample3.nonSafeAction();
            }
        };

        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();
    }
}

public class DemoExample3 {
    public int sharedState;
  
    public void nonSafeAction() {
        while (sharedState < 100000) {
            synchronized (this) {
                int former = sharedState++;
                int latter = sharedState;
                if (former != latter - 1) {
                    System.out.println("Observed data race, former is " +
                            former + ", " + "latter is " + latter);
                }
            }
        }
    }
  
    public static void main(String[] args) throws InterruptedException {
        final DemoExample3 demoExample3 = new DemoExample3();
        Thread thread1 = new Thread() {
            @Override
            public void run() {
                demoExample3.nonSafeAction();
            }
        };
  
        Thread thread2 = new Thread() {
            @Override
            public void run() {
                demoExample3.nonSafeAction();
            }
        };

        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();
    }
}

這次看下加上synchronized關鍵字的列印出來的結果：

// Handles the uncommon case in locking, i.e., contention or an inflated lock.
JRT_BLOCK_ENTRY(void, SharedRuntime::complete_monitor_locking_C(oopDesc* _obj, BasicLock* lock, JavaThread* thread))
  // Disable ObjectSynchronizer::quick_enter() in default config
  // on AARCH64 and ARM until JDK-8153107 is resolved.
  if (ARM_ONLY((SyncFlags & 256) != 0 &&)
      AARCH64_ONLY((SyncFlags & 256) != 0 &&)
      !SafepointSynchronize::is_synchronizing()) {
    // Only try quick_enter() if we're not trying to reach a safepoint
    // so that the calling thread reaches the safepoint more quickly.
    if (ObjectSynchronizer::quick_enter(_obj, thread, lock)) return;
  }
  // NO_ASYNC required because an async exception on the state transition destructor
  // would leave you with the lock held and it would never be released.
  // The normal monitorenter NullPointerException is thrown without acquiring a lock
  // and the model is that an exception implies the method failed.
  JRT_BLOCK_NO_ASYNC
  oop obj(_obj);
  if (PrintBiasedLockingStatistics) {
    Atomic::inc(BiasedLocking::slow_path_entry_count_addr());
  }
  Handle h_obj(THREAD, obj);
  //在 JVM 啟動時，我們可以指定是否開啟偏向鎖
  if (UseBiasedLocking) {
  // Retry fast entry if bias is revoked to avoid unnecessary inflation
  //fast_enter 是我們熟悉的完整鎖獲取路徑
    ObjectSynchronizer::fast_enter(h_obj, lock, true, CHECK);
  } else {
  //slow_enter 則是繞過偏向鎖，直接進入輕量級鎖獲取邏輯
    ObjectSynchronizer::slow_enter(h_obj, lock, CHECK);
  }
  assert(!HAS_PENDING_EXCEPTION, "Should have no exception here");
  JRT_BLOCK_END
JRT_END

// -----------------------------------------------------------------------------
//  Fast Monitor Enter/Exit
// This the fast monitor enter. The interpreter and compiler use
// some assembly copies of this code. Make sure update those code
// if the following function is changed. The implementation is
// extremely sensitive to race condition. Be careful.
void ObjectSynchronizer::fast_enter(Handle obj, BasicLock* lock,
                                    bool attempt_rebias, TRAPS) {
  if (UseBiasedLocking) {
    if (!SafepointSynchronize::is_at_safepoint()) {
      //biasedLocking定義了偏向鎖相關操作，revoke_and_rebias revokeatsafepoint 則定義了當檢測到安全點時的處理
      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，鎖升級
  slow_enter(obj, lock, THREAD);
}
 
// -----------------------------------------------------------------------------
// Interpreter/Compiler Slow Case
// This routine is used to handle interpreter/compiler slow case
// We don't need to use fast path here, because it must have been
// failed in the interpreter/compiler code.
void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {
  markOop mark = obj->mark();
  assert(!mark->has_bias_pattern(), "should not see bias pattern here");
  if (mark->is_neutral()) {
    // Anticipate successful CAS -- the ST of the displaced mark must
    // be visible <= the ST performed by the CAS.
    // 將目前的 Mark Word 複製到 Displaced Header 上
    lock->set_displaced_header(mark);
    // 利用 CAS 設置對象的 Mark Wo
    if (mark == obj()->cas_set_mark((markOop) lock, mark)) {
      return;
    }
    // Fall through to inflate() …
    // 檢查存在競爭
  } 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”);
    // 清除
    lock->set_displaced_header(NULL);
    return;
  }
  // The object header will never be displaced to this lock,
  // so it does not matter what the value is, except that it
  // must be non-zero to avoid looking like a re-entrant lock,
  // and must not look locked either.
  // 重置 Displaced Header
  lock->set_displaced_header(markOopDesc::unused_mark());
  //鎖膨脹
  ObjectSynchronizer::inflate(THREAD,
                              obj(),
                              inflate_cause_monitor_enter)->enter(THREAD);
}
// This routine is used to handle interpreter/compiler slow case
// We don't need to use fast path here, because it must have
// failed in the interpreter/compiler code. Simply use the heavy
// weight monitor should be ok, unless someone find otherwise.
void ObjectSynchronizer::slow_exit(oop object, BasicLock* lock, TRAPS) {
  fast_exit(object, lock, THREAD);
}
 
//鎖膨脹
ObjectMonitor * ATTR ObjectSynchronizer::inflate (Thread * Self, oop object) {
  // Inflate mutates the heap ...
  // Relaxing assertion for bug 6320749.
  assert (Universe::verify_in_progress() ||
          !SafepointSynchronize::is_at_safepoint(), "invariant") ;
 
 
  for (;;) {//自旋
      const markOop mark = object->mark() ;
      assert (!mark->has_bias_pattern(), "invariant") ;
 
      // 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() ;//獲取指向ObjectMonitor的指針
          assert (inf->header()->is_neutral(), "invariant");
          assert (inf->object() == object, "invariant") ;
          assert (ObjectSynchronizer::verify_objmon_isinpool(inf), "monitor is invalid");
          return inf ;
      }
  
      // CASE: inflation in progress - inflating over a stack-lock.膨脹等待（其他線程正在從輕量級鎖轉為膨脹鎖）
      // Some other thread is converting from stack-locked to inflated.
      // Only that thread can complete inflation -- other threads must wait.
      // The INFLATING value is transient.
      // Currently, we spin/yield/park and poll the markword, waiting for inflation to finish.
      // We could always eliminate polling by parking the thread on some auxiliary list.
      if (mark == markOopDesc::INFLATING()) {
         TEVENT (Inflate: spin while INFLATING) ;
         ReadStableMark(object) ;
         continue ;
      }
  
      // CASE: stack-locked棧鎖（輕量級鎖）
      // Could be stack-locked either by this thread or by some other thread.
      //
      // Note that we allocate the objectmonitor speculatively, _before_ attempting
      // to install INFLATING into the mark word.  We originally installed INFLATING,
      // allocated the objectmonitor, and then finally STed the address of the
      // objectmonitor into the mark.  This was correct, but artificially lengthened
      // the interval in which INFLATED appeared in the mark, thus increasing
      // the odds of inflation contention.
      //
      // We now use per-thread private objectmonitor free lists.
      // These list are reprovisioned from the global free list outside the
      // critical INFLATING...ST interval.  A thread can transfer
      // multiple objectmonitors en-mass from the global free list to its local free list.
      // This reduces coherency traffic and lock contention on the global free list.
      // Using such local free lists, it doesn't matter if the omAlloc() call appears
      // before or after the CAS(INFLATING) operation.
      // See the comments in omAlloc().
 
      if (mark->has_locker()) {
          ObjectMonitor * m = omAlloc (Self) ;//獲取一個可用的ObjectMonitor
          // Optimistically prepare the objectmonitor - anticipate successful CAS
          // We do this before the CAS in order to minimize the length of time
          // in which INFLATING appears in the mark.
          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) {//CAS失敗//CAS失敗，說明衝突了，自旋等待//CAS失敗，說明衝突了，自旋等待//CAS失敗，說明衝突了，自旋等待
             omRelease (Self, m, true) ;//釋放監視器鎖
             continue ;       // Interference -- just retry
          } 
 
          // We've successfully installed INFLATING (0) into the mark-word.
          // This is the only case where 0 will appear in a mark-work.
          // Only the singular thread that successfully swings the mark-word
          // to 0 can perform (or more precisely, complete) inflation.
          //
          // Why do we CAS a 0 into the mark-word instead of just CASing the
          // mark-word from the stack-locked value directly to the new inflated state?
          // Consider what happens when a thread unlocks a stack-locked object.
          // It attempts to use CAS to swing the displaced header value from the
          // on-stack basiclock back into the object header.  Recall also that the
          // header value (hashcode, etc) can reside in (a) the object header, or
          // (b) a displaced header associated with the stack-lock, or (c) a displaced
          // header in an objectMonitor.  The inflate() routine must copy the header
          // value from the basiclock on the owner's stack to the objectMonitor, all
          // the while preserving the hashCode stability invariants.  If the owner
          // decides to release the lock while the value is 0, the unlock will fail
          // and control will eventually pass from slow_exit() to inflate.  The owner
          // will then spin, waiting for the 0 value to disappear.   Put another way,
          // the 0 causes the owner to stall if the owner happens to try to
          // drop the lock (restoring the header from the basiclock to the object)
          // while inflation is in-progress.  This protocol avoids races that might
          // would otherwise permit hashCode values to change or "flicker" for an object.
          // Critically, while object->mark is 0 mark->displaced_mark_helper() is stable.
          // 0 serves as a "BUSY" inflate-in-progress indicator
          // fetch the displaced mark from the owner's stack.
          // The owner can't die or unwind past the lock while our INFLATING
          // object is in the mark.  Furthermore the owner can't complete
          // an unlock on the object, either.
          markOop dmw = mark->displaced_mark_helper() ;
          assert (dmw->is_neutral(), "invariant") ;
          //CAS成功，設置ObjectMonitor的_header、_owner和_object等
          // Setup monitor fields to proper values -- prepare the monitor
          m->set_header(dmw) ;

          // Optimization: if the mark->locker stack address is associated
          // with this thread we could simply set m->_owner = Self and
          // m->OwnerIsThread = 1. Note that a thread can inflate an object
          // that it has stack-locked -- as might happen in wait() -- directly
          // with CAS.  That is, we can avoid the xchg-NULL .... ST idiom.
          m->set_owner(mark->locker());
          m->set_object(object);
          // TODO-FIXME: assert BasicLock->dhw != 0.
 
          // Must preserve store ordering. The monitor state must
          // be stable at the time of publishing the monitor address.
          guarantee (object->mark() == markOopDesc::INFLATING(), "invariant") ;
          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 無鎖
      // TODO-FIXME: for entry we currently inflate and then try to CAS _owner.
      // If we know we're inflating for entry it's better to inflate by swinging a
      // pre-locked objectMonitor pointer into the object header.   A successful
      // CAS inflates the object *and* confers ownership to the inflating thread.
      // In the current implementation we use a 2-step mechanism where we CAS()
      // to inflate and then CAS() again to try to swing _owner from NULL to Self.
      // An inflateTry() method that we could call from fast_enter() and slow_enter()
      // would be useful.
 
      assert (mark->is_neutral(), "invariant");
      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 ;
  }
}

1、整個膨脹過程在自旋下完成；

2、mark->has_monitor()方法判斷當前是否為重量級鎖，即Mark Word的鎖標識位為 10，如果當前狀態為重量級鎖，執行步驟（3），否則執行步驟（4）；

3、mark->monitor()方法獲取指向ObjectMonitor的指針，並返回，說明膨脹過程已經完成；

4、如果當前鎖處於膨脹中，說明該鎖正在被其它線程執行膨脹操作，則當前線程就進行自旋等待鎖膨脹完成，這裡需要註意一點，雖然是自旋操作，但不會一直占用cpu資源，每隔一段時間會通過os::NakedYield方法放棄cpu資源，或通過park方法掛起；如果其他線程完成鎖的膨脹操作，則退出自旋並返回；

5、如果當前是輕量級鎖狀態，即鎖標識位為 00，膨脹過程如下：

•     通過omAlloc方法，獲取一個可用的ObjectMonitor monitor，並重置monitor數據；
•     通過CAS嘗試將Mark Word設置為markOopDesc:INFLATING，標識當前鎖正在膨脹中，如果CAS失敗，說明同一時刻其它線程已經將Mark Word設置為markOopDesc:INFLATING，當前線程進行自旋等待膨脹完成；
•     如果CAS成功，設置monitor的各個欄位：_header、_owner和_object等，並返回；

6、如果是無鎖，重置監視器值；

• 偏向鎖是指一段同步代碼一直被一個線程所訪問，那麼該線程會自動獲取鎖，降低獲取鎖的代價。
• 在大多數情況下，鎖總是由同一線程多次獲得，不存在多線程競爭，所以出現了偏向鎖。其目標就是在只有一個線程執行同步代碼塊時能夠提高性能。
• 當一個線程訪問同步代碼塊並獲取鎖時，會在Mark Word里存儲鎖偏向的線程ID。線上程進入和退出同步塊時不再通過CAS操作來加鎖和解鎖，而是檢測Mark Word里是否存儲著指向當前線程的偏向鎖。引入偏向鎖是為了在無多線程競爭的情況下儘量減少不必要的輕量級鎖執行路徑，因為輕量級鎖的獲取及釋放依賴多次CAS原子指令，而偏向鎖只需要在置換ThreadID的時候依賴一次CAS原子指令即可。
• 偏向鎖只有遇到其他線程嘗試競爭偏向鎖時，持有偏向鎖的線程才會釋放鎖，線程不會主動釋放偏向鎖。偏向鎖的撤銷，需要等待全局安全點（在這個時間點上沒有位元組碼正在執行），它會首先暫停擁有偏向鎖的線程，判斷鎖對象是否處於被鎖定狀態。撤銷偏向鎖後恢復到無鎖（標誌位為“01”）或輕量級鎖（標誌位為“00”）的狀態。
• 偏向鎖在JDK 6及以後的JVM里是預設啟用的。可以通過JVM參數關閉偏向鎖：-XX:-UseBiasedLocking=false，關閉之後程式預設會進入輕量級鎖狀態。

//創建一個啥都沒有的類： 
public class TestDemo {}
 
public class DemoExample {
    static TestDemo testDemo;
    public static void main(String[] args) throws Exception {
        //此處睡眠50000ms，取消jvm預設偏向鎖延遲4000ms
        Thread.sleep(5000);
        testDemo= new TestDemo();
 
        //hash計算？
        //testDemo.hashCode();
 
        System.out.println("befor lock");
        //無鎖：偏向鎖？
        System.out.println(ClassLayout.parseInstance(testDemo).toPrintable());
 
        synchronized (testDemo){
            System.out.println("lock ing");
            System.out.println(ClassLayout.parseInstance(testDemo).toPrintable());
        }
 
        System.out.println("after lock");
        System.out.println(ClassLayout.parseInstance(testDemo).toPrintable());
    }
}