• 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏吧

源码分析:升级版的读写锁 StampedLock

开发技术 开发技术 1周前 (11-19) 6次浏览

简介

StampedLock 是JDK1.8 开始提供的一种锁, 是对之前介绍的读写锁 ReentrantReadWriteLock 的功能增强。StampedLock 有三种模式:Writing(读)、Reading(写)、Optimistic Reading(乐观度),StampedLock 的功能不是基于AQS来实现的,而是完全自己内部实现的功能,不支持重入。在加锁的时候会返回一个戳,解锁的时候需要传入,匹配完成解锁操作。

官方使用示例

class Point {
    private double x, y;
    private final StampedLock sl = new StampedLock();

    void move(double deltaX, double deltaY) { // an exclusively locked method
        // 写锁-独占资源
        long stamp = sl.writeLock();
        try {
            x += deltaX;
            y += deltaY;
        } finally {
            sl.unlockWrite(stamp);
        }
    }

    double distanceFromOrigin() { // A read-only method
        // 只读的方法,比较乐观,认为读的过程中不会有写,所以这里是乐观度
        long stamp = sl.tryOptimisticRead();
        double currentX = x, currentY = y;
        if (!sl.validate(stamp)) { // 检查乐观读锁后是否有其他写锁发生
            // 获取一个普通的读锁
            stamp = sl.readLock();
            try {
                currentX = x;
                currentY = y;
            } finally {
                // 释放读锁
                sl.unlockRead(stamp);
            }
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }

    void moveIfAtOrigin(double newX, double newY) { // upgrade
        // Could instead start with optimistic, not read mode
        long stamp = sl.readLock();
        try {
            while (x == 0.0 && y == 0.0) {
                // 普通读锁转换成写锁,返回0为转换失败
                long ws = sl.tryConvertToWriteLock(stamp);
                if (ws != 0L) {
                    stamp = ws;
                    x = newX;
                    y = newY;
                    break;
                }  else {
                    sl.unlockRead(stamp);
                    stamp = sl.writeLock();
                }
            }
        } finally {
            sl.unlock(stamp);
        }
    }
}

官方demo中用到的 api 主要有获取写锁(writeLock())、释放写锁(unlockWrite(stamp))、获取普通读锁(readLock())、释放普通读锁(unlockRead(stamp))、获取乐观读锁(tryOptimisticRead())、检测乐观读版本(validate(stamp))、普通读锁转换成写锁(tryConvertToWriteLock(stamp))。下面分析源码的时候也主要根据这几个方法来分析。

源码分析

主要内部类

  1. 等待节点:WNode
    用于维护 CLH 队列的节点,源码如下:

    static final class WNode {
        volatile WNode prev;  // 前驱节点
        volatile WNode next;  // 后继节点
        volatile WNode cowait;    // 链接的读者列表
        volatile Thread thread;   // 线程
        volatile int status;      // 状态 0, WAITING, or CANCELLED
        final int mode;           // 两种模式:RMODE or WMODE
        WNode(int m, WNode p) { mode = m; prev = p; }
    }
    
  2. ReadWriteLockView:实现了ReadWriteLock接口,提供了读写锁获取接口

    final class ReadWriteLockView implements ReadWriteLock {
        public Lock readLock() { return asReadLock(); }
        public Lock writeLock() { return asWriteLock(); }
    }
    
  3. ReadLockView 和 WriteLockView:都实现了Lock接口,并实现了所有的方法

主要属性

  1. CLH 队列

    /** Head of CLH queue */
    private transient volatile WNode whead;
    /** Tail (last) of CLH queue */
    private transient volatile WNode wtail;
    
  2. 其他常量属性

    /** CPU的核心数量,用来控制自旋的次数 */
    private static final int NCPU = Runtime.getRuntime().availableProcessors();
    /** 获取锁入队前最大重试次数:CPU核心数大于1:64,否则0 */
    private static final int SPINS = (NCPU > 1) ? 1 << 6 : 0;
    /** 等待队列的头结点,获取锁最大重试次数:CPU核心数大于1:1024,否则0 */
    private static final int HEAD_SPINS = (NCPU > 1) ? 1 << 10 : 0;
    /** 再次阻塞前最大重试次数:CPU核心数大于1:65536,否则0  */
    private static final int MAX_HEAD_SPINS = (NCPU > 1) ? 1 << 16 : 0;
    /** The period for yielding when waiting for overflow spinlock */
    private static final int OVERFLOW_YIELD_RATE = 7; // must be power 2 - 1
    
