前言

请看上篇：Java 对象头那点事

文章中的源码都有不同程度缩减，来源于openjdk8的开源代码（tag:jdk8-b120）。

锁粗化过程
image

偏向锁

①：markword中保存的线程ID是自己且epoch等于class的epoch，则说明是偏向锁重入。
②：偏向锁若已禁用，进行撤销偏向锁。
③：偏向锁开启，都进行进行重偏向操作。
④：若进行了锁撤销操作或重偏向操作失败，则需要升级为轻量级锁或者进一步升级为重量级锁。

锁对象在发送锁竞争后会升级为偏向锁，不过当不发生锁竞争时，锁对象依然会升级为偏向锁，这种情况叫匿名偏向。
当jvm启动4s后，会默认给新建的对象加上偏向锁。

上代码：

  1.         <dependency>
  2.             <groupId>org.openjdk.jol</groupId>
  3.             <artifactId>jol-core</artifactId>
  4.             <version>0.8</version>
  5.         </dependency>

这个包下的工具类的功能有：

  1.          // 查看对象内部结构
  2.          System.out.println(ClassLayout.parseInstance(bingo).toPrintable());
  3.          // 查看对象外部信息
  4.          System.out.println(GraphLayout.parseInstance(bingo).toPrintable());
  5.          // 查看对象总大小
  6.          System.out.println(GraphLayout.parseInstance(bingo).totalSize());

默认JVM是开启指针压缩，可以通过vm参数开启关闭指针压缩：-XX:-UseCompressedOops。

当创建锁对象前不进行休眠4s的操作:

  1.     @Test
  2.     public void mark() throws InterruptedException {
  3.         Bingo bingo = new Bingo();
  4.         bingo.setP(1);
  5.         bingo.setB(false);
  6.         // 查看对象内部结构
  7.         System.out.println(ClassLayout.parseInstance(bingo).toPrintable());
  8.         System.out.println("\n++++++++++++++++++++++++++\n");
  9.         synchronized (bingo) {
  10.             // 查看对象内部结构
  11.             System.out.println(ClassLayout.parseInstance(bingo).toPrintable());
  12.         }
  13.     }

image

看我标红线的后三位的值，由于启动过快，锁直接从无锁升级成了轻量级锁。

当创建锁对象前进行休眠4s的操作:

  1.     @Test
  2.     public void mark() throws InterruptedException {
  3.         TimeUnit.SECONDS.sleep(4);
  5.         Bingo bingo = new Bingo();
  6.         bingo.setP(1);
  7.         bingo.setB(false);
  8.         // 查看对象内部结构
  9.         System.out.println(ClassLayout.parseInstance(bingo).toPrintable());
  10.         System.out.println("\n++++++++++++++++++++++++++\n");
  11.         synchronized (bingo) {
  12.             // 查看对象内部结构
  13.             System.out.println(ClassLayout.parseInstance(bingo).toPrintable());
  14.         }
  15.     }

image

当在程序启动4s后创建锁对象，就会默认偏向。

因为偏向锁不会自动释放，因此当锁对象处于偏向锁时，另一个线程进来只能依托VM判断上一个获取偏向锁的线程是否存活、是否退出持有锁来决定是锁升级还是进行重偏向。

①：偏向锁的撤销必须等待VM全局安全点(安全点指所有java线程都停在安全点，只有vm线程运行)。
②：撤销偏向锁恢复到无锁（标志位为 01）或轻量级锁（标志位为 00）的状态。
③：只要发生锁竞争，就会进行锁撤销。

备注：
当开启偏向锁时，若持有偏向锁的线程仍然存活且未退出同步代码块，锁升级为轻量级锁/重量级锁之前会进行偏向锁撤销操作。
如果是升级为轻量级锁，撤销之后需要创建Lock Record 来保存之前的markword信息。

批量偏向/撤销概念：
参考1：https://www.cnblogs.com/LemonFive/p/11248248.html

  • 批量重偏向
    当一个线程同时持有同一个类的多个对象的偏向锁时（这些对象的锁竞争不激烈），执行完同步代码块后，如果另一个线程也要持有这些对象的锁，当对象数量达到一定程度时，会触发批量重偏向机制（进行过批量重偏向的对象不可再进行批量重偏向）。
  • 批量锁撤销
    当触发批量重偏向后，会触发批量撤销机制。

阈值定义在globals.hpp中：

  1.   // 批量重偏向阈值
  2.   product(intx, BiasedLockingBulkRebiasThreshold, 20)
  3.   // 批量锁撤销阈值
  4.   product(intx, BiasedLockingBulkRevokeThreshold, 40)

可以在VM启动参数中通过-XX:BiasedLockingBulkRebiasThreshold 和 -XX:BiasedLockingBulkRevokeThreshold 来手动设置阈值。

偏向锁的撤销和重偏向的代码（过于复杂）在biasedLocking.cpp中：

  1. void BiasedLocking::revoke_at_safepoint(Handle h_obj) {
  2.   assert(SafepointSynchronize::is_at_safepoint(), "must only be called while at safepoint");
  3.   oop obj = h_obj();
  4.   HeuristicsResult heuristics = update_heuristics(obj, false);
  5.   if (heuristics == HR_SINGLE_REVOKE) {
  6.     // 重偏向
  7.     revoke_bias(obj, false, false, NULL);
  8.   } else if ((heuristics == HR_BULK_REBIAS) ||
  9.              (heuristics == HR_BULK_REVOKE)) {
  10.     // 批量撤销或重偏向
  11.     bulk_revoke_or_rebias_at_safepoint(obj, (heuristics == HR_BULK_REBIAS), false, NULL);
  12.   }
  13.   clean_up_cached_monitor_info();
  14. }

