编程语言
首页 > 编程语言> > Java中的偏向锁，轻量级锁， 重量级锁解析

Java中的偏向锁，轻量级锁， 重量级锁解析

作者：互联网

1. Java 中的锁

在 Java 中主要2种加锁机制:

2. 一些先修知识

2.1  Java 对象头

字宽（Word）: 内存大小的单位概念， 对于 32 位处理器 1 Word = 4 Bytes， 64 位处理器 1 Word = 8 Bytes

每一个 Java 对象都至少占用 2 个字宽的内存(数组类型占用3个字宽)。

非数组类型的对象头的结构如下图

在这里插入图片描述

说明：

下面的图片来自参考论文 Eliminating Synchronization-Related Atomic Operations with Biased Locking and Bulk Rebiasing ， 可以与上面的表格进行比对参照， 更为清晰, 可以看出来， 标志位（tag bits）可以直接确定唯一的一种锁状态

在这里插入图片描述

2.2 CAS指令

CAS （Compare And Swap） 指令是一个CPU层级的原子性操作指令。 在 Intel 处理器中， 其汇编指令为 cmpxchg。

该指令概念上存在 3 个参数， 第一个参数【目标地址】， 第二个参数【值1】， 第三个参数【值2】， 指令会比较【目标地址存储的内容】和 【值1】 是否一致， 如果一致， 则将【值 2】 填写到【目标地址】， 其语义可以用如下的伪代码表示。

function cas(p , old , new ) returns bool {
    if *p ≠ old { // *p 表示指针p所指向的内存地址
        return false
    }
    *p ← new
    return true
}

注意： 该指令是是原子性的， 也就是说 CPU 执行该指令时， 是不会被中断执行其他指令的。

2.3 “CAS”实现的"无锁"算法常见误区

2.3.1 误区一： 通过简单应用 “比较后再赋值” 的操作即可轻松实现很多无锁算法

CAS 指令的一个不可忽略的特征是原子性。 在 CPU 层面， CAS 指令的执行是有原子性语义保证的， 如果 CAS 操作放在应用层面来实现， 则需要我们自行保证其原子性。 否则就会发生如下描述的问题：

// 下列的函数如果不是线程互斥的, 是错误的 CAS 实现
function cas( p , old , new) returns bool {
    if *p ≠ old { // 此处的比较操作进行时， 可以同时有多个线程通过该判断
        return false
    }
    *p ← new // 多个线程的赋值操作会相互覆盖， 造成程序逻辑的错误
    return true
}

2.3.2 误区二： CAS 操作的 ABA 问题

大部分网络博文对 ABA 问题的常见描述是： 应用 CAS 操作时， 目标地址的值刚开始为 A， 工作线程/进程 读取后， 进行了一系列运算， 计算得出了新值 C， 在此期间， 目标地址的值被其他线程已经进行了不止一次修改， 其值已经从 A 被改为 B ， 又改回 A， 此时便会发生同步问题。

上面的描述是其实是错误的， 思考一下就会发现， 如果工作线程的操作目的是将目标地址的值从 A 改为 C， 那么即便在这期间目标地址的值经过了其他线程或进程的多次修改， 其语义依旧是正确的。

例如目前要将某银行账号的余额扣除 50， 通过 CAS 保证同步 ：

  1. 首先读取原有余额为 100
  2. 计算余额应该赋值为 100 - 50 = 50
  3. 此时该线程被挂起， 该账户同时又发生了转入 150 和转出 150 的操作， 余额经历了 100 -》250 -》100 的变动
  4. 线程被唤醒， 进行 CAS 赋值操作 cas（p， 100， 50） ， 正常得以执行。
  5. 该账户的余额依旧是正确的

通过上述例子就可以发现， ABA 的问题并不在于多次修改。 查阅一下 CAS 的 wiki 解释， 就会发现， ABA 真正的问题是， 假如目标地址的内容被多次修改以后， 虽然从二进制上来看是依旧是 A， 但是其语义已经不是 A 。例如， 发生了整数溢出， 内存回收等等。

2.4 栈帧（Stack Frame） 的概念

2.5 轻量级加锁的过程

轻量级加锁的过程在参考文章一中有较为的描述以及配图， 这里直接将其摘抄过来， 做轻微整理和调整。

图2.1

在这里插入图片描述

2.6 重量级加锁的过程

3. synchronized 关键字之锁的升级（偏向锁->轻量级锁->重量级锁）

前面提到过， synchronized 代码块是由一对 monitorenter/moniterexit 字节码指令实现， monitor 是其同步实现的基础， Java SE1.6 为了改善性能， 使得 JVM 会根据竞争情况， 使用如下 3 种不同的锁机制

上述这三种机制的切换是根据竞争激烈程度进行的， 在几乎无竞争的条件下， 会使用偏向锁， 在轻度竞争的条件下， 会由偏向锁升级为轻量级锁， 在重度竞争的情况下， 会升级到重量级锁。

注意 JVM 提供了关闭偏向锁的机制， JVM 启动命令指定如下参数即可

-XX:-UseBiasedLocking

