高性能解决线程饥饿利器 StampedLock

做者：码哥字节 公众号：码哥字节

如需转载请联系我（微信ID）：MageByte1024

概览

在JDK 1.8 引入 StampedLock，能够理解为对 ReentrantReadWriteLock 在某些方面的加强，在原先读写锁的基础上新增了一种叫乐观读（Optimistic Reading）的模式。该模式并不会加锁，因此不会阻塞线程，会有更高的吞吐量和更高的性能。

它的设计初衷是做为一个内部工具类，用于开发其余线程安全的组件，提高系统性能，而且编程模型也比ReentrantReadWriteLock 复杂，因此用很差就很容易出现死锁或者线程安全等莫名其妙的问题。

跟着“码哥字节”带着问题一块儿学习StampedLock给咱们带来了什么…

• 有了ReentrantReadWriteLock，为什么还要引入StampedLock？
• 什么是乐观读？
• 在读多写少的并发场景下，StampedLock如何解决写线程难以获取锁的线程“饥饿”问题？
• 什么样的场景使用？
• 实现原理分析，是经过 AQS 实现仍是其余的？

特性

三种访问数据模式：

• Writing（独占写锁）：writeLock 方法会使线程阻塞等待独占访问，可类比ReentrantReadWriteLock 的写锁模式，同一时刻有且只有一个写线程获取锁资源；
• Reading（悲观读锁）：readLock方法，容许多个线程同时获取悲观读锁，悲观读锁与独占写锁互斥，与乐观读共享。
• Optimistic Reading（乐观读）：这里须要注意了，是乐观读，并无加锁。也就是不会有 CAS 机制而且没有阻塞线程。仅当当前未处于 Writing 模式 tryOptimisticRead 才会返回非 0 的邮戳（Stamp），若是在获取乐观读以后没有出现写模式线程获取锁，则在方法validate返回 true ，容许多个线程获取乐观读以及读锁。同时容许一个写线程获取写锁。

支持读写锁相互转换

ReentrantReadWriteLock 当线程获取写锁后能够降级成读锁，可是反过来则不行。

StampedLock提供了读锁和写锁相互转换的功能，使得该类支持更多的应用场景。

注意事项

1. StampedLock是不可重入锁，若是当前线程已经获取了写锁，再次重复获取的话就会死锁；
2. 都不支持 Conditon 条件将线程等待；
3. StampedLock 里的写锁和悲观读锁加锁成功以后，都会返回一个 stamp；而后解锁的时候，须要传入这个 stamp。

详解乐观读带来的性能提高

那为什么 StampedLock 性能比 ReentrantReadWriteLock 好？

关键在于StampedLock 提供的乐观读，咱们知道ReentrantReadWriteLock 支持多个线程同时获取读锁，可是当多个线程同时读的时候，全部的写线程都是阻塞的。

StampedLock 的乐观读容许一个写线程获取写锁，因此不会致使全部写线程阻塞，也就是当读多写少的时候，写线程有机会获取写锁，减小了线程饥饿的问题，吞吐量大大提升。

这里可能你就会有疑问，居然同时容许多个乐观读和一个先线程同时进入临界资源操做，那读取的数据多是错的怎么办？

是的，乐观读不能保证读取到的数据是最新的，因此将数据读取到局部变量的时候须要经过 lock.validate(stamp) 椒盐虾是否被写线程修改过，如果修改过则须要上悲观读锁，再从新读取数据到局部变量。

同时因为乐观读并非锁，因此没有线程唤醒与阻塞致使的上下文切换，性能更好。

其实跟数据库的“乐观锁”有殊途同归之妙，它的实现思想很简单。咱们举个数据库的例子。

在生产订单的表 product_doc 里增长了一个数值型版本号字段 version，每次更新 product_doc 这个表的时候，都将 version 字段加 1。

select id，... ，version
from product_doc
where id = 123

在更新的时候匹配 version 才执行更新。

update product_doc
set version = version + 1,...
where id = 123 and version = 5

数据库的乐观锁就是查询的时候将 version 查出来，更新的时候利用 version 字段验证，如果相等说明数据没有被修改，读取的数据是安全的。

这里的 version 就相似于 StampedLock 的 Stamp。

使用示例

模仿写一个将用户id与用户名数据保存在 共享变量 idMap 中，而且提供 put 方法添加数据、get 方法获取数据、以及 putIfNotExist 先从 map 中获取数据，若没有则模拟从数据库查询数据并放到 map 中。

public class CacheStampedLock {
    /**
     * 共享变量数据
     */
    private final Map<Integer, String> idMap = new HashMap<>();
    private final StampedLock lock = new StampedLock();

    /**
     * 添加数据，独占模式
     */
    public void put(Integer key, String value) {
        long stamp = lock.writeLock();
        try {
            idMap.put(key, value);
        } finally {
            lock.unlockWrite(stamp);
        }
    }