参考2：

对于存在明显多线程竞争的场景下使用偏向锁是不合适的，比如生产者-消费者队列。生产者线程获得了偏向锁，消费者线程再去获得锁的时候，就涉及到这个偏向锁的撤销（revoke）操作，而这个撤销是比较昂贵的。那么怎么判断这些对象是否适合偏向锁呢？jvm采用以类为单位的做法，其内部为每个类维护一个偏向锁计数器，对其对象进行偏向锁的撤销操作进行计数。当这个值达到指定阈值的时候，jvm就认为这个类的偏向锁有问题，需要进行重偏向（rebias）。对所有属于这个类的对象进行重偏向的操作叫批量重偏向（bulk rebias），之前的做法是对heap进行遍历，后来引入epoch。当需要bulk rebias时，对这个类的epoch值加1，以后分配这个类的对象的时候mark字段里就是这个epoch值了，同时还要对当前已经获得偏向锁的对象的epoch值加1，这些锁数据记录在方法栈里。这样判断这个对象是否获得偏向锁的条件就是：mark字段后3位是101，thread字段跟当前线程相同，epoch字段跟所属类的epoch值相同。如果epoch值不一样，即使thread字段指向当前线程，也是无效的，相当于进行过了rebias，只是没有对对象的mark字段进行更新。如果这个类的revoke计数器继续增加到一个阈值，那个jvm就认为这个类不适合偏向锁了，就要进行bulk revoke。于是多了一个判断条件，要查看所属类的字段，看看是否允许对这个类使用偏向锁。

轻量级锁

轻量级体现在线程会尝试在自己的堆栈中创建Lock Record存储锁对象的相关信息，不需要在内核态和用户态之间进行切换，不需要操作系统进行调度。

拿到轻量级锁线程堆栈：
image

Lock Record主要分为两部分：

  • obj
    指向锁对象本身。重入时也如此。
  • displaced header（缩写为hdr）
    第一次拿到锁时hdr存放的是encode加密后的markword，重入时存放null。

思考：为什么锁重入时hdr存放的是null，而不是用计数器来实现呢？
假设一个场景，当一个线程同时拿到A、B、C…N 多个锁的时候，那么线程的堆栈中，肯定有多个锁对象的Lock Record,
如：

  1. synchronized(a){
  2.     synchronized(b){
  3.         synchronized(c){
  4.             // do something
  5.             synchronized(a){
  6.                 // do something
  7.             }
  8.         }
  9.     }
  10. }

当锁a重入时，如果用计数器，还得遍历当前线程堆栈拿到第一次的Lock Record，解锁时也要遍历，效率必然低下。作为jdk底层代码必然讲究效率。
以上纯属个人看法（欢迎交流）。

①：使用遍历方式将当前线程堆栈中属于该锁对象的Lock Record 指向Null。
②：CAS还原markword为无锁状态。
③：第②步失败需要升级为重量级锁。
image

  • 优点
    在线程接替/交替执行的情况下，锁竞争比较小，可以避免成为重量级锁而引起的性能问题。

  • 缺点
    当锁竞争比较激烈、多线程同事竞争锁的时候，需要从轻量级升级为重量级，产生了额外的开销。

加锁
加锁、解锁流程的代码在InterpreterRuntime.cpp中。
这是我从github拉下来的源码：

  1.       /**
  2.        * （轻量级锁）加锁流程
  3.        * */
  4.       CASE(_monitorenter): {
  5.         // （锁对象本身）
  6.         oop lockee = STACK_OBJECT(-1);
  7.         // derefing's lockee ought to provoke implicit null check
  8.         CHECK_NULL(lockee);
  9.         // find a free monitor or one already allocated for this object
  10.         // if we find a matching object then we need a new monitor
  11.         // since this is recursive enter
  12.         BasicObjectLock* limit = istate->monitor_base();
  13.         BasicObjectLock* most_recent = (BasicObjectLock*) istate->stack_base();
  14.         // （这个entry就是大家常说的Lock Record吧）
  15.         BasicObjectLock* entry = NULL;
  16.         while (most_recent != limit ) {
  17.           if (most_recent->obj() == NULL) entry = most_recent;
  18.           else if (most_recent->obj() == lockee) break;
  19.           most_recent++;
  20.         }
  21.         if (entry != NULL) {
  22.           entry->set_obj(lockee);
  23.           // （构建一个无锁状态的mark word）
  24.           markOop displaced = lockee->mark()->set_unlocked();
  25.           // （放到lock record 中）
  26.           entry->lock()->set_displaced_header(displaced);
  27.           // 锁对象的markword是否为这个无锁的displaced markword
  28.           // （CAS替换失败说明锁对象的markword 不是无所状态）
  29.           if (Atomic::cmpxchg_ptr(entry, lockee->mark_addr(), displaced) != displaced) {
  30.             // Is it simple recursive case?
  31.             // （判断是否是锁重入）
  32.             if (THREAD->is_lock_owned((address) displaced->clear_lock_bits())) {
  33.               // （如果是重入场景，那么新的Lock Record 设置为Null）
  34.               entry->lock()->set_displaced_header(NULL);
  35.             } else {
  36.               // （不是锁重入，且抢锁失败，说明锁竞争激烈，升级为重量级。进入重量级锁抢锁流程）
  37.               CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
  38.             }
  39.           }
  40.           UPDATE_PC_AND_TOS_AND_CONTINUE(1, -1);
  41.         } else {
  42.           istate->set_msg(more_monitors);
  43.           UPDATE_PC_AND_RETURN(0); // Re-execute
  44.         }
  45.       }

