分析与实战：高性能读写印戳锁StampedLock

向右看齐 2024-03-17 11:14 61阅读 0赞

#### 文章目录 ####

*   *  前言
     *  StampedLock原理
     *  StampedLock源码解读
     *   *  CLH双向队列缓存阻塞节点
         *  加解锁使用CAS修改STATE
         *  乐观读和锁的转换
     *  实战演示
     *   *  1、创建缓存工具类
         *  2、创建测试用例
         *  3、查看测试结果
     *  写在最后

### 前言 ###

在之前的博文中我们讲到了[可重入读写锁ReentrantReadWriteLock][ReentrantReadWriteLock]，读写锁机制是读写互斥、写写互斥、读读共享。这里面就有一个问题，在线程加的读锁是一个悲观读锁，会阻塞其他线程加写锁。在我们高并发读多写少的场景下会影响写操作，验证将阻塞写操作以致于影响业务操作。那有没有一种方式可以在线程对资源加读锁后，资源还能够被加写锁呢，只要读锁线程在操作数据之前进行票据验证，不就能够达到数据的一致性了吗。所以，我们引入今天的主角——高性能读写印戳锁StampedLock。

### StampedLock原理 ###

StampedLock就如同它的含义一样，是一个贴上邮票印记的锁，我们叫它印戳锁。印戳印戳就是在进行加锁的时候会会返回一个印戳，在解锁的时候会传入这个印戳来保证加解锁为同一个线程。

> 和其他的锁机制一样，印戳锁StampedLock和AQS一样都是用CLH虚拟双向FIFO队列实现同步功能，并定义STATE值标识队列头部线程占用资源状态。由于StampedLock都是加的非公平锁，当队列头部节点线程获取到资源后会增加STATE值，释放锁会减少STATE值，如果STATE == 0会唤醒后续节点的自旋线程获取资源。

> 当然，印戳锁StampedLock最重要的是它的乐观读锁。大家都知道普通读写锁是互斥的，加了读锁是不能够加写锁的。但是印戳锁加了乐观读锁，其他线程是可以继续加写锁，只是在乐观读锁线程进行数据操作时候需要进行数据验证，以保证数据的一致性。

### StampedLock源码解读 ###

#### CLH双向队列缓存阻塞节点 ####

对于StampedLock的源码解读，其实意义并不是很大。大概就是提供了虚拟CLH双向队列来保存阻塞节点，通过STATE值来表示资源是否能够被加锁。如下源码所示，提供了CLH双向队列来满足锁机制：

//等待节点
    static final class WNode {
        volatile WNode prev;
        volatile WNode next;
        volatile WNode cowait;    // list of linked readers
        volatile Thread thread;   // non-null while possibly parked
        volatile int status;      // 0, WAITING, or CANCELLED
        final int mode;           // RMODE or WMODE
        WNode(int m, WNode p) { mode = m; prev = p; }
    }

#### 加解锁使用CAS修改STATE ####

查看加解锁源码：

//获取普通阻塞写锁
    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));
    }
    
    /**
     * 
     *获取非阻塞写锁
     */
    public long tryWriteLock() {
        long s, next;
        return ((((s = state) & ABITS) == 0L &&
                 U.compareAndSwapLong(this, STATE, s, next = s + WBIT)) ?
                next : 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));
    }
    
    /**
     * 获取非阻塞读锁
     */
    public long tryReadLock() {
        for (;;) {
            long s, m, next;
            if ((m = (s = state) & ABITS) == WBIT)
                return 0L;
            else if (m < RFULL) {
                if (U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))
                    return next;
            }
            else if ((next = tryIncReaderOverflow(s)) != 0L)
                return next;
        }
    }
    
    //写锁解锁
    public void unlockWrite(long stamp) {
        WNode h;
        if (state != stamp || (stamp & WBIT) == 0L)
            throw new IllegalMonitorStateException();
        state = (stamp += WBIT) == 0L ? ORIGIN : stamp;
        if ((h = whead) != null && h.status != 0)
            release(h);
    }
    
    //读锁解锁
    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();
            if (m < RFULL) {
                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)
                break;
        }
    }

> 如上源码所示，印戳锁的加解锁都提供了阻塞与非阻塞机制。加锁后通过CAS增加STATE值，并返回stamp票据，在解锁后通过传入的stamp票据进行线程验证保证加解锁为同一个线程，并使用CAS减少STATE值。解锁操作的同时会验证当前节点是否是头节点，是否存在后续节点，如果存在后续阻塞节点会唤醒该节点占用资源。

#### 乐观读和锁的转换 ####

印戳锁提供了乐观读锁和读锁与写锁转换、写锁和读锁转换方法。