    /**
     * 读取数据，只读方法
     */
    public String get(Integer key) {
        // 1. 尝试经过乐观读模式读取数据，非阻塞
        long stamp = lock.tryOptimisticRead();
        // 2. 读取数据到当前线程栈
        String currentValue = idMap.get(key);
        // 3. 校验是否被其余线程修改过,true 表示未修改，不然须要加悲观读锁
        if (!lock.validate(stamp)) {
            // 4. 上悲观读锁，并从新读取数据到当前线程局部变量
            stamp = lock.readLock();
            try {
                currentValue = idMap.get(key);
            } finally {
                lock.unlockRead(stamp);
            }
        }
        // 5. 若校验经过，则直接返回数据
        return currentValue;
    }

    /**
     * 若是数据不存在则从数据库读取添加到 map 中,锁升级运用
     * @param key
     * @param value 能够理解成从数据库读取的数据，假设不会为 null
     * @return
     */
    public String putIfNotExist(Integer key, String value) {
        // 获取读锁，也能够直接调用 get 方法使用乐观读
        long stamp = lock.readLock();
        String currentValue = idMap.get(key);
        // 缓存为空则尝试上写锁从数据库读取数据并写入缓存
        try {
            while (Objects.isNull(currentValue)) {
                // 尝试升级写锁
                long wl = lock.tryConvertToWriteLock(stamp);
                // 不为 0 升级写锁成功
                if (wl != 0L) {
                    // 模拟从数据库读取数据, 写入缓存中
                    stamp = wl;
                    currentValue = value;
                    idMap.put(key, currentValue);
                    break;
                } else {
                    // 升级失败，释放以前加的读锁并上写锁，经过循环再试
                    lock.unlockRead(stamp);
                    stamp = lock.writeLock();
                }
            }
        } finally {
            // 释放最后加的锁
            lock.unlock(stamp);
        }
        return currentValue;
    }
}

上面的使用例子中，须要引发注意的是 get()和 putIfNotExist() 方法，第一个使用了乐观读，使得读写能够并发执行，第二个则是使用了读锁转换成写锁的编程模型，先查询缓存，当不存在的时候从数据库读取数据并添加到缓存中。

在使用乐观读的时候必定要按照固定模板编写，不然很容易出 bug，咱们总结下乐观读编程模型的模板：

public void optimisticRead() {
    // 1. 非阻塞乐观读模式获取版本信息
    long stamp = lock.tryOptimisticRead();
    // 2. 拷贝共享数据到线程本地栈中
    copyVaraibale2ThreadMemory();
    // 3. 校验乐观读模式读取的数据是否被修改过
    if (!lock.validate(stamp)) {
        // 3.1 校验未经过，上读锁
        stamp = lock.readLock();
        try {
            // 3.2 拷贝共享变量数据到局部变量
            copyVaraibale2ThreadMemory();
        } finally {
            // 释放读锁
            lock.unlockRead(stamp);
        }
    }
    // 3.3 校验经过，使用线程本地栈的数据进行逻辑操做
    useThreadMemoryVarables();
}

使用场景和注意事项

对于读多写少的高并发场景 StampedLock 的性能很好，经过乐观读模式很好的解决了写线程“饥饿”的问题，咱们可使用StampedLock 来代替ReentrantReadWriteLock ，可是须要注意的是 StampedLock 的功能仅仅是 ReadWriteLock 的子集，在使用的时候，仍是有几个地方须要注意一下。

1. StampedLock 是不可重入锁，使用过程当中必定要注意；
2. 悲观读、写锁都不支持条件变量 Conditon ，当须要这个特性的时候须要注意；
3. 若是线程阻塞在 StampedLock 的 readLock() 或者 writeLock() 上时，此时调用该阻塞线程的 interrupt() 方法，会致使 CPU 飙升。因此，使用 StampedLock 必定不要调用中断操做，若是须要支持中断功能，必定使用可中断的悲观读锁 readLockInterruptibly() 和写锁 writeLockInterruptibly()。这个规则必定要记清楚。

原理分析

咱们发现它并不像其余锁同样经过定义内部类继承 AbstractQueuedSynchronizer抽象类而后子类实现模板方法实现同步逻辑。可是实现思路仍是有相似，依然使用了 CLH 队列来管理线程，经过同步状态值 state 来标识锁的状态。

其内部定义了不少变量，这些变量的目的仍是跟 ReentrantReadWriteLock 同样，将状态为按位切分，经过位运算对 state 变量操做用来区分同步状态。

好比写锁使用的是第八位为 1 则表示写锁，读锁使用 0-7 位，因此通常状况下获取读锁的线程数量为 1-126，超过之后，会使用 readerOverflow int 变量保存超出的线程数。

自旋优化

对多核 CPU 也进行必定优化，NCPU 获取核数，当核数目超过 1 的时候，线程获取锁的重试、入队钱的重试都有自旋操做。主要就是经过内部定义的一些变量来判断，如图所示。

等待队列