可以看得出来，这部分代码并没有体现出偏向锁的逻辑，有大佬给出原因，可以参考这篇博客：https://www.jianshu.com/p/4758852cbff4

其他大佬解析后的代码：

点击查看代码 
  1. CASE(_monitorenter): {
  2.   // lockee 就是锁对象
  3.   oop lockee = STACK_OBJECT(-1);
  4.   // derefing's lockee ought to provoke implicit null check
  5.   CHECK_NULL(lockee);
  6.   // code 1：找到一个空闲的Lock Record
  7.   BasicObjectLock* limit = istate->monitor_base();
  8.   BasicObjectLock* most_recent = (BasicObjectLock*) istate->stack_base();
  9.   BasicObjectLock* entry = NULL;
  10.   while (most_recent != limit ) {
  11.     if (most_recent->obj() == NULL) entry = most_recent;
  12.     else if (most_recent->obj() == lockee) break;
  13.     most_recent++;
  14.   }
  15.   //entry不为null，代表还有空闲的Lock Record
  16.   if (entry != NULL) {
  17.     // code 2：将Lock Record的obj指针指向锁对象
  18.     entry->set_obj(lockee);
  19.     int success = false;
  20.     uintptr_t epoch_mask_in_place = (uintptr_t)markOopDesc::epoch_mask_in_place;
  21.     // markoop即对象头的mark word
  22.     markOop mark = lockee->mark();
  23.     intptr_t hash = (intptr_t) markOopDesc::no_hash;
  24.     // code 3：如果锁对象的mark word的状态是偏向模式
  25.     if (mark->has_bias_pattern()) {
  26.       uintptr_t thread_ident;
  27.       uintptr_t anticipated_bias_locking_value;
  28.       thread_ident = (uintptr_t)istate->thread();
  29.      // code 4：这里有几步操作，下文分析
  30.       anticipated_bias_locking_value =
  31.         (((uintptr_t)lockee->klass()->prototype_header() | thread_ident) ^ (uintptr_t)mark) &
  32.         ~((uintptr_t) markOopDesc::age_mask_in_place);
  33.      // code 5：如果偏向的线程是自己且epoch等于class的epoch
  34.       if  (anticipated_bias_locking_value == 0) {
  35.         // already biased towards this thread, nothing to do
  36.         if (PrintBiasedLockingStatistics) {
  37.           (* BiasedLocking::biased_lock_entry_count_addr())++;
  38.         }
  39.         success = true;
  40.       }
  41.        // code 6：如果偏向模式关闭，则尝试撤销偏向锁
  42.       else if ((anticipated_bias_locking_value & markOopDesc::biased_lock_mask_in_place) != 0) {
  43.         markOop header = lockee->klass()->prototype_header();
  44.         if (hash != markOopDesc::no_hash) {
  45.           header = header->copy_set_hash(hash);
  46.         }
  47.         // 利用CAS操作将mark word替换为class中的mark word
  48.         if (Atomic::cmpxchg_ptr(header, lockee->mark_addr(), mark) == mark) {
  49.           if (PrintBiasedLockingStatistics)
  50.             (*BiasedLocking::revoked_lock_entry_count_addr())++;
  51.         }
  52.       }
  53.          // code 7：如果epoch不等于class中的epoch，则尝试重偏向
  54.       else if ((anticipated_bias_locking_value & epoch_mask_in_place) !=0) {
  55.         // 构造一个偏向当前线程的mark word
  56.         markOop new_header = (markOop) ( (intptr_t) lockee->klass()->prototype_header() | thread_ident);
  57.         if (hash != markOopDesc::no_hash) {
  58.           new_header = new_header->copy_set_hash(hash);
  59.         }
  60.         // CAS替换对象头的mark word
  61.         if (Atomic::cmpxchg_ptr((void*)new_header, lockee->mark_addr(), mark) == mark) {
  62.           if (PrintBiasedLockingStatistics)
  63.             (* BiasedLocking::rebiased_lock_entry_count_addr())++;
  64.         }
  65.         else {
  66.           // 重偏向失败，代表存在多线程竞争，则调用monitorenter方法进行锁升级
  67.           CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
  68.         }
  69.         success = true;
  70.       }
  71.       else {
  72.          // 走到这里说明当前要么偏向别的线程，要么是匿名偏向（即没有偏向任何线程）
  73.         // code 8：下面构建一个匿名偏向的mark word，尝试用CAS指令替换掉锁对象的mark word
  74.         markOop header = (markOop) ((uintptr_t) mark & ((uintptr_t)markOopDesc::biased_lock_mask_in_place |(uintptr_t)markOopDesc::age_mask_in_place |epoch_mask_in_place));
  75.         if (hash != markOopDesc::no_hash) {
  76.           header = header->copy_set_hash(hash);
  77.         }
  78.         markOop new_header = (markOop) ((uintptr_t) header | thread_ident);
  79.         // debugging hint
  80.         DEBUG_ONLY(entry->lock()->set_displaced_header((markOop) (uintptr_t) 0xdeaddead);)
  81.         if (Atomic::cmpxchg_ptr((void*)new_header, lockee->mark_addr(), header) == header) {
  82.            // CAS修改成功
  83.           if (PrintBiasedLockingStatistics)
  84.             (* BiasedLocking::anonymously_biased_lock_entry_count_addr())++;
  85.         }
  86.         else {
  87.           // 如果修改失败说明存在多线程竞争，所以进入monitorenter方法
  88.           CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
  89.         }
  90.         success = true;
  91.       }
  92.     }
  94.     // 如果偏向线程不是当前线程或没有开启偏向模式等原因都会导致success==false
  95.     if (!success) {
  96.       // 轻量级锁的逻辑
  97.       //code 9: 构造一个无锁状态的Displaced Mark Word，并将Lock Record的lock指向它
  98.       markOop displaced = lockee->mark()->set_unlocked();
  99.       entry->lock()->set_displaced_header(displaced);
  100.       //如果指定了-XX:+UseHeavyMonitors，则call_vm=true，代表禁用偏向锁和轻量级锁
  101.       bool call_vm = UseHeavyMonitors;
  102.       // 利用CAS将对象头的mark word替换为指向Lock Record的指针
  103.       if (call_vm || Atomic::cmpxchg_ptr(entry, lockee->mark_addr(), displaced) != displaced) {
  104.         // 判断是不是锁重入
  105.         if (!call_vm && THREAD->is_lock_owned((address) displaced->clear_lock_bits())) {
  106.           //code 10: 如果是锁重入，则直接将Displaced Mark Word设置为null
  107.           entry->lock()->set_displaced_header(NULL);
  108.         } else {
  109.           CALL_VM(InterpreterRuntime::monitorenter(THREAD, entry), handle_exception);
  110.         }
  111.       }
  112.     }
  113.     UPDATE_PC_AND_TOS_AND_CONTINUE(1, -1);
  114.   } else {
  115.     // lock record不够，重新执行
  116.     istate->set_msg(more_monitors);
  117.     UPDATE_PC_AND_RETURN(0); // Re-execute
  118.   }
  119. }