    /** 用于读取器计数的位数 */
    private static final int LG_READERS = 7;
    
    // 用来计算state值的常量
    private static final long RUNIT = 1L;  // 读单位
    // 写锁的标识位 十进制:128  二进制位标示:1000 0000 
    private static final long WBIT  = 1L << LG_READERS;  
    // 读状态标识 admol 十进制:127  二进制: 0111 1111
    private static final long RBITS = WBIT - 1L;   
    // 读锁的最大标识  十进制:126 二进制 :0111 1110 
    private static final long RFULL = RBITS - 1L;  
    // 用来读取读写状态  十进制:255 二进制:1111 1111
    private static final long ABITS = RBITS | WBIT; 
    // ~255 ==  11111111111111111111111111111111111111111111111111111111 1000 0000
    // -128
    private static final long SBITS = ~RBITS; 
    
    // 同步状态state的初始值 256  二进制:0001 0000 0000
    private static final long ORIGIN = WBIT << 1;
    
    // 中断
    private static final long INTERRUPTED = 1L;
    
    // 节点的状态
    private static final int WAITING   = -1;
    private static final int CANCELLED =  1;
    
    // 节点的模式 
    private static final int RMODE = 0;
    private static final int WMODE = 1;
    
    /** 同步状态 初始值 256 0001 0000 0000*/
    private transient volatile long state;
    /** 读计数饱和时的额外读取器计数 */
    private transient int readerOverflow;
    

    StampedLock 虽然没有继承AQS,但是属性上很相似,都有一个CLH队列,和一个同步状态值state, StampedLock用8位来表示读写锁状态,前7位是用来标识读锁状态的,第8位标识写锁占用,如果读锁数量超过了126(0111 1110 ),超出的用readerOverflow来计数。

构造方法

public StampedLock() {
   // 初始值 256,  二进制:0001 0000 0000
    state = ORIGIN;
}

获取写锁:writeLock()

源码展示:

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

代码分析:

  1. 首先执行的是 ((s = state) & ABITS)== 0L,用来表示读锁和写锁是否可以被获取
    解析:第一次时,state 初始值是256,ABITS是255,计算过程:0001 0000 0000 && 0000 1111 1111 ,结算结果为0。
  2. 执行CAS操作:U.compareAndSwapLong(this, STATE, s, next = s + WBIT))
    解析:STATE 是state字段的内存地址相对于此对象的内存地址的偏移量,s 是期望值, next = s + WBIT 是更新值;s+WBIT 也就是 256 + 128,next计算结果为384,用二进制位表示就是0001 1000 0000,也就是将第8位设置为1,就是获得写锁。如果CAS 更新成功,返回next值,成功获得写锁。
  3. 如果CAS执行失败,则执行acquireWrite(false, 0L) ,进入等待队列获取锁

acquireWrite 代码分析:

// interruptible 是否要检查中断
// deadline:0 一直等待获取锁
private long acquireWrite(boolean interruptible, long deadline) {
    // node:即将入队排队的节点
    // p:当前排队节点入队之前的尾节点
    WNode node = null, p;
    // 第一次自旋:排队节点入队列自旋
    for (int spins = -1;;) { // spin while enqueuing
        long m, s, ns;
        // 这个if 和外面一样的,从代码运行到这期间所有没有被释放
        if ((m = (s = state) & ABITS) == 0L) {
            // CAS 再次尝试获取下写锁
            if (U.compareAndSwapLong(this, STATE, s, ns = s + WBIT))
                // 成功获取写锁
                return ns;
        }  else if (spins < 0)  // 走到这,说明上面还是没获取到写锁,写锁被占用了,m的值为128
            // 1. 确定自旋的次数 spins: 64 or 0 
            spins = (m == WBIT && wtail == whead) ? SPINS : 0;
        else if (spins > 0) {
            // 2.自旋的次数大于0,随机减一次自旋次数,直到减到spins为0(by.精灵王 这里实际是空转,没什么特点的逻辑处理)
            if (LockSupport.nextSecondarySeed() >= 0)
                --spins;
        } else if ((p = wtail) == null) { // initialize queue
            // 3.自旋spins减到0后会立马执行到这里
            // p被赋值为尾节点  
            // 初始化队列, WMODE:写,null:前驱节点
            WNode hd = new WNode(WMODE, null);
            if (U.compareAndSwapObject(this, WHEAD, null, hd))
                // 初始化队列时,尾节点等于头节点
                wtail = hd;
        } else if (node == null)
            // 4.初始化队列后,下一次自旋,构建当前排队节点,并指定了其尾节点
            node = new WNode(WMODE, p);
        else if (node.prev != p) // 如果当前节点的前驱不是尾节点
            // 5.前驱节点设置为之前队列的尾节点
            node.prev = p;
        else if (U.compareAndSwapObject(this, WTAIL, p, node)) {
            // 6. CAS 更新尾节点为当前排队的节点 
            p.next = node;
            // 退出自旋
            break;
        }
    }
    // 第二次自旋
    for (int spins = -1;;) {
        WNode h, np, pp; int ps; 
        if ((h = whead) == p) { // 如果头节点和之前的尾节点p是同一个, 说明马上应该轮到node节点获得锁
            if (spins < 0)
                // ① 设置自旋次数  1024 or 0
                spins = HEAD_SPINS;
            else if (spins < MAX_HEAD_SPINS)
                // 第一次自旋1024次没有获取到锁,这次自旋翻倍
                // 自旋次数*2  2048   继续进入到下面的自旋
                spins <<= 1;
            for (int k = spins;;) { // spin at head
                // ② 第三次自旋,不断尝试获得锁(自旋1024或者2048次),直到成功获得锁 或者 break
                long s, ns;
                // ((s = state) & ABITS) == 0L  表示锁没有被占用
                if (((s = state) & ABITS) == 0L) {
                    // CAS 修改state值
                    if (U.compareAndSwapLong(this, STATE, s,  ns = s + WBIT)) {
                        // CAS 修改成功获得锁,设置新的头结点 
                        whead = node;
                        node.prev = null;
                        return ns;
                    }
                } else if (LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
                    // 随机立减自旋次数  自旋次数为0时跳出自旋循环
                    break;
            }
        } else if (h != null) { // help release stale waiters
            // 头节点不为空
            // 进入情景:写锁被获取,队列中很多等待获取读锁的线程,写锁释放,读锁被唤醒后可能进入到这里
            WNode c; Thread w;
            while ((c = h.cowait) != null) { // 自旋
                // 头节点的 cowait不为空
                // h.cowait 修改成节点的下一个cowait
                if (U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) && (w = c.thread) != null)
                    // 唤醒 cowait 里面的线程
                    U.unpark(w);
            }
        }
        if (whead == h) { // 如果头结点没有变化
            // P 是之前的尾节点
            if ((np = node.prev) != p) { 
                // != 之前的尾节点,也就是说当前节点的前驱节点不是尾节点时
                if (np != null)
                    // 保存尾节点和当前节点的连接关系
                    (p = np).next = node;   // stale
            }
            else if ((ps = p.status) == 0)
                // ③ 上面第三次自旋break后会进入到这里,修改尾节点状态
                // 更新尾节点的状态为WAITING:-1, 然后继续回到第二次自旋的地方,重新开始自旋
                U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
            else if (ps == CANCELLED) {
                // p节点状态是取消,则删除p节点
                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)
                    // 设置了超时时间 已经超时,取消当前node节点
                    return cancelWaiter(node, node, false);
                Thread wt = Thread.currentThread();
                // 为给定地址设置值,忽略修饰限定符的访问限制,与此类似操作还有: putInt,putDouble,putLong,putChar等
                U.putObject(wt, PARKBLOCKER, this);
                // node节点指向当前线程
                node.thread = wt;
                // p.status < 0 的只有-1 也就是WAITING 
                if (p.status < 0 && (p != h || (state & ABITS) != 0L) && whead == h && node.prev == p)
                    // 阻塞当前线程
                    U.park(false, time);  // emulate LockSupport.park
                // 线程被唤醒后,清除节点的线程
                node.thread = null;
                U.putObject(wt, PARKBLOCKER, null);
                if (interruptible && Thread.interrupted()) // 要检查中断 && 线程有被中断
                    // 取消当前node节点
                    return cancelWaiter(node, node, true);
            }
        }
    }
}

