StampedLock原理分析

背景

ReentrantReadWriteLock可以让多个读线程同时持有读锁（只要写锁未被占用），而写锁是独占的，所以在读多写少的场景上可以提高吞吐量，比如hdfs文件系统。但是，读写锁如果使用不当，很容易产生“饥饿”问题，比如在读线程非常多，写线程很少的情况下，很容易导致写线程“饥饿”，虽然使用“公平”策略可以一定程度上缓解这个问题，但是“公平”策略是以牺牲系统吞吐量为代价的。

导致写线程饥饿的情况：当线程 A 持有读锁读取数据时，线程 B 要获取写锁修改数据就只能到队列里排队。此时又来了线程 C 读取数据，那么线程 C 就可以获取到读锁，而要执行写操作线程 B 就要等线程 C 释放读锁。由于该场景下读操作远远大于写的操作，此时可能会有很多线程来读取数据而获取到读锁，那么要获取写锁的线程 B 就只能一直等待下去，最终导致饥饿。

由于读写锁ReentrantReadWriteLock以上的问题，所以引入了StampedLock，可以这样理解：StampedLock 是对读写锁的增强与优化。下面我们逐步解析下其中的具体原理。

StampedLock简介

StampedLock是JUC并发包里面JDK1.8版本新增的一个锁。从上述篇幅可知，ReentrantReadWriteLock导致写线程饥饿的原因是读锁和写锁互斥，StampedLock 提供了解决这一问题的方案 - 乐观读 Optimistic reading，即一个线程获取的乐观读锁之后，不会阻塞线程获取写锁。

该锁提供了三种模式的读写控制，当调用获取锁的系列函数的时候，会返回一个long 型的变量，该变量被称为戳记（stamp),这个戳记代表了锁的状态。try系列获取锁的函数，当获取锁失败后会返回为0的stamp值。当调用释放锁和转换锁的方法时候需要传入获取锁时候返回的stamp值。

StampedLockd的内部实现是基于CLH锁的，CLH锁原理：锁维护着一个等待线程队列，所有申请锁且失败的线程都记录在队列。一个节点代表一个线程，保存着一个标记位locked,用以判断当前线程是否已经释放锁。当一个线程试图获取锁时，从队列尾节点作为前序节点，循环判断所有的前序节点是否已经成功释放锁。如下图所示：

StampedLock提供了三种模式来控制读写操作：写锁 writeLock、悲观读锁 readLock、乐观读 Optimistic reading。

写锁 writeLock

类似 ReentrantReadWriteLock 的写锁，独占锁，当一个线程获取该锁后，其它请求的线程必须等待。没有线程持有悲观读锁或者写锁的时候才可以获取到该锁。请求该锁成功后会返回一个 stamp 票据变量用来表示该锁的版本，当释放该锁时候需要将这个 stamp 作为参数传入解锁方法。

悲观读锁 readLock

类似 ReentrantReadWriteLock 的读锁，共享锁，同时多个线程可以获取该锁。在没有线程获取独占写锁的情况下，同时多个线程可以获取该锁。请求该锁成功后会返回一个 stamp 票据变量用来表示该锁的版本，当释放该锁时候需要 unlockRead 并传递参数 stamp。

悲观读锁：悲观的认为在具体操作数据前其他线程会对自己操作的数据进行修改，所以当前线程获取到悲观读锁的之后会阻塞线程获取写锁。

写锁与悲观读锁的相关使用如下：

final StampedLock sl = new StampedLock();// 获取/释放悲观读锁示意代码long stamp = sl.readLock();try {  //省略业务相关代码} finally {  sl.unlockRead(stamp);}// 获取/释放写锁示意代码long stamp = sl.writeLock();try {  //省略业务相关代码} finally {  sl.unlockWrite(stamp);}

乐观读 OptimisticReading

StampedLock 的性能之所以比 ReadWriteLock 还要好，其关键是 StampedLock 支持乐观读的方式。ReadWriteLock 支持多个线程同时读，但是当多个线程同时读的时候，所有的写操作会被阻塞；而 StampedLock 提供的乐观读，是允许一个线程获取写锁的，也就是说不是所有的写操作都被阻塞。