下图展现了一个对象在创建（allocate） 后， 根据偏斜锁机制是否打开， 对象 MarkWord 状态以不同方式转换的过程

这里写图片描述

上图在参考文章一中的中文翻译对照图如下

在这里插入图片描述

 

4. 无锁 -> 偏向锁

在这里插入图片描述

从上图可以看到 ， 偏向锁的获取方式是将对象头的 MarkWord 部分中， 标记上线程ID， 以表示哪一个线程获得了偏向锁。 具体的赋值逻辑如下：

这里写图片描述

下面是 Open Jdk/ JDK 8 源码 中检测一个对象是否处于可偏向状态的源码

  // Indicates that the mark has the bias bit set but that it has not
  // yet been biased toward a particular thread
  bool is_biased_anonymously() const {
    return (has_bias_pattern() && (biased_locker() == NULL));
  }

应评论区一位朋友的提问， 进一步摘抄一下markOop.hpp中的方法定义

// Biased Locking accessors.
  // These must be checked by all code which calls into the
  // ObjectSynchronizer and other code. The biasing is not understood
  // by the lower-level CAS-based locking code, although the runtime
  // fixes up biased locks to be compatible with it when a bias is
  // revoked.
  bool has_bias_pattern() const {
    return (mask_bits(value(), biased_lock_mask_in_place) == biased_lock_pattern);
  }
  JavaThread* biased_locker() const {
    assert(has_bias_pattern(), "should not call this otherwise");
    return (JavaThread*) ((intptr_t) (mask_bits(value(), ~(biased_lock_mask_in_place | age_mask_in_place | epoch_mask_in_place))));
  }

从上面的偏向锁机制描述中，可以注意到：偏向锁的 撤销（revoke） 是一个很特殊的操作， 为了执行撤销操作， 需要等待全局安全点（Safe Point）， 此时间点所有的工作线程都停止了字节码的执行。

5. 偏向锁的撤销（Revoke）

如上文提到的， 偏向锁的撤销（Revoke） 操作并不是将对象恢复到无锁可偏向的状态， 而是在偏向锁的获取过程中， 发现了竞争时， 直接将一个被偏向的对象“升级到” 被加了轻量级锁的状态。 这个操作的具体完成方式如下：

在偏向锁 CAS 更新操作失败以后， 等待到达全局安全点。

6. 偏向锁的批量再偏向（Bulk Rebias）机制

偏向锁这个机制很特殊， 别的锁在执行完同步代码块后， 都会有释放锁的操作， 而偏向锁并没有直观意义上的“释放锁”操作。

那么作为开发人员， 很自然会产生的一个问题就是， 如果一个对象先偏向于某个线程， 执行完同步代码后， 另一个线程就不能直接重新获得偏向锁吗？ 答案是可以， JVM 提供了批量再偏向机制（Bulk Rebias）机制

该机制的主要工作原理如下：

上述的逻辑可以在 JDK 源码中得到验证。

在 sharedRuntime.cpp 中， 下面代码是 synchronized 的主要逻辑

Handle h_obj(THREAD, obj);
  if (UseBiasedLocking) {
    // Retry fast entry if bias is revoked to avoid unnecessary inflation
    ObjectSynchronizer::fast_enter(h_obj, lock, true, CHECK);
  } else {
    ObjectSynchronizer::slow_enter(h_obj, lock, CHECK);
  }

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);
}

该函数中再次保险性地做了偏斜锁是否开启的检查（UseBiasedLocking）。

当系统不处于安全点时， 代码通过方法 revoke_and_rebias 这个函数尝试获取偏斜锁， 如果获取成功就可以直接返回了， 如果不成功则进入轻量级锁的获取过程。

revoke_and_rebias 这个函数名称就很有意思， 说明该函数中包含了 revoke 的操作也包含了 rebias 的操作

revoke_and_rebias 函数的定义在 biasedLocking.cpp