解锁

  1.       /**
  2.        * （轻量级锁）解锁流程
  3.        * */
  4.       CASE(_monitorexit): {
  5.         oop lockee = STACK_OBJECT(-1);
  6.         CHECK_NULL(lockee);
  8.         BasicObjectLock* limit = istate->monitor_base();
  9.         BasicObjectLock* most_recent = (BasicObjectLock*) istate->stack_base();
  10.         // （挨个遍历当前线程栈中的Lock Record）
  11.         while (most_recent != limit ) {
  12.           // (Lock Record 的obj是否是需解锁的锁对象)
  13.           if ((most_recent)->obj() == lockee) {
  14.             BasicLock* lock = most_recent->lock();
  15.             markOop header = lock->displaced_header();
  16.             // （将obj设置为null（作删除处理））
  17.             most_recent->set_obj(NULL);
  18.             // If it isn't recursive we either must swap old header or call the runtime
  19.             if (header != NULL) {
  20.               // （非重入，CAS替换对象头的markword 为Lock Rocord中的displaced markword）
  21.               if (Atomic::cmpxchg_ptr(header, lockee->mark_addr(), lock) != lock) {
  22.                 // restore object for the slow case
  23.                 // （替换失败，表示锁已膨胀为重量级锁，此时markword指向ObjectMonitor的地址）
  24.                 most_recent->set_obj(lockee);
  25.                 // (走重量级锁的锁退出流程)
  26.                 CALL_VM(InterpreterRuntime::monitorexit(THREAD, most_recent), handle_exception);
  27.               }
  28.             }
  29.             UPDATE_PC_AND_TOS_AND_CONTINUE(1, -1);
  30.           }
  31.           most_recent++;
  32.         }
  33.         // Need to throw illegal monitor state exception
  34.         CALL_VM(InterpreterRuntime::throw_illegal_monitor_state_exception(THREAD), handle_exception);
  35.         ShouldNotReachHere();
  36.       }

重量级锁

重量级锁是基于monitor模型进行实现的。

重量级锁是如何体现重量级的？
①：需要创建monitor，包含阻塞队列、竞争队列、继承者、锁拥有者等大量数据，会占用大量内存。
②：需要调用操作系统对线程进行park、unpark操作，会涉及到cpu在用户态和内核态之间切换，开销大。
③：monitor所运行的VM线程（内核线程）需要操作系统将那些调度，耗费时间。

image

①：monitor并不是一下子初始化完成的。
②：monitor在初始化的过程中，如果有线程进来获取锁，则会进行自旋。
③：线程进入monitor后会被封装成一个ObjectWaiter（双向链表结构），然后park住当前线程。当有线程退出锁后会进行unpark操作（唤醒操作涉及到操作系统，会产生额外的开销）。

ObjectWaiter的结构：

  1. class ObjectWaiter : public StackObj {
  2.   // ...
  3.   ObjectWaiter * volatile _next;
  4.   ObjectWaiter * volatile _prev;
  5.   Thread*       _thread;
  6.   // ...
  7. };

image

  1.   volatile markOop        _header;                   // displaced object header word - mark
  2.   void*   volatile        _object;                   // backward object pointer - strong root
  3.   void *  volatile        _owner;                    // pointer to owning thread OR BasicLock
  4.   volatile                jlong _previous_owner_tid; // thread id of the previous owner of the monitor
  5.   volatile intptr_t       _recursions;               // recursion count, 0 for first entry
  6.   int                     OwnerIsThread ;            // _owner is (Thread *) vs SP/BasicLock
  7.   ObjectWaiter * volatile _cxq ;                     // LL of recently-arrived threads blocked on entry.
  8.   ObjectWaiter * volatile _EntryList ;               // Threads blocked on entry or reentry.
  9.   Thread * volatile       _succ ;                    // Heir presumptive thread - used for futile wakeup throttling
  10.   volatile intptr_t       _count;
  11.   volatile intptr_t       _waiters;                  // number of waiting threads
  12.   ObjectWaiter * volatile _WaitSet;                  // LL of threads wait()ing on the monitor