注意这里，我们用的是“乐观读”这个词，而不是“乐观读锁”，是要提醒你，乐观读这个操作是无锁的，所以相比较 ReadWriteLock 的读锁，乐观读的性能更好一些。

获取的时候，不需要通过 CAS 设置锁的状态，如果当前没有线程持有写锁，直接简单的返回一个非 0 的 stamp 版本信息，表示获取锁成功。释放的时候，并没有使用 CAS 设置锁状态所以不需要显示的释放该锁。

乐观读如何保证数据一致性的呢？

乐观读在获取 stamp 时，会将需要的数据拷贝一份出来。在真正进行读取操作时，验证 stamp 是否可用。如何验证 stamp 是否可用呢？从获取 stamp 到真正进行读取操作这段时间内，如果有线程获取了写锁，stamp 就失效了。如果 stamp 可用就可以直接读取原来拷贝出来的数据，如果 stamp 不可用，就重新拷贝一份出来用。我们操作的是方法栈里面的数据，也就是一个快照，所以最多返回的不是最新的数据，但是一致性还是得到保障的。

乐观读：乐观的认为在具体操作数据前其他线程不会对自己操作的数据进行修改，所以当前线程获取到乐观读锁的之后不会阻塞线程获取写锁。

为了保证数据一致性，在具体操作数据前要检查一下自己操作的数据是否经过修改操作了，如果进行了修改操作，就重新读一次。

乐观读在读多写少的情况下提供更好的性能，因为乐观读不需要进行 CAS 设置锁的状态而只是简单的测试状态。

文中下面这段代码是出自 Java SDK 官方示例，并略做了修改。在 distanceFromOrigin() 这个方法中，首先通过调用 tryOptimisticRead() 获取了一个stamp，这里的 tryOptimisticRead() 就是我们前面提到的乐观读。之后将共享变量 x 和 y 读入方法的局部变量中，不过需要注意的是，由于 tryOptimisticRead() 是无锁的，所以共享变量 x 和 y 读入方法局部变量时，x 和 y 有可能被其他线程修改了。因此最后读完之后，还需要再次验证一下是否存在写操作，这个验证操作是通过调用 validate(stamp) 来实现的。