获取写锁过程总结:

  1. 首先检查锁有没有被占用
    1. 没有被占用,尝试 CAS 修改state值,CAS 修改成功则获得锁,返回新的 state 值。
    2. CAS 修改失败的话进入下面自旋的逻辑
  2. 第一层自旋(节点入队):
    1. 先检查下锁有没有被占用((m = (s = state) & ABITS) == 0L), CAS 尝试一下获取锁,获取失败再继续自旋
    2. 入队之前会自旋64次(CPU核心数大于1),期间不做任何处理
    3. 初始化排队队列的队头队尾节点,当前节点加入到队尾,CAS 更新尾节点,更新成功则退出第一次自旋
  3. 开始第二层自旋(尝试获取锁,阻塞线程),第二层自旋和第三层自旋嵌套执行的:
    1. 如果头节点和之前的尾节点p还是是同一个(没有其他获取锁的节点排队已经入队), 说明马上应该轮到node节点获得锁(排队的只有node节点)。
      1. 初始化第三层自旋次数(第一次1024,第二次2048),开启第三层自旋
        1. 位运算检查锁是否有被释放((s = state) & ABITS) == 0L),CAS 修改 state 值,修改成功,退出,返回新的state值
      • 这里其实就是自旋和park线程之间性能的一个权衡,马上就要获得锁了,是自旋还是阻塞线程继续等,这里选择了先自旋1024次,如果没有获得锁,继续自旋2048次,如果还是没获得锁,则退出第三层自旋,回到第二层自旋,准备阻塞当前线程。
    2. 如果排队的头结点不为空,检查头结点的cowait 链表,如果不为空,自旋 CAS 修改头节点的cowait, 尝试唤醒整个链的节点线程
    3. 第三层自旋完成后还是没有获取到锁,阻塞当前线程,等待被唤醒,被唤醒后继续第二层自旋获取锁,重复这个过程,直到获取锁成功推出。

释放写锁:unlockWrite(stamp)

public void unlockWrite(long stamp) {
    WNode h;
    // state != stamp  检查解锁与加锁的版本是否匹配
    // (stamp & WBIT) == 0L 为true的话说明锁没有被占用
    if (state != stamp || (stamp & WBIT) == 0L)
        // 抛出异常
        throw new IllegalMonitorStateException();
    // 释放写锁,会增加state的版本
    // stamp += WBIT 等于二进制(第一次加写锁和解锁) 0001 1000 0000 + 0000 1000 0000 == 0010 0000 0000
    // 解锁会把stamp 的二进制第8位设置为0
    // 相当于重新赋值state值
    state = (stamp += WBIT) == 0L ? ORIGIN : stamp;
    if ((h = whead) != null && h.status != 0) // 头结点不为 && 状态不为初始状态0(一般是WAITING -1),说明队列中有排队获取锁的线程
         // 唤醒头节点的后继节点
         release(h);
}
private void release(WNode h) {
    if (h != null) {
        // q节点:头节点的有效后继节点
        // w: 需要唤醒的线程
        WNode q; Thread w;
        // 将头节点的状态设置成0
        U.compareAndSwapInt(h, WSTATUS, WAITING, 0);
        if ((q = h.next) == null || q.status == CANCELLED) { // 如果头节点的后继为空 或者 是取消状态
            // 就从排队的队尾找一个有效的节点
            for (WNode t = wtail; t != null && t != h; t = t.prev)
                if (t.status <= 0)
                    q = t;
        }
        // 找到了有效的节点,唤醒其线程
        if (q != null && (w = q.thread) != null)
            U.unpark(w);
    }
}