image

阻塞队列中的线程进入_cxq、_EntryList队列的过程有着不同的策略：

  • policy == 0，头插_EntryList
  • policy == 1，尾插_EntryList
  • policy == 2，头插_cxq
  • policy == 3，尾插_cxq

加锁第一阶段
这部分代码并没有创建monitor。
大部分工作是对锁状态做判断、安全点的检查，考虑无锁、轻量级锁的重入情况，因为锁升级为重量级锁就直接进内核态了，消耗资源太多。

InterpreterRuntime.cpp#monitorenter源码：

  1. IRT_ENTRY_NO_ASYNC(void, InterpreterRuntime::monitorenter(JavaThread* thread, BasicObjectLock* elem))
  2. #ifdef ASSERT
  3.   thread->last_frame().interpreter_frame_verify_monitor(elem);
  4. #endif
  5.   if (PrintBiasedLockingStatistics) {
  6.     Atomic::inc(BiasedLocking::slow_path_entry_count_addr());
  7.   }
  8.   Handle h_obj(thread, elem->obj());
  9.   assert(Universe::heap()->is_in_reserved_or_null(h_obj()),
  10.          "must be NULL or an object");
  11.   // 开启偏向锁
  12.   if (UseBiasedLocking) {
  13.     // Retry fast entry if bias is revoked to avoid unnecessary inflation
  14.     ObjectSynchronizer::fast_enter(h_obj, elem->lock(), true, CHECK);
  15.   } else {
  16.     ObjectSynchronizer::slow_enter(h_obj, elem->lock(), CHECK);
  17.   }
  18.   assert(Universe::heap()->is_in_reserved_or_null(elem->obj()),
  19.          "must be NULL or an object");
  20. #ifdef ASSERT
  21.   thread->last_frame().interpreter_frame_verify_monitor(elem);
  22. #endif
  23. IRT_END

主要还是看ObjectSynchronizer::fast_enter、ObjectSynchronizer::slow_enter，这部分源码在synchronizer.cpp中。

  1. void ObjectSynchronizer::fast_enter(Handle obj, BasicLock* lock, bool attempt_rebias, TRAPS) {
  2.  // 开启偏向锁
  3.  if (UseBiasedLocking) {
  4.     if (!SafepointSynchronize::is_at_safepoint()) {
  5.       // 不在安全点(安全点指所有java线程都停在安全点，只有vm线程运行)，需要撤销并重偏向
  6.       BiasedLocking::Condition cond = BiasedLocking::revoke_and_rebias(obj, attempt_rebias, THREAD);
  7.       if (cond == BiasedLocking::BIAS_REVOKED_AND_REBIASED) {
  8.         return;
  9.       }
  10.     } else {
  11.       assert(!attempt_rebias, "can not rebias toward VM thread");
  12.       // 在安全点进行偏向锁的撤销
  13.       BiasedLocking::revoke_at_safepoint(obj);
  14.     }
  15.     assert(!obj->mark()->has_bias_pattern(), "biases should be revoked by now");
  16.  }
  18.  // 上述操作是要保证在进入重量级锁之前锁状态应该处于轻量级锁
  19.  slow_enter (obj, lock, THREAD) ;
  20. }
  22. /**
  23.  * slow enter
  24.  * 主要对锁状态做判断，考虑无锁、轻量级锁的重入情况，因为锁升级为重量级锁就直接进内核态了，消耗资源太多。
  25.  * */
  26. void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {
  27.   markOop mark = obj->mark();
  28.   assert(!mark->has_bias_pattern(), "should not see bias pattern here");
  30.   // （mark word是无锁状态）
  31.   if (mark->is_neutral()) {
  32.     lock->set_displaced_header(mark);
  33.     if (mark == (markOop) Atomic::cmpxchg_ptr(lock, obj()->mark_addr(), mark)) {
  34.       TEVENT (slow_enter: release stacklock) ;
  35.       return ;
  36.     }
  37.   } else
  38.   // （如果是锁重入）
  39.   if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {
  40.     assert(lock != mark->locker(), "must not re-lock the same lock");
  41.     assert(lock != (BasicLock*)obj->mark(), "don't relock with same BasicLock");
  42.     lock->set_displaced_header(NULL);
  43.     return;
  44.   }
  46.   // markword的值设置为值为marked_value的markword（不能看起来无锁，也不能看起来像持有偏向锁、轻量级锁的情况）
  47.   lock->set_displaced_header(markOopDesc::unused_mark());
  48.   // 膨胀为重量级锁，enter方法后面进入重量级锁的抢占流程
  49.   ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
  50. }

如果是进入fast_enter()，那么就会再进行一次偏向锁开启的判断，再进入slow_enter()的逻辑中去，那么为什么不开始就直接进行slow_enter呢？就为了判断下锁偏向和撤销吗？这部分逻辑也完全可以写到slow_enter中去。这么写的原因未知。

加锁第二阶段
形成monitor，用来调度竞争锁的线程。