//获取乐观锁并返回票据
    public long tryOptimisticRead() {
        long s;
        return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
    }
    
    //验证票据是否改变
    public boolean validate(long stamp) {
        U.loadFence();
        return (stamp & SBITS) == (state & SBITS);
    }

> 如上所示，StampedLock提供获取乐观锁的方法，获取到乐观读锁并返回一个票据。我们在实际业务中需要使用时需要通过票据再次验证是否已经更改，如果更改则表示数据已经失效，需要重新获取。

对于锁的转换我们继续查看源码：

/**
     * 将锁转为写锁
     */
    public long tryConvertToWriteLock(long stamp) {
        long a = stamp & ABITS, m, s, next;
        while (((s = state) & SBITS) == (stamp & SBITS)) {
            if ((m = s & ABITS) == 0L) {
                if (a != 0L)
                    break;
                if (U.compareAndSwapLong(this, STATE, s, next = s + WBIT))
                    return next;
            }
            else if (m == WBIT) {
                if (a != m)
                    break;
                return stamp;
            }
            else if (m == RUNIT && a != 0L) {
                if (U.compareAndSwapLong(this, STATE, s,
                                         next = s - RUNIT + WBIT))
                    return next;
            }
            else
                break;
        }
        return 0L;
    }
    
    /**
     * 将锁转为读锁
     */
    public long tryConvertToReadLock(long stamp) {
        long a = stamp & ABITS, m, s, next; WNode h;
        while (((s = state) & SBITS) == (stamp & SBITS)) {
            if ((m = s & ABITS) == 0L) {
                if (a != 0L)
                    break;
                else if (m < RFULL) {
                    if (U.compareAndSwapLong(this, STATE, s, next = s + RUNIT))
                        return next;
                }
                else if ((next = tryIncReaderOverflow(s)) != 0L)
                    return next;
            }
            else if (m == WBIT) {
                if (a != m)
                    break;
                state = next = s + (WBIT + RUNIT);
                if ((h = whead) != null && h.status != 0)
                    release(h);
                return next;
            }
            else if (a != 0L && a < WBIT)
                return stamp;
            else
                break;
        }
        return 0L;
    }
    
    /**
     * 将锁转为客观锁
     */
    public long tryConvertToOptimisticRead(long stamp) {
        long a = stamp & ABITS, m, s, next; WNode h;
        U.loadFence();
        for (;;) {
            if (((s = state) & SBITS) != (stamp & SBITS))
                break;
            if ((m = s & ABITS) == 0L) {
                if (a != 0L)
                    break;
                return s;
            }
            else if (m == WBIT) {
                if (a != m)
                    break;
                state = next = (s += WBIT) == 0L ? ORIGIN : s;
                if ((h = whead) != null && h.status != 0)
                    release(h);
                return next;
            }
            else if (a == 0L || a >= WBIT)
                break;
            else if (m < RFULL) {
                if (U.compareAndSwapLong(this, STATE, s, next = s - RUNIT)) {
                    if (m == RUNIT && (h = whead) != null && h.status != 0)
                        release(h);
                    return next & SBITS;
                }
            }
            else if ((next = tryDecReaderOverflow(s)) != 0L)
                return next & SBITS;
        }
        return 0L;
    }

> 如上源码所示，印戳锁提供了读锁转为写锁、写锁转为读锁、写锁可以转为乐观读锁、读锁转为乐观读锁。需要注意的是在进行乐观读锁转换后，需要传入票据验证数据是否已经失效。

### 实战演示 ###

其实印戳锁StampedLock的使用场景就是读多写少的场景，这种场景普通读写锁也能够满足。但是印戳锁增加了乐观读机制，这样会让写锁更加的高效。

> 以下实战演示用内存缓存数据进行演示，实际生产中不建议内存保存缓存，应当采用响应的内存缓存数据库，比如redis等等。

#### 1、创建缓存工具类 ####