释放写锁过程总结:

  1. 检查锁印章戳是否匹配,锁是否有被占用,检查不通过抛出异常
  2. 通过位运算stamp += WBIT计算新的state值,state 二进制位的第8位会被设置成0就是写锁解锁
  3. 检测队列中是否有排队获取锁的线程
    1. 唤醒下一个等待获取锁的线程(unpark(thread)

获取普通读锁:readLock()

public long readLock() {
    long s = state, next;  // bypass acquireRead on common uncontended case
    // 在没有线程获得锁的情况下,s的初始值是256 
    // whead == wtail 为true:表示队列为空
    // (s & ABITS) < RFULL: 已获取读锁的数小于最大值126
    return ((whead == wtail && (s & ABITS) < RFULL &&  U.compareAndSwapLong(this, STATE, s, next = s + RUNIT)) 
					? next : acquireRead(false, 0L));
}

代码解析:

  1. 队列为空 && 已获取读锁次数小于126 && CAS 修改 state 值
    1. 条件完全成立,成功获得锁,返回新的state值
  2. 没有成功,进入到acquireRead(false, 0L)方法排队获取锁

acquireRead源码展示(代码超100行,需耐心观看):

private long acquireRead(boolean interruptible, long deadline) {
        // p节点为尾节点  node为入队节点
        WNode node = null, p;
        // 第一层大循环 第一次自旋,是不是和获取写锁的很像?
        for (int spins = -1;;) {
            WNode h;
            if ((h = whead) == (p = wtail)) { // 首尾节点相等,说明队列为空,有线程在排队不会进入if
                // 前面没获取到锁,队列又为空,是不是应该马上就是当前线程获取锁了?
                // 第二次自旋 自旋64次 目的是为了看马上能不能获取锁(排队队列为空,没线程排队时,会在这里自旋获取锁)
                for (long m, s, ns;;) {
                    // 这里是个三目运算,代码太长,拆开来看
                    // (m = (s = state) & ABITS) < RFULL;和进入readLock()方法时的条件一样,判断读锁的数是否达到最大值,只有写锁被获取,这里就是false
                    // 我们假设前面写锁被获取了,现在获取读锁,m 就是128,大于RFULL 126
                    // 没有达到最大值, CAS 修改状态值 U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT)
                    // 超过最大值了,记得前面那个readerOverflow属性不?在tryIncReaderOverflow这累加   
                    if ((m = (s = state) & ABITS) < RFULL ? U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
                        (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))
                        return ns;
                    else if (m >= WBIT) {// if条件成立,说明说明被占用
                        // 
                        if (spins > 0) {
                            if (LockSupport.nextSecondarySeed() >= 0)
                                --spins; // 随机减自旋次数
                        } else {
                            if (spins == 0) { // 自旋次数减到0了,还没获取到读锁
                                WNode nh = whead, np = wtail;
                                if ((nh == h && np == p) || (h = nh) != (p = np))
                                    break; // 退出自旋
                            } 
                            spins = SPINS; // 初始自旋次数 64次
                        }
                    }
                }
               // 上面到这里都是在处理队列为空,马上要获取到锁的情况
            } 
						if (p == null) { // 尾节点为空,初始化排队队列,有线程在排队时不会进入到这里
                WNode hd = new WNode(WMODE, null);
                if (U.compareAndSwapObject(this, WHEAD, null, hd)) // CAS 设置头节点
                    // 运行到这里后,会回到第一次的自旋,再次进入到第二次自旋,这次 spins 为0,只会自旋一次
                    wtail = hd;
            } else if (node == null)
                // 初始化当前入队排队节点,有线程在排队时,直接进入到这里,然后继续第一次自旋
                node = new WNode(RMODE, p); 
            else if (h == p || p.mode != RMODE) { // 队列为空 或者 尾节点不是读模式
                // 排队节点入队
                if (node.prev != p)
                    node.prev = p; // 设置排队节点的前驱节点
                    // 继续第一次自旋     
                else if (U.compareAndSwapObject(this, WTAIL, p, node)) {  // CAS 修改尾节点
                    p.next = node; // 老的尾节点的后继节点为当前节点
                    // 进入到这里会退出第一层自旋, 直接进入到下面第二层的大的自旋
                    break;
                }                
            } else if (!U.compareAndSwapObject(p, WCOWAIT, node.cowait = p.cowait, node))
                // 上面那个if分支进不去,只有条件:队列不为空 and  尾节点是读模式 为真
                // 进入到了这里,说明CAS失败
                // 这里的CAS 就是把当前节点加入到尾节点的cowait栈里面
                // 从这里可以看出加入的顺序是个栈结构,先把旧的尾节点的cowait赋值给node节点的cowait,然后再把node节点赋值给尾节点
                node.cowait = null;
            else { 
                // 进入到这,说明上面的if分支都没有进去,尾节点不为空,当前节点不为空,队列不为空,尾节点是读模式,上面CAS修改成功
                // 总结一下进入到这里的条件就是,有个线程获得了写锁还没释放,队列中有读线程在排队
                // 第三次自旋,有线程在排队获取锁时,会进入到这里自旋
                for (;;) {
                    WNode pp, c; Thread w;
                    if ((h = whead) != null && (c = h.cowait) != null &&
                        U.compareAndSwapObject(h, WCOWAIT, c, c.cowait) &&
                        (w = c.thread) != null) // help release
                        // 头节点不为空,且其cowait节点不为空,唤醒整个cowait栈的线程
                        U.unpark(w);
                    if (h == (pp = p.prev) || h == p || pp == null) {
                        // 头节点等于尾节点的前驱节点 或者头节点等于尾节点 或者 尾节点的前驱节点为空
                        // 说明还是马上轮到自己获得锁       
                        long m, s, ns;
                        do {
                            // 判断是否可以使用CAS获取读锁             
                            if ((m = (s = state) & ABITS) < RFULL ?  
                                U.compareAndSwapLong(this, STATE, s,  ns = s + RUNIT) :
                                (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L))
                                return ns;
                        } while (m < WBIT); // m < WBIT时表示写锁没有被占用,一直尝试获取锁
                    }
                    if (whead == h && p.prev == pp) { // 对头没有发生变化 ,队尾也没发生变化
                        long time;
                        if (pp == null || h == p || p.status > 0) { // 队尾的前驱节点为空或者 头节点等于尾节点 或者 老的尾节点被取消(>0的状态只有1,取消)
                            node = null; // 抛弃当前节点,退出当前循环,回到第一层的自旋,重新构建节点
                            break;
                        }
                        if (deadline == 0L)
                            time = 0L;
                        else if ((time = deadline - System.nanoTime()) <= 0L) // 超时
                            return cancelWaiter(node, p, false); // 取消节点
                        Thread wt = Thread.currentThread();
                        U.putObject(wt, PARKBLOCKER, this);
                        node.thread = wt;
                        if ((h != pp || (state & ABITS) == WBIT) &&
                            whead == h && p.prev == pp)
                            U.park(false, time); // 阻塞当前线程
                        // 线程被唤醒,开始继续自旋获取锁  
                        node.thread = null;
                        U.putObject(wt, PARKBLOCKER, null);
                        if (interruptible && Thread.interrupted())
                            return cancelWaiter(node, p, true);
                    }
                }
            }
        }
        // 第二层大的循环 
        for (int spins = -1;;) {
            WNode h, np, pp; int ps;
            if ((h = whead) == p) {  // 如果队列为空,说明马上轮到当前线程获得锁了
                // 这个大的if 里面做的就是获取锁
                if (spins < 0)
                    // 初始化本次自旋获取锁的次数:1024次
                    spins = HEAD_SPINS;
                else if (spins < MAX_HEAD_SPINS)
                    // 上面1024次自旋没有获取到锁,就自旋翻倍:2048次,继续下面的自旋
                    spins <<= 1;
                // 开始自旋获取锁
                for (int k = spins;;) { // spin at head
                    long m, s, ns;
                    // 这个if条件 检查了是否可以获取锁,如果可以就CAS获取锁
                    if ((m = (s = state) & ABITS) < RFULL ?
                        U.compareAndSwapLong(this, STATE, s, ns = s + RUNIT) :
                        (m < WBIT && (ns = tryIncReaderOverflow(s)) != 0L)) {
                        // 进入到这个if里面,说明就获得了锁
                        WNode c; Thread w;
                        // 这里的node节点是还没有绑定thread的
                        whead = node;
                        node.prev = null;
                        // 要唤醒当前node节点中的所有cowait节点线程
                        // 当前节点是在上面的第一层大的自旋入队的,其他获取读锁的节点都是挂在这个节点的cowait下的
                        while ((c = node.cowait) != null) {
                            if (U.compareAndSwapObject(node, WCOWAIT,
                                                       c, c.cowait) &&
                                (w = c.thread) != null)
                                U.unpark(w); // 唤醒线程
                        }
                        return ns;
                    }  else if (m >= WBIT &&  LockSupport.nextSecondarySeed() >= 0 && --k <= 0)
                        // 上面没有获取到锁,自旋减次数,直到为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)
                    // 检查尾节点的状态,为0则更新成-1,回到第二层大的循环开始处
                    U.compareAndSwapInt(p, WSTATUS, 0, WAITING);
                else if (ps == CANCELLED) { // p节点被取消,删除这个节点
                    if ((pp = p.prev) != null) {
                        node.prev = pp;
                        pp.next = node;
                    }
                } else { // 上面2048次自旋后还是没获取到锁,进入到最终阻塞线程的环节
                    long time;
                    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) == WBIT) && whead == h && node.prev == p)
                        U.park(false, time); // 阻塞线程
                    // 线程被唤醒了,继续执行 第二层大的自旋获取锁
                    node.thread = null;
                    U.putObject(wt, PARKBLOCKER, null);
                    if (interruptible && Thread.interrupted()) // 被唤醒了,发现要求中断线程 并且线程被中断了
                        return cancelWaiter(node, node, true); // 取消当前节点
                }
            }
        }
    }