先看锁的膨胀过程：

  1. ObjectMonitor * ATTR ObjectSynchronizer::inflate (Thread * Self, oop object) {
  2.   // 自旋
  3.   for (;;) {
  4.       const markOop mark = object->mark() ;
  5.       assert (!mark->has_bias_pattern(), "invariant") ;
  7.       // The mark can be in one of the following states:
  8.       // *  Inflated     - just return（膨胀完成，直接返回）
  9.       // *  Stack-locked - coerce it to inflated（轻量级加锁状态）
  10.       // *  INFLATING    - busy wait for conversion to complete（膨胀中）
  11.       // *  Neutral      - aggressively inflate the object.（无锁状态）
  12.       // *  BIASED       - Illegal.  We should never see this（）（偏向锁，非法，这里不能出现）
  14.       // CASE: inflated
  15.       if (mark->has_monitor()) {
  16.           ObjectMonitor * inf = mark->monitor() ;
  17.           assert (inf->header()->is_neutral(), "invariant");
  18.           assert (inf->object() == object, "invariant") ;
  19.           assert (ObjectSynchronizer::verify_objmon_isinpool(inf), "monitor is invalid");
  20.           return inf ;
  21.       }
  23.       // 膨胀中，进行下一轮自旋
  24.       if (mark == markOopDesc::INFLATING()) {
  25.          TEVENT (Inflate: spin while INFLATING) ;
  26.          ReadStableMark(object) ;
  27.          continue ;
  28.       }
  30.       // 轻量级锁状态
  31.       if (mark->has_locker()) {
  32.           // 为当前线程分配一个monitor
  33.           ObjectMonitor * m = omAlloc (Self) ;
  34.           m->Recycle();
  35.           m->_Responsible  = NULL ;
  36.           m->OwnerIsThread = 0 ;
  37.           m->_recursions   = 0 ;
  38.           m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;   // Consider: maintain by type/class
  40.           // CAS操作：尝试将markword设置为INFLATING状态，失败进行下一轮自旋
  41.           markOop cmp = (markOop) Atomic::cmpxchg_ptr (markOopDesc::INFLATING(), object->mark_addr(), mark) ;
  42.           if (cmp != mark) {
  43.              omRelease (Self, m, true) ;
  44.              continue ;       // Interference -- just retry
  45.           }
  47.           markOop dmw = mark->displaced_mark_helper() ;
  48.           assert (dmw->is_neutral(), "invariant") ;
  50.           m->set_header(dmw) ;
  51.           m->set_owner(mark->locker());
  52.           m->set_object(object);
  54.           guarantee (object->mark() == markOopDesc::INFLATING(), "invariant") ;
  55.           object->release_set_mark(markOopDesc::encode(m));
  57.           if (ObjectMonitor::_sync_Inflations != NULL) ObjectMonitor::_sync_Inflations->inc() ;
  58.           TEVENT(Inflate: overwrite stacklock) ;
  59.           if (TraceMonitorInflation) {
  60.             if (object->is_instance()) {
  61.               ResourceMark rm;
  62.               tty->print_cr("Inflating object " INTPTR_FORMAT " , mark " INTPTR_FORMAT " , type %s",
  63.                 (void *) object, (intptr_t) object->mark(),
  64.                 object->klass()->external_name());
  65.             }
  66.           }
  67.           return m ;
  68.       }
  70.       /**
  71.       * 走到这里说明1：monitor 未膨胀完成 2：monitor不在膨胀过程中 3：锁状态也不是轻量级状态
  72.       * 能走到这里说明锁状态已经变为无锁状态了
  73.       */
  74.       assert (mark->is_neutral(), "invariant");
  75.       ObjectMonitor * m = omAlloc (Self) ;
  76.       m->Recycle();
  77.       m->set_header(mark);
  78.       m->set_owner(NULL);
  79.       m->set_object(object);
  80.       m->OwnerIsThread = 1 ;
  81.       m->_recursions   = 0 ;
  82.       m->_Responsible  = NULL ;
  83.       m->_SpinDuration = ObjectMonitor::Knob_SpinLimit ;       // consider: keep metastats by type/class
  84.      // (省略部分代码)
  86.       return m ;
  87.   }
  88. }

ObjectSynchronizer::omAlloc的作用：

尝试从线程的本地omFreeList 分配。线程将首先尝试从其本地列表中分配，然后从全局列表中，只有在那些尝试失败后，线程才会尝试实例化新的监视器。线程本地空闲列表占用 加热 ListLock 并改善分配延迟，并减少共享全局列表上的一致性流量。

总之我也没看懂，大概就是分配一个monitor给该线程用…

加锁第三阶段
当monitor形成之后，线程是阻塞还是拿到锁执行同步块代码，就看线程自己的运气了。

线程进入monitor：

  1. void ATTR ObjectMonitor::EnterI (TRAPS) {
  2.     // 省略部分代码...
  4.     // 尝试获取锁
  5.     if (TryLock (Self) > 0) {
  6.         return ;
  7.     }
  8.     DeferredInitialize () ;
  9.     // 不死心，再来一次
  10.     if (TrySpin (Self) > 0) {
  11.         return ;
  12.     }
  14.     ObjectWaiter node(Self) ;
  15.     Self->_ParkEvent->reset() ;
  16.     node._prev   = (ObjectWaiter *) 0xBAD ;
  17.     node.TState  = ObjectWaiter::TS_CXQ ;
  19.     ObjectWaiter * nxt ;
  20.     for (;;) {
  21.         // 头插_cxq
  22.         node._next = nxt = _cxq ;
  23.         if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;
  25.         // 还来？
  26.         if (TryLock (Self) > 0) {
  27.             return ;
  28.         }
  29.     }
  30.     // 省略部分代码...
  32.     for (;;) {
  33.         if (TryLock (Self) > 0) break ;
  34.         assert (_owner != Self, "invariant") ;
  35.         if ((SyncFlags & 2) && _Responsible == NULL) {
  36.            Atomic::cmpxchg_ptr (Self, &_Responsible, NULL) ;
  37.         }
  39.         // park self
  40.         if (_Responsible == Self || (SyncFlags & 1)) {
  41.             TEVENT (Inflated enter - park TIMED) ;
  42.             Self->_ParkEvent->park ((jlong) RecheckInterval) ;
  43.             // Increase the RecheckInterval, but clamp the value.
  44.             RecheckInterval *= 8 ;
  45.             if (RecheckInterval > 1000) RecheckInterval = 1000 ;
  46.         } else {
  47.             TEVENT (Inflated enter - park UNTIMED) ;
  48.             Self->_ParkEvent->park() ;
  49.         }
  51.         // 唤醒后又可以进行抢锁啦~
  52.         if (TryLock(Self) > 0) break ;
  53.         // 省略部分代码...
  54.     }
  55.     return ;
  56. }