队列的节点经过 WNode 定义，如上图所示。等待队列的节点相比 AQS 更简单，只有三种状态分别是：

• 0：初始状态；
• -1：等待中；
• 取消；

另外还有一个字段 cowait ，经过该字段指向一个栈，保存读线程。结构如图所示

同时定义了两个变量分别指向头结点与尾节点。

/** Head of CLH queue */
private transient volatile WNode whead;
/** Tail (last) of CLH queue */
private transient volatile WNode wtail;

另外有一个须要注意点就是 cowait， 保存全部的读节点数据，使用的是头插法。

当读写线程竞争造成等待队列的数据以下图所示：

获取写锁

public long writeLock() {
    long s, next;  // bypass acquireWrite in fully unlocked case only
    return ((((s = state) & ABITS) == 0L &&
             U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
            next : acquireWrite(false, 0L));
}

获取写锁，若是获取失败则构建节点放入队列，同时阻塞线程，须要注意的时候该方法不响应中断，如需中断须要调用 writeLockInterruptibly()。不然会形成高 CPU 占用的问题。

(s = state) & ABITS 标识读锁和写锁未被使用，那么久直接执行 U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) CAS 操做将第八位设置 1，标识写锁占用成功。CAS失败的话则调用 acquireWrite(false, 0L)加入等待队列，同时将线程阻塞。

另外acquireWrite(false, 0L) 方法很复杂，运用大量自旋操做，好比自旋入队列。

获取读锁

public long readLock() {
    long s = state, next;  // bypass acquireRead on common uncontended case
    return ((whead == wtail && (s & ABITS) < RFULL &&
             U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ?
            next : acquireRead(false, 0L));
}

获取读锁关键步骤

(whead == wtail && (s & ABITS) < RFULL若是队列为空而且读锁线程数未超过限制，则经过 U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))CAS 方式修改 state 标识获取读锁成功。

不然调用 acquireRead(false, 0L) 尝试使用自旋获取读锁，获取不到则进入等待队列。

acquireRead

当 A 线程获取了写锁，B 线程去获取读锁的时候，调用 acquireRead 方法，则会加入阻塞队列，并阻塞 B 线程。方法内部依然很复杂，大体流程梳理后以下：

1. 若是写锁未被占用，则当即尝试获取读锁，经过CAS修改状态为标志成功则直接返回。
2. 若是写锁被占用，则将当前线程包装成 WNode 读节点，并插入等待队列。若是是写线程节点则直接放入队尾，不然放入队尾专门存放读线程的 WNode cowait 指向的栈。栈结构是头插法的方式插入数据，最终唤醒读节点，从栈顶开始。

释放锁

不管是 unlockRead 释放读锁仍是 unlockWrite释放写锁，整体流程基本都是经过 CAS 操做，修改 state 成功后调用 release 方法唤醒等待队列的头结点的后继节点线程。

1. 想将头结点等待状态设置为 0 ，标识即将唤醒后继节点。
2. 唤醒后继节点经过CAS方式获取锁，若是是读节点则会唤醒 cowait 锁指向的栈全部读节点。

释放读锁

unlockRead(long stamp) 若是传入的 stamp 与锁持有的 stamp 一致，则释放非排它锁，内部主要是经过自旋 + CAS 修改 state 成功，在修改 state 以前作了判断是否超过读线程数限制，如果小于限制才经过CAS 修改 state 同步状态，接着调用 release 方法唤醒 whead 的后继节点。

释放写锁

unlockWrite(long stamp) 若是传入的 stamp 与锁持有的 stamp 一致，则释放写锁，whead 不为空，且当前节点状态 status ！= 0 则调用 release 方法唤醒头结点的后继节点线程。

总结

StampedLock 并不能彻底代替ReentrantReadWriteLock ，在读多写少的场景下由于乐观读的模式，容许一个写线程获取写锁，解决了写线程饥饿问题，大大提升吞吐量。

在使用乐观读的时候须要注意按照编程模型模板方式去编写，不然很容易形成死锁或者意想不到的线程安全问题。

它不是可重入锁，且不支持条件变量 Conditon。而且线程阻塞在 readLock() 或者 writeLock() 上时，此时调用该阻塞线程的 interrupt() 方法，会致使 CPU 飙升。若是须要中断线程的场景，必定要注意调用悲观读锁 readLockInterruptibly() 和写锁 writeLockInterruptibly()。

另外唤醒线程的规则和 AQS 相似，先唤醒头结点，不一样的是 StampedLock 唤醒的节点是读节点的时候，会唤醒此读节点的 cowait 锁指向的栈的全部读节点，可是唤醒与插入的顺序相反。

推荐阅读

如下几篇文章阅读量与读者反馈都很好，推荐你们阅读：

• RESTful 最佳实践
• Tomcat 架构原理解析到架构设计借鉴
• Tomcat 高并发之道原理拆解与性能调优
• 数据库系统设计概述

参考内容

[Java多线程进阶（十一）—— J.U.C之locks框架：StampedLock]