获取普通读锁总结:

  • 位运算检查写锁是否被占用,读锁是否超限制
    1. 满足条件,直接CAS 修改state值,并返回新的state值
  • 开始第一层大的自旋,第一层大的自旋里面包含了两种情况不同的自旋:
    1. 第一种自旋情况:排队的队列为空,没有其他线程在排队等待锁时
      • 这种情况说明,锁虽然已经被占用,但是马上就应该是我得到锁了,所以我先在这儿自旋(64次)等等你释放锁,免得阻塞我自己,之后唤醒还需要成本
      • 自旋没有获取到锁,会退出第一层大的自旋,进入到第二层大的自旋
    2. 第二种自旋情况:写锁被占用,排队的队列不为空,队尾是读模式时
      • 这种情况,会不断尝试获取锁,阻塞线程,等待被唤醒,一直在这个自旋里面,直到获得锁,或者超时中断被取消,不会进入到第二层大的自旋
  • 第二层大的自旋
    • 这一层的自旋是对第一层里面第一种自旋情况(马上轮到我获得锁,但是前面持有锁的线程就是不释放)的补充,因为没有线程在排队,只要前面的线程释放了锁,马上就可以获得锁了,所以这一层还是在自旋获得锁,只不过自旋次数有增加
      • 首先会尝试自旋1024次获得锁,如果前面还没释放锁,再自旋2048次
    • 如果2048次之后还是没有等到前面的锁释放,就阻塞当前线程,等待被唤醒,直到获得锁,或者超时中断被取消