class Point {    private int x, y;    final StampedLock sl = new StampedLock();    //计算到原点的距离      double distanceFromOrigin() {        // 乐观读        long stamp = sl.tryOptimisticRead();        // 读入局部变量，        // 读的过程数据可能被修改        int curX = x, curY = y;        //判断执行读操作期间，        //是否存在写操作，如果存在，        //则sl.validate返回false        if (!sl.validate(stamp)) {            // 升级为悲观读锁            stamp = sl.readLock();            try {                curX = x;                curY = y;            } finally {                //释放悲观读锁                sl.unlockRead(stamp);            }        }        return Math.sqrt(curX * curX + curY * curY);    }}

在上面这个代码示例中，如果执行乐观读操作的期间，存在写操作，会把乐观读升级为悲观读锁。这个做法挺合理的，否则你就需要在一个循环里反复执行乐观读，直到执行乐观读操作的期间没有写操作（只有这样才能保证 x 和 y 的正确性和一致性），而循环读会浪费大量的 CPU。升级为悲观读锁，代码简练且不易出错，建议你在具体实践时也采用这样的方法。

如果你曾经用过数据库的乐观锁，可能会发现 StampedLock 的乐观读和数据库的乐观锁有异曲同工之妙。数据库里的乐观锁，查询的时候需要把 version 字段查出来，更新的时候要利用 version 字段做验证。这个 version 字段就类似于 StampedLock 里面的 stamp。这样对比着看，相信你会更容易理解 StampedLock 里乐观读的用法。

StampedLock实现原理

StampedLock虽然不像其它锁一样定义了内部类来实现AQS框架，但是StampedLock的基本实现思路还是利用CLH队列进行线程的管理，通过同步状态值来表示锁的状态和类型。

StampedLock内部定义了很多常量，定义这些常量的根本目的还是和ReentrantReadWriteLock一样，对同步状态值按位切分，以通过位运算对State进行操作.

锁状态

StampedLock 提供了写锁、悲观读锁、乐观读三种模式的锁，如何维护锁状态呢？StampedLock 的锁状态用 long 类型的 state 表示，类似ReentrantReadWriteLock，通过将 state 按位切分的方式表示不同的锁状态。

悲观读锁：state 的前 7 位（0-7 位）表示获取读锁的线程数，如果超过 0-7 位的最大容量 126，则使用一个名为 readerOverflow 的 int 整型保存超出数。

写锁：state 第 8 位为写锁标志，0 表示未被占用，1 表示写锁被占用。state 第 8-64 位表示写锁的获取次数，次数超过 64 位最大容量则重新从 1 开始。

乐观读：不需要维护锁状态，但是在具体操作数据前要检查一下自己操作的数据是否经过修改操作，也就是验证是否有线程获取过写锁。

如果只用第 8 位来标志写锁，那么来看乐观写锁的使用过程：

检查是否有写锁，state 第 8 位为 0，没有写锁，拷贝数据；
检查是否有线程获取过写锁，state 第 8 位为 0，没有线程获取过，直接使用原来拷贝的数据。

发现其中的问题了吗？第一次检查 state 第 8 位为 0 之后，有线程获取写锁修改数据并释放了写锁，那么之后在检查是否有线程获取过写锁时 state 第 8 位还是 0，认为没有线程获取过写锁，可能导致数据不一致。

也就是 ABA 问题， ABA 问题的解决办法就是加版本号，将原来的 A->B->A 就变成了 1A->2B->3A。StampedLock 同样采用这种方法，将获取写锁的次数作为版本号，也就是乐观读锁的票据，写锁释放时次数加 1，也就是 state 第 8 位加 1。

state原始状态为     //...0001 0000 0000
获取写锁            //...0001 1000 0000
释放写锁次数加1      //...0010 0000 0000
获取写锁           // ...0010 1000 0000
释放写锁次数加1     //...0011 0000 0000

JDK 设计的精妙之处还在于，获取写锁后 state 第 8 位为 1，释放写锁时 state 第 8 位加 1 使第 8 位变回 0，既记录了写锁次数，又可以保证 state 的第 8 位一个位置来标志写锁

属性

锁状态相关属性

// 一个单位的读锁        0000... 0000 0000 0001
private static final long RUNIT = 1L;   // 一个单位的写锁        0000... 0000 1000 0000
private static final long WBIT = 1L << LG_READERS;   // 读状态标识            0000... 0000 0111 1111
private static final long RBITS = WBIT - 1L; // 读锁最大数量          0000... 0000 0111 1110
private static final long RFULL = RBITS - 1L;    // 用于获取读写状态      0000... 0000 1111 1111
private static final long ABITS = RBITS | WBIT;  //                       1111... 1111 1000 0000
private static final long SBITS = ~RBITS;               // 锁state初始值，0000... 0001 0000 0000
private static final long ORIGIN = WBIT << 1;/** 锁队列状态， 当处于写模式时第8位为1，读模式时前7为为1-126
（附加的readerOverflow用于当读者超过126时） */
private transient volatile long state;/** 将state超过 RFULL=126的值放到readerOverflow字段中 */
private transient int readerOverflow;

给出这些常量的比特位，等下看源码过程中会频繁用到：

节点

StampedLock中，等待队列的结点要比 AQS 中简单些，仅仅三种状态。0：初始状态；-1：等待中；1：取消。结点的定义中有个 cowait 字段，该字段指向一个栈，用于保存读线程。

// 结点状态
private static final int WAITING = -1;
private static final int CANCELLED = 1;// 结点的读写模式
private static final int RMODE = 0;
private static final int WMODE = 1;/** Wait nodes */
static final class WNode {volatile WNode prev;volatile WNode next;volatile WNode cowait; // 读模式使用该结点形成栈volatile Thread thread; // non-null while possibly parkedvolatile int status; // 0, WAITING, or CANCELLEDfinal int mode; // RMODE or WMODEWNode(int m, WNode p) {mode = m;prev = p;}
}/** CLH队头结点 */
private transient volatile WNode whead;
/** CLH队尾结点 */
private transient volatile WNode wtail;

写锁的获取与释放

写锁的获取：

可以获取写锁的条件：没有线程占用悲观读锁和写锁；
获取写锁，state 写锁位加 1，此时写锁标志位变为 1，返回邮戳 stamp;
获取失败，加入同步队列等待被唤醒。

写锁的释放：

传入获取写锁时的 stamp 验证；
stamp 值被修改，抛出异常；
stamp 正确，state 写锁位加 1，此时写锁标志位变为 0；
唤醒同步队列等锁线程。

/** * 获取写锁，如果获取失败，进入阻塞 */public long writeLock() {    long s, next;    return ((((s = state) & ABITS) == 0L && // 没有读写锁             U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?              // cas操作尝试获取写锁             // 获取成功后返回next，失败则进行后续处理，排队也在后续处理中            next : acquireWrite(false, 0L)); }/** * 释放写锁 */public void unlockWrite(long stamp) {    WNode h;    //stamp值被修改，或者写锁已经被释放，抛出错误    if (state != stamp || (stamp & WBIT) == 0L)         throw new IllegalMonitorStateException();        //加0000 1000 0000来记录写锁的变化，同时改变写锁状态    state = (stamp += WBIT) == 0L ? ORIGIN : stamp;     if ((h = whead) != null && h.status != 0)        release(h);// 唤醒等待队列的队首结点}/** * 尝试自旋的获取写锁, 获取不到则阻塞线程 * * @param interruptible true 表示检测中断, 如果线程被中断过,  * 则最终返回INTERRUPTED * @param deadline      如果非0, 则表示限时获取 * @return 非0表示获取成功, INTERRUPTED表示中途被中断过 */private long acquireWrite(boolean interruptible, long deadline) {    WNode node = null, p;    /**     * 自旋入队操作     * 如果没有任何锁被占用, 则立即尝试获取写锁, 获取成功则返回.     * 如果存在锁被使用, 则将当前线程包装成独占结点, 并插入等待队列尾部     */    for (int spins = -1; ; ) {        long m, s, ns;        if ((m = (s = state) & ABITS) == 0L) {      // 没有任何锁被占用            if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))    // 尝试立即获取写锁                return ns;                                                 // 获取成功直接返回        } else if (spins < 0)            spins = (m == WBIT && wtail == whead) ? SPINS : 0;        else if (spins > 0) {            if (LockSupport.nextSecondarySeed() >= 0)                --spins;        } else if ((p = wtail) == null) {       // 队列为空, 则初始化队列, 构造队列的头结点            WNode hd = new WNode(WMODE, null);            if (U.compareAndSwapObject(this, WHEAD, null, hd))                wtail = hd;        } else if (node == null)               // 将当前线程包装成写结点            node = new WNode(WMODE, p);        else if (node.prev != p)            node.prev = p;        else if (U.compareAndSwapObject(this, WTAIL, p, node)) {          // 链接结点至队尾              p.next = node;            break;        }    }    for (int spins = -1; ; ) {        WNode h, np, pp;        int ps;        // 如果当前结点是队首结点, 则立即尝试获取写锁        if ((h = whead) == p) {                 if (spins < 0)                spins = HEAD_SPINS;            else if (spins < MAX_HEAD_SPINS)                spins <<= 1;            for (int k = spins; ; ) { // spin at head                long s, ns;                if (((s = state) & ABITS) == 0L) {  // 写锁未被占用                    if (U.compareAndSwapLong(this, STATE, s,                        ns = s + WBIT)) {   // CAS修改State: 占用写锁                        // 将队首结点从队列移除                        whead = node;                        node.prev = null;                        return ns;                    }                } else if (LockSupport.nextSecondarySeed() >= 0                 && --k <= 0)                    break;            }        } else if (h != null) {  // 唤醒头结点的栈中的所有读线程            WNode c;            Thread w;            while ((c = h.cowait) != null) {                if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait)                 && (w = c.thread) != null)                    U.unpark(w);            }        }        if (whead == h) {            if ((np = node.prev) != p) {                if (np != null)                    (p = np).next = node;   // stale            } else if ((ps = p.status) == 0)            // 将当前结点的前驱置为WAITING,             // 表示当前结点会进入阻塞, 前驱将来需要唤醒我                        U.compareAndSwapInt(p, WSTATUS, 0, WAITING);            else if (ps == CANCELLED) {                if ((pp = p.prev) != null) {                    node.prev = pp;                    pp.next = node;                }            } else {        // 阻塞当前调用线程                long time;  // 0 argument to park means no timeout                if (deadline == 0L)                    time = 0L;                else if ((time = deadline - System.nanoTime()) <= 0L)                    return cancelWaiter(node, node, false);                Thread wt = Thread.currentThread();                U.putObject(wt, PARKBLOCKER, this);                node.thread = wt;                if (p.status < 0 && (p != h || (state & ABITS) != 0L)                   && whead == h && node.prev == p)                  // emulate LockSupport.park                    U.park(false, time);                    node.thread = null;                U.putObject(wt, PARKBLOCKER, null);                if (interruptible && Thread.interrupted())                    return cancelWaiter(node, node, true);            }        }    }}

悲观锁的获取与释放

悲观锁的获取：

获取悲观读锁条件：没有线程占用写锁；
读锁标志位+1，返回邮戳 stamp;
获取失败加入同步队列。

悲观锁的释放：

传入邮戳 stamp 验证；
stamp 验证失败，抛异常；
stamp 验证成功，读锁标志位-1，唤醒同步队列等锁线程。

/** * 获取悲观读锁，如果写锁被占用，线程阻塞 */public long readLock() {    long s = state, next;    //队列为空，无写锁，同时读锁未溢出，尝试获取读锁    return ((whead == wtail && (s & ABITS) < RFULL       && U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) ?           //cas尝试获取读锁+1      //获取读锁成功，返回s + RUNIT，失败进入后续处理，类似acquireWrite            next : acquireRead(false, 0L));     }/** * 释放悲观读锁 */public void unlockRead(long stamp) {    long s, m; WNode h;    for (;;) {        if (((s = state) & SBITS) != (stamp & SBITS)         || (stamp & ABITS) == 0L || (m = s & ABITS) == 0L || m == WBIT)            throw new IllegalMonitorStateException();        //小于最大记录值（最大记录值127超过后放在readerOverflow变量中）            if (m < RFULL) {          //cas尝试释放读锁-1            if (U.compareAndSwapLong(this, STATE, s, s - RUNIT)) {                  if (m == RUNIT && (h = whead) != null && h.status != 0)                    release(h);                break;            }        }        else if (tryDecReaderOverflow(s) != 0L) //readerOverflow - 1            break;    }}

乐观读的获取

乐观读锁因为实际上没有获取过锁，所以也就没有释放锁的过程。

/** * 尝试获取乐观锁 * 写锁被占用，返回state第8-64位的写锁记录；没被占用返回0 */public long tryOptimisticRead() {    long s;    return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;}/** * 验证乐观锁获取之后是否有过写操作 */public boolean validate(long stamp) {  // 之前的所有load操作在内存屏障之前完成，对应的还有storeFence()及fullFence()    U.loadFence();     return (stamp & SBITS) == (state & SBITS);  //比较是否有过写操作}

另外，StampedLock相比ReentrantReadWriteLock，对多核CPU进行了优化，可以看到，当CPU核数超过1时，会有一些自旋操作:

注意事项

对于读多写少的场景 StampedLock 性能很好，简单的应用场景基本上可以替代 ReadWriteLock，但是 StampedLock 的功能仅仅是 ReadWriteLock 的子集，在使用的时候，还是有几个地方需要注意一下：

StampedLock 在命名上并没有增加 Reentrant，想必你已经猜测到 StampedLock 应该是不可重入的。事实上，的确是这样的，StampedLock 不支持重入。这个是在使用中必须要特别注意的；
StampedLock 的悲观读锁、写锁都不支持条件变量；
需要特别注意，那就是：如果线程阻塞在 StampedLock 的 readLock() 或者 writeLock() 上时，此时调用该阻塞线程的 interrupt() 方法，会导致 CPU 飙升。例如下面的代码中，线程 T1 获取写锁之后将自己阻塞，线程 T2 尝试获取悲观读锁，也会阻塞；如果此时调用线程 T2 的 interrupt() 方法来中断线程 T2 的话，你会发现线程 T2 所在 CPU 会飙升到 100%。

final StampedLock lock = new StampedLock();Thread T1 = new Thread(()->{  // 获取写锁  lock.writeLock();  // 永远阻塞在此处，不释放写锁  LockSupport.park();});T1.start();// 保证T1获取写锁Thread.sleep(100);Thread T2 = new Thread(()->  //阻塞在悲观读锁  lock.readLock());T2.start();// 保证T2阻塞在读锁Thread.sleep(100);//中断线程T2//会导致线程T2所在CPU飙升T2.interrupt();T2.join();

所以，使用 StampedLock 一定不要调用中断操作，如果需要支持中断功能，一定使用可中断的悲观读锁 readLockInterruptibly() 和写锁 writeLockInterruptibly()。这个规则一定要记清楚。

StampedLock 支持锁的降级（通过 tryConvertToReadLock() 方法实现）和升级（通过 tryConvertToWriteLock() 方法实现），但是建议你要慎重使用。下面的代码也源自 Java 的官方示例，隐藏了一个 Bug，就是在锁升级成功的时候，最后没有释放最新的写锁，可以在if块的break上加个stamp=ws进行释放。

private double x, y;final StampedLock sl = new StampedLock();// 存在问题的方法void moveIfAtOrigin(double newX, double newY){ long stamp = sl.readLock(); try {  while(x == 0.0 && y == 0.0){    long ws = sl.tryConvertToWriteLock(stamp);    if (ws != 0L) {      x = newX;      y = newY;      break;    } else {      sl.unlockRead(stamp);      stamp = sl.writeLock();    }  } } finally {  sl.unlock(stamp);}

重点回顾

读写锁在读线程非常多，写线程很少的情况下可能会导致写线程饥饿，JDK1.8 新增的StampedLock通过乐观读锁来解决这一问题。StampedLock有三种访问模式：

写锁 writeLock：功能和读写锁的写锁类似；
悲观读锁 readLock：功能和读写锁的读锁类似；
乐观读 Optimistic reading：一种优化的读模式。

乐观读：乐观的认为在具体操作数据前其他线程不会对自己操作的数据进行修改，所以当前线程获取到乐观读的之后不会阻塞线程获取写锁。为了保证数据一致性，在具体操作数据前要检查一下自己操作的数据是否经过修改操作了，如果进行了修改操作，就重新读一次。因为乐观读不需要进行 CAS 设置锁的状态而只是简单的测试状态，所以在读多写少的情况下有更好的性能。

所有获取锁的方法，都返回一个票据 Stamp，Stamp 为 0 表示获取失败，其余都表示成功；所有释放锁的方法，都需要一个票据 Stamp，这个 Stamp 必须是和成功获取锁时得到的 Stamp 一致。

StampedLock 通过将 state 按位切分的方式表示不同的锁状态。

悲观读锁：state 的 0-7 位表示获取读锁的线程数，如果超过 0-7 位的最大容量 126，则使用一个名为 readerOverflow 的 int 整型保存超出数。

写锁：state 第 8 位为写锁标志，0 表示未被占用，1 表示写锁被占用。state 第 8-64 位表示写锁的获取次数，次数超过 64 位最大容量则重新从 1 开始。