/**
     * StampedLockDemo
     * 模拟缓存
     * @author senfel
     * @version 1.0
     * @date 2023/5/25 11:23
     */
    @Slf4j
    public class StampedLockDemo {
    
        /**
         * 创建一个印戳锁
         */
        private static final StampedLock stampedLock = new StampedLock();
    
        /**
         * map内存缓存模拟缓存
         * 实际使用场景建议用缓存数据库redis等
         */
        private static final Map<String,String> cacheMap = new HashMap<>();
    
    
        /**
         * 添加缓存
         * @param key
         * @param value
         * @author senfel
         * @date 2023/5/25 11:26
         * @return void
         */
        public static Boolean putCache(String key,String value){
            long stamp = stampedLock.writeLock();
            try{
                //获取到写锁
                log.error("线程{}获取到写锁，进行缓存写入",Thread.currentThread().getName());
                cacheMap.put(key,value);
                log.error("线程{}写入数据当前缓存中的数据有：{}",Thread.currentThread().getName(), JSONObject.toJSONString(cacheMap));
                return true;
            }catch (Exception e){
                log.error("线程{}添加缓存异常：{}",Thread.currentThread().getName(),e.getMessage());
                return false;
            }finally {
               stampedLock.unlockWrite(stamp);
            }
        }
    
        /**
         * 获取缓存
         * @param key
         * @author senfel
         * @date 2023/5/25 11:34
         * @return java.lang.String
         */
        public static String getCache(String key){
            try{
                long stamp = stampedLock.tryOptimisticRead();
                String value = null;
                if(0 != stamp){
                    log.error("线程{}获取到乐观读锁",Thread.currentThread().getName());
                    value = cacheMap.get(key);
                    if(!stampedLock.validate(stamp)){
                        //校验不通过降级为悲观读
                        log.error("线程{}获取到乐观读锁校验不通过降级为悲观读",Thread.currentThread().getName());
                        try {
                            stamp = stampedLock.readLock();
                            value = cacheMap.get(key);
                        }catch (Exception e){
                            throw e;
                        }finally {
                            stampedLock.unlockRead(stamp);
                        }
    
                    }
                }else{
                    log.error("线程{}未获取到乐观读锁，尝试悲观读",Thread.currentThread().getName());
                    try {
                        stamp = stampedLock.readLock();
                        value = cacheMap.get(key);
                    }catch (Exception e){
                        throw e;
                    }finally {
                        stampedLock.unlockRead(stamp);
                    }
                }
                log.error("线程{}获取到的数据为：{}",Thread.currentThread().getName(), value);
                return value;
            }catch (Exception e){
                log.error("线程{}获取缓存异常：{}",Thread.currentThread().getName(),e.getMessage());
            }
            return null;
        }
    
    }

#### 2、创建测试用例 ####

/**
     * 印戳锁测试
     * @author senfel
     * @date 2023/5/25 12:38
     * @return void
     */
    @Test
    public void stampedLockTest() throws Exception{
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        //等待结束
        CountDownLatch countDownLatch = new CountDownLatch(6);
        //保证同时执行
        CyclicBarrier cyclicBarrier = new CyclicBarrier(6);
        for(int i=0;i<3;i++){
            int finalI = i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try{
                        cyclicBarrier.await();
                    }catch (Exception e){
                        log.error("线程相互等待异常：{}",e.getMessage());
                    }
                    StampedLockDemo.putCache(finalI +"", "senfel"+finalI);
                    countDownLatch.countDown();
                }
            });
        }
        for(int i=0;i<3;i++){
            int finalI = i;
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try{
                        cyclicBarrier.await();
                    }catch (Exception e){
                        log.error("线程相互等待异常：{}",e.getMessage());
                    }
                    StampedLockDemo.getCache(finalI+"");
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.error("测试完成，关闭线程池");
    }

#### 3、查看测试结果 ####

> 线程pool-1-thread-4获取到乐观读锁  
> 线程pool-1-thread-2获取到写锁，进行缓存写入  
> 线程pool-1-thread-5获取到乐观读锁  
> 线程pool-1-thread-6获取到乐观读锁  
> 线程pool-1-thread-4获取到乐观读锁校验不通过降级为悲观读  
> 线程pool-1-thread-5获取到乐观读锁校验不通过降级为悲观读  
> 线程pool-1-thread-6获取到乐观读锁校验不通过降级为悲观读  
> 线程pool-1-thread-2写入数据当前缓存中的数据有：\{“1”:“senfel1”\}  
> 线程pool-1-thread-3获取到写锁，进行缓存写入  
> 线程pool-1-thread-3写入数据当前缓存中的数据有：\{“1”:“senfel1”,“2”:“senfel2”\}  
> 线程pool-1-thread-1获取到写锁，进行缓存写入  
> 线程pool-1-thread-1写入数据当前缓存中的数据有：\{“0”:“senfel0”,“1”:“senfel1”,“2”:“senfel2”\}  
> 线程pool-1-thread-5获取到的数据为：senfel1  
> 线程pool-1-thread-6获取到的数据为：senfel2  
> 线程pool-1-thread-4获取到的数据为：senfel0

### 写在最后 ###

印戳锁StampedLock是一个高性能的读写锁，使用CLH队列来保存等待线程节点，STATE来标识资源加解锁。印戳锁内部提供了读锁、写锁、乐观读锁，当使用乐观读锁的时候其他线程是可以继续加写锁，只是在乐观读锁线程进行数据操作时候需要进行数据验证，以保证数据一致性。

[ReentrantReadWriteLock]: https://blog.csdn.net/weixin_39970883/article/details/130807620?spm=1001.2014.3001.5502