cowait栈分析

下面代码是main线程先获取写锁不释放,之后T0,T1,T2,T3线程先后去获取读锁,最后断点观察整个排队队列的情况

StampedLock sl = new StampedLock();
long stamp = sl.writeLock();
// 先让T0线程去排队到尾节点
TimeUnit.SECONDS.sleep(1);
new Thread(new Runnable(){
    @SneakyThrows
    @Override
    public void run(){
        long stamp = sl.readLock();
        System.out.println(stamp + "   x");
    }
},"T0").start();

// 之后T1线程来获取读
TimeUnit.SECONDS.sleep(3);
new Thread(new Runnable(){
    @SneakyThrows
    @Override
    public void run(){
        long stamp = sl.readLock();
        System.out.println(stamp + "   x");
    }
},"T1").start();
TimeUnit.SECONDS.sleep(3);
new Thread(new Runnable(){
    @SneakyThrows
    @Override
    public void run(){
        long stamp = sl.readLock();
        System.out.println(stamp + "   x");
    }
},"T2").start();
TimeUnit.SECONDS.sleep(3); // 在这里先断点,进入到这里后,再到源码位置去断点,就可以看到如下图的情况了
new Thread(new Runnable(){
    @SneakyThrows
    @Override
    public void run(){
        long stamp = sl.readLock();
        System.out.println(stamp + "   x");
    }
},"T3").start();

运行代码后,断点截图:
源码分析:升级版的读写锁 StampedLock

他们的关系可以用如下表示,横向是链表,纵向是cowait栈。
源码分析:升级版的读写锁 StampedLock

释放读锁:unlockWrite(stamp)

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)) { // CAS 更新state值
                if (m == RUNIT && (h = whead) != null && h.status != 0) // 头结点不为空
                    release(h); // 唤醒下一个节点
                break;
            }
        } else if (tryDecReaderOverflow(s) != 0L) 
            // 读锁个数饱和溢出,尝试减少readerOverflow
            break;
    }
}
private void release(WNode h) {
    if (h != null) {
        WNode q; Thread w;
        U.compareAndSwapInt(h, WSTATUS, WAITING, 0);
        if ((q = h.next) == null || q.status == CANCELLED) {
            for (WNode t = wtail; t != null && t != h; t = t.prev)
                if (t.status <= 0)
                    q = t;
        }
        if (q != null && (w = q.thread) != null)
            U.unpark(w);
    }
}

释放读锁的逻辑也比较简单,和释放写锁的逻辑很相识,唤醒下一个节点的release方法也完全一致

获取乐观读锁:tryOptimisticRead()

public long tryOptimisticRead() {
    long s;
    return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
}

乐观读锁的逻辑也比较简单,就一个三目运算,((s = state) & WBIT) == 0L 就是看写锁是否有被占用,写锁被占用返回0,否则返回写锁没被占用的包含高位版本有效戳(也就是写锁的版本)。

检测乐观读版本:validate(stamp)

public boolean validate(long stamp) {
    // 插入内存屏障,禁止load操作重排序。
    // 由于StampedLock提供的乐观读锁不阻塞写线程获取读锁,当线程共享变量从主内存load到线程工作内存时,会存在数据不一致问题
    // 解决锁状态校验运算发生重排序导致锁状态校验不准确的问题
	  U.loadFence(); 
	  return (stamp & SBITS) == (state & SBITS);
}