果然synchronized不是公平锁，不过这也太不公平了。

解锁第一阶段
owner在退出持有锁的时候，会根据monitor的QMode策略，决定继承者的选取方式，选定继承者之前owner仍然会持有锁，以保证并行性。

  1. void ATTR ObjectMonitor::exit(bool not_suspended, TRAPS) {
  2.    // 省略部分代码...
  4.    // 重入次数递减至0
  5.    if (_recursions != 0) {
  6.      _recursions--;        // this is simple recursive enter
  7.      TEVENT (Inflated exit - recursive) ;
  8.      return ;
  9.    }
  11.    if ((SyncFlags & 4) == 0) {
  12.       _Responsible = NULL ;
  13.    }
  15.    // 自旋
  16.    for (;;) {
  17.       // (...) 省略部分代码
  19.       ObjectWaiter * w = NULL ;
  20.       int QMode = Knob_QMode ;
  22.       // 绕过EntryList，直接从_cxq中唤醒线程作为下一个继承者用于竞争锁
  23.       if (QMode == 2 && _cxq != NULL) {
  24.           w = _cxq ;
  25.           assert (w != NULL, "invariant") ;
  26.           assert (w->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
  27.           ExitEpilog (Self, w) ;
  28.           return ;
  29.       }
  31.       // 将_cxq队列中的线程移到_EntryList尾部
  32.       if (QMode == 3 && _cxq != NULL) {
  33.           w = _cxq ;
  34.           for (;;) {
  35.              assert (w != NULL, "Invariant") ;
  36.              ObjectWaiter * u = (ObjectWaiter *) Atomic::cmpxchg_ptr (NULL, &_cxq, w) ;
  37.              if (u == w) break ;
  38.              w = u ;
  39.           }
  40.           assert (w != NULL              , "invariant") ;
  42.           ObjectWaiter * q = NULL ;
  43.           ObjectWaiter * p ;
  44.           for (p = w ; p != NULL ; p = p->_next) {
  45.               guarantee (p->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
  46.               p->TState = ObjectWaiter::TS_ENTER ;
  47.               p->_prev = q ;
  48.               q = p ;
  49.           }
  51.           ObjectWaiter * Tail ;
  52.           for (Tail = _EntryList ; Tail != NULL && Tail->_next != NULL ; Tail = Tail->_next) ;
  53.           if (Tail == NULL) {
  54.               _EntryList = w ;
  55.           } else {
  56.               // _EntryList 的tail的next执行_cxq的头部
  57.               Tail->_next = w ;
  58.               w->_prev = Tail ;
  59.           }
  60.       }
  62.       // 将_cxq队列中的线程移到_EntryList头部
  63.       if (QMode == 4 && _cxq != NULL) {
  64.           // 如此可以保证最近竞争锁线程处于_EntryList的头部
  65.           w = _cxq ;
  66.           for (;;) {
  67.              assert (w != NULL, "Invariant") ;
  68.              ObjectWaiter * u = (ObjectWaiter *) Atomic::cmpxchg_ptr (NULL, &_cxq, w) ;
  69.              if (u == w) break ;
  70.              w = u ;
  71.           }
  72.           assert (w != NULL              , "invariant") ;
  74.           ObjectWaiter * q = NULL ;
  75.           ObjectWaiter * p ;
  76.           for (p = w ; p != NULL ; p = p->_next) {
  77.               guarantee (p->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
  78.               p->TState = ObjectWaiter::TS_ENTER ;
  79.               p->_prev = q ;
  80.               q = p ;
  81.           }
  83.           // 此时q为_cxq对了的tail线程
  84.           if (_EntryList != NULL) {
  85.               q->_next = _EntryList ;
  86.               _EntryList->_prev = q ;
  87.           }
  88.           _EntryList = w ;
  89.       }
  91.       // 若_EntryList不为空，QMode = 3 || QMode = 4 会唤醒_EntryList头部线程作为下一位继承者，并进行unpark操作
  92.       w = _EntryList  ;
  93.       if (w != NULL) {
  94.           assert (w->TState == ObjectWaiter::TS_ENTER, "invariant") ;
  95.           ExitEpilog (Self, w) ;
  96.           return ;
  97.       }
  99.       w = _cxq ;
  100.       if (w == NULL) continue ;
  102.       /*
  103.       * 能走到这里说明在这步采用线程进入_cxq队列，前面的操作中_cxq和_EntryList都是空队列
  104.       */
  105.       for (;;) {
  106.           assert (w != NULL, "Invariant") ;
  107.           ObjectWaiter * u = (ObjectWaiter *) Atomic::cmpxchg_ptr (NULL, &_cxq, w) ;
  108.           if (u == w) break ;
  109.           w = u ;
  110.       }
  111.       TEVENT (Inflated exit - drain cxq into EntryList) ;
  113.       assert (w != NULL              , "invariant") ;
  114.       assert (_EntryList  == NULL    , "invariant") ;
  116.       if (QMode == 1) {
  117.          ObjectWaiter * s = NULL ;
  118.          ObjectWaiter * t = w ;
  119.          ObjectWaiter * u = NULL ;
  120.          // 将_cxq队列反转，s为反转之后的_cxq
  121.          while (t != NULL) {
  122.              guarantee (t->TState == ObjectWaiter::TS_CXQ, "invariant") ;
  123.              t->TState = ObjectWaiter::TS_ENTER ;
  124.              u = t->_next ;
  125.              t->_prev = u ;
  126.              t->_next = s ;
  127.              s = t;
  128.              t = u ;
  129.          }
  130.          // 将反转倒序之后的_cxq放进_EntryList中
  131.          _EntryList  = s ;
  132.          assert (s != NULL, "invariant") ;
  133.       } else {
  134.          // QMode == 0 or QMode == 2
  135.          _EntryList = w ;
  136.          ObjectWaiter * q = NULL ;
  137.          ObjectWaiter * p ;
  138.          // 将_cxq由单向链表转为双向链表
  139.          for (p = w ; p != NULL ; p = p->_next) {
  140.              guarantee (p->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
  141.              p->TState = ObjectWaiter::TS_ENTER ;
  142.              p->_prev = q ;
  143.              q = p ;
  144.          }
  145.       }
  147.       if (_succ != NULL) continue;
  149.       w = _EntryList  ;
  150.       if (w != NULL) {
  151.           guarantee (w->TState == ObjectWaiter::TS_ENTER, "invariant") ;
  152.           ExitEpilog (Self, w) ;
  153.           return ;
  154.       }
  155.    }
  156. }

解锁第二阶段
唤醒继承者，让它去尝试获取锁。

  1. // 选取继承者、唤醒继承者队列的头部线程（代码就不看了）：
  2. void ObjectMonitor::ExitEpilog (Thread * Self, ObjectWaiter * Wakee) {
  3.    // Exit protocol:
  4.    // 1. ST _succ = wakee
  5.    // 2. membar #loadstore|#storestore;
  6.    // 2. ST _owner = NULL
  7.    // 3. unpark(wakee)
  8. }

总结

1：无论偏向锁、轻量级锁、重量级锁，都是可重入的。所以熟悉JAVA并发包的ReentrantLock重入锁机制是有必要的。
2：只有重量级锁需要操作系统去进行调度竞争锁的线程。
3：偏向锁的撤销不是为了使锁降级为无锁状态，而是需要先降级再转变为轻量级锁状态。
4：偏向锁的撤销需要等待全局安全点，且锁撤销有一定的开销。所以在多线程竞争激烈的情况下，可以实现关闭偏向锁来进行性能调优。

想看源码的看这些文件。
image

其他优化
JDK1.6 对锁的实现引入了大量的优化，如偏向锁、轻量级锁、自旋锁、适应性自旋锁、锁消除、锁粗化等技术来减少锁操作的开销。

①：适应性自旋
升级为重量级锁之前，会尝试自旋一定次数（默认10次，可通过参数-XX : PreBlockSpin来更改）来延缓进入重量级锁的过程。
优点：若真的成功则可以避免锁升级，减少线程进入monitor从而带来的一系列开销。同时当前线程不会经历挂起-唤醒的过程，可以更快响应。
缺点：会一直占用cpu，若自旋失败则是额外的浪费。

②：锁粗化
将连在一起的加锁、解锁操作扩大范围，只进行一次性加锁、解锁操作。
如：

  1.      Object lock = new Object();
  2.      List<String> list = new ArrayList();
  3.      synchronized(lock){
  4.          list.add("a");
  5.      }
  6.      synchronized(lock){
  7.          list.add("b");
  8.      }
  9.      synchronized(lock){
  10.          list.add("c");
  11.      }

优化为：

  1.      Object lock = new Object();
  2.      List<String> list = new ArrayList();
  3.      synchronized(lock){
  4.          list.add("a");
  5.          list.add("b");
  6.          list.add("c");
  7.      }

③：锁消除
若当前线程创建的对象分配在堆，但不会被其他线程使用，那么这段代码就可以不加锁。
或者根据逃逸分析，当前线程new的对象不会被其他线程使用，那么也不需要加锁。

其他问题
①：当所状态为偏向锁时，如何存储hashcode信息？
若hashCode方法的调用是在对象已经处于偏向锁状态时调用，它的偏向状态会被立即撤销，并且锁会升级为重量级锁。

②：什么线程复用？
两个线程间隔5s启动，markword中thread信息一摸一样这个现象实际上就是JVM线程复用。

本文参考文章：
①： 小米信息部技术团队-synchronized 实现原理
②：synchronized的jvm源码加锁流程分析聊锁的意义
③：Java对象的内存布局
④：盘一盘 synchronized （二）—— 偏向锁批量重偏向与批量撤销
⑤：https://www.bbsmax.com/A/xl56qY9rJr/
⑥：Java并发编程：Synchronized底层优化（偏向锁、轻量级锁）

感触：上网搜很难看到自己想要的内容，甚至有的文章还会起误导性作用。果然还是要好好学习，厉害的大佬比比皆是。在性能调优上哪有什么最优解，只有合适与不合适，重在选择与取舍。

版权声明：本文为竹根七原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/zgq7/p/16257327.html