BiasedLocking::Condition BiasedLocking::revoke_and_rebias(Handle obj, bool attempt_rebias, TRAPS) {
  assert(!SafepointSynchronize::is_at_safepoint(), "must not be called while at safepoint");

  // We can revoke the biases of anonymously-biased objects
  // efficiently enough that we should not cause these revocations to
  // update the heuristics because doing so may cause unwanted bulk
  // revocations (which are expensive) to occur.
  markOop mark = obj->mark();
  if (mark->is_biased_anonymously() && !attempt_rebias) {
      /* 
		    进一步查看源码可得知， is_biased_anonymously() 为 true 的条件是对象处于可偏向状态， 
		    且 线程ID  为空， 表示尚未偏向于任意一个线程。 
		    此分支是进行对象的 hashCode 计算时会进入的， 根据 markWord 结构可以看到， 
		    当一个对象处于可偏向状态时， markWord 中 hashCode 的存储空间是被占用的
		    所以需要 revoke 可偏向状态， 以提供存储 hashCode 的空间
		 */
    
    // We are probably trying to revoke the bias of this object due to
    // an identity hash code computation. Try to revoke the bias
    // without a safepoint. This is possible if we can successfully
    // compare-and-exchange an unbiased header into the mark word of
    // the object, meaning that no other thread has raced to acquire
    // the bias of the object.
    markOop biased_value       = mark;
    markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());
    markOop res_mark = obj->cas_set_mark(unbiased_prototype, mark);
    if (res_mark == biased_value) {
      return BIAS_REVOKED;
    }
  } else if (mark->has_bias_pattern()) {
    Klass* k = obj->klass();
    markOop prototype_header = k->prototype_header();
    if (!prototype_header->has_bias_pattern()) {
      // This object has a stale bias from before the bulk revocation
      // for this data type occurred. It's pointless to update the
      // heuristics at this point so simply update the header with a
      // CAS. If we fail this race, the object's bias has been revoked
      // by another thread so we simply return and let the caller deal
      // with it.
      markOop biased_value       = mark;
      markOop res_mark = obj->cas_set_mark(prototype_header, mark);
      assert(!obj->mark()->has_bias_pattern(), "even if we raced, should still be revoked");
      return BIAS_REVOKED;
    } else if (prototype_header->bias_epoch() != mark->bias_epoch()) { 
      // The epoch of this biasing has expired indicating that the
      // object is effectively unbiased. Depending on whether we need
      // to rebias or revoke the bias of this object we can do it
      // efficiently enough with a CAS that we shouldn't update the
      // heuristics. This is normally done in the assembly code but we
      // can reach this point due to various points in the runtime
      // needing to revoke biases.
      if (attempt_rebias) {
	    /*
			下面的代码就是尝试通过 CAS 操作， 将本线程的 ThreadID 尝试写入对象头中
		*/
        assert(THREAD->is_Java_thread(), "");
        markOop biased_value       = mark;
        markOop rebiased_prototype = markOopDesc::encode((JavaThread*) THREAD, mark->age(), prototype_header->bias_epoch());
        markOop res_mark = obj->cas_set_mark(rebiased_prototype, mark);
        if (res_mark == biased_value) {
          return BIAS_REVOKED_AND_REBIASED;
        }
      } else {
        markOop biased_value       = mark;
        markOop unbiased_prototype = markOopDesc::prototype()->set_age(mark->age());
        markOop res_mark = obj->cas_set_mark(unbiased_prototype, mark);
        if (res_mark == biased_value) {
          return BIAS_REVOKED;
        }
      }
    }
  }

7. 偏向锁 -> 轻量级锁

从之前的描述中可以看到， 存在超过一个线程竞争某一个对象时， 会发生偏向锁的撤销操作。 有趣的是， 偏向锁撤销后， 对象可能处于两种状态。

一种是不可偏向的无锁状态， 如下图（之所以不允许偏向， 是因为已经检测到了多于一个线程的竞争， 升级到了轻量级锁的机制）

这里写图片描述

另一种是不可偏向的已锁 ( 轻量级锁) 状态

这里写图片描述

之所以会出现上述两种状态， 是因为偏向锁不存在解锁的操作， 只有撤销操作。 触发撤销操作时：

轻量级加锁过程：

首先根据标志位判断出对象状态处于不可偏向的无锁状态( 如下图）

这里写图片描述

在当前线程的栈桢（Stack Frame）中创建用于存储锁记录（lock record）的空间，并将对象头中的Mark Word复制到锁记录中，官方称为Displaced Mark Word。如果在此过程中发现；

然后线程尝试使用 CAS 操作将对象头中的 Mark Word 替换为指向锁记录的指针。

下图展示了两个线程竞争锁， 最终导致锁膨胀为重量级锁的过程。

**注意： 下图中第一个标绿 MarkWord 的起始状态是HashCode|age|0|01 是偏向锁未被启用时， 分配对象后的状态， 所以在图中并没有偏向锁这一流程的体现， 是直接从无锁状态进入了轻量级锁的状态。

这里写图片描述

8. 重量级锁

重量级锁依赖于操作系统的互斥量（mutex） 实现， 其具体的详细机制此处暂不展开， 日后可能补充。 此处暂时只需要了解该操作会导致进程从用户态与内核态之间的切换， 是一个开销较大的操作。

9. 存疑的问题

  1. 在锁膨胀的图例中， 线程 2 在线程 1 尚未释放锁时， 即将对象头修改为指向重量级锁的状态， 这个操作具体如何完成， 是否需要等待全局安全点？笔者尚未细究
  2. 轻量级锁的第一次获取时， 如果 CAS 操作失败， 按照 聊聊并发（二）Java SE1.6中的Synchronized 的描述， 会进行自旋的尝试。 但按照 Synchronization and Object Locking 的描述， 会去检测已加的锁是归属于自身线程， 没有提到自旋操作。 具体哪一种是正确的行为， 有待研究源码。
  3. biasedLocking.cpp中的方法 revoke_and_rebias 存在 4 个条件分支， 其中笔者添加了注释的两个分支其主要逻辑已经清晰， 但未添加注释的两个分支具体逻辑笔者尚不清楚， 有待进一步研究。
fedorafrog 发布了8 篇原创文章 · 获赞 0 · 访问量 8498 私信 关注

标签：Java,对象,mark,CAS,bias,线程,轻量级,重量级,偏向
来源： https://blog.csdn.net/fedorafrog/article/details/104490363