返回true:表示期间没有写锁发生,读锁为所谓

返回false:表示期间有写锁发生

那这里是怎么计算的呢?

SBITS为-128,用二进制表示是:1111 1111 1111 1000 0000

两种情况:

  1. 假如先获取乐观锁,再获取读锁;
    乐观锁返回的stamp为256,二进制位是 0001 0000 0000;
    获取读锁之后state值是257,二进制位是 0001 0000 0001;
    它们分别于与-128 进行与运算后都是0001 0000 0000,也就是十进制256,返回true;
  2. 假如先获取乐观锁,再获取写锁;
    乐观锁返回的stamp为256,二进制位是 0001 0000 0000;
    获取写锁之后state值是384,二进制位是 0001 1000 0000;
    它们分别与-128 进行与运算后,相当与 256 == 384,结果肯定返回false;

普通读锁转换成写锁:tryConvertToWriteLock(stamp)

public long tryConvertToWriteLock(long stamp) {
    // m标识最新的锁标识
    // a标识被转换的锁的锁标识
    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))
                // s:是之前的锁状态
                // s - RUNIT:就是释放读锁
                //  + WBIT :就是加写锁(进入之前写锁没被占用)
                return next; // 返回最新的锁状态
        } else
            break;  // 其他情况,全部返回0,转换失败
    }
    return 0L; // 返回0,标识转换写锁失败
}

普通读锁转换成写锁过程总结:

  1. 如果转换前是写锁,直接返回写锁
  2. 如果转换前是读锁,转换期间,写锁被占用,返回0,转换失败
  3. 如果转换前是读锁,写锁没有被占用,释放读锁,加写锁,返回写锁,转换成功
  4. 其他情况,全部返回0,转换失败

StampedLock 总结

  1. StampedLock 是一种支持乐观读锁的高级版读写锁
  2. StampedLock 没有使用AQS 同步框架,而是完全自己实现的同步状态state 和 CLH队列维护算法
  3. 同步状态state的低7位标识读锁的数量,第8位标识写锁是否被占用,高24位记录写锁的版本,每次释放写锁会版本位置会加1
  4. 写锁每次获取state会加128,释放也会加128,读锁是加减1
  5. StampedLock 的连续多个读锁线程,只有第一个是在队列上,后面的读线程都存在第一个线程的cowait栈结构上
  6. StampedLock 唤醒一个读锁线程后,读锁线程会唤醒所有在它cowait栈上的等待读锁线程
  7. StampedLock 用到了大量的自旋操作,适合持有锁时间比较短的任务,持有锁时间长的话等待的线程自旋后还是会阻塞自己。
  8. StampedLock 同一个线程先获取读锁,再获取写锁也会死锁
  9. StampedLock 写锁不支持重入,读锁支持重入
  10. StampedLock 不支持条件锁
  11. StampedLock 不支持公平锁,上来有条件就 CAS 尝试获得锁

StampedLock 与 ReentrantReadWriteLock的区别总结

使用功能上的区别:

  1. StampedLock 支持乐观读锁,RRWL 没有
  2. StampedLock 支持锁转换,tryConvertToXXXXX(stamp)
  3. StampedLock 写锁不支持重入,RRWL 支持重入
  4. StampedLock 不支持条件锁,RRWL 支持条件锁
  5. StampedLock 不支持公平锁,RRWL 支持公平锁

底层实现的区别:

  1. StampedLock 没有使用同步框架AQS,RRWL 是基于AQS 来实现排队、阻塞、唤醒等功能的
  2. StampedLock 获取锁时,会直接使用CAS尝试获得锁(不公平,不看排队),会根据CPU核心数来决定自旋次数等待获取锁
  3. StampedLock 的 CLH 队列中连续的读线程只有首个节点存储在队列中,后面的节点都存储的首个节点的cowait栈中,即 1→5→4→3→2→1 这种顺序。
  4. StampedLock 中同步状态 state 被分成了三部分,第8位记录的是写锁的状态,低7位记录读锁的次数,其他位记录的是写锁的版本
  5. RRWL 中同步状态 state 被分成两部分,高16位记录读锁次数,低16位记录写锁次数
  6. StampedLock 唤醒一个读锁线程后,读线程会唤醒所有在它cowait栈上的等待读锁线程

喜欢 (0)