再次认识ReentrantReadWriteLock读写锁

摘要：但是不管怎样，在一个线程已经获取锁后，在释放前再次获取锁是一个合理的需求，而且并不生硬。那么如果考虑重入，也很简单，在加锁时将的值累加即可，表示同一个线程重入此锁的次数，当归零，即表示释放完毕。

最近研究了一下juc包的源码。
在研究ReentrantReadWriteLock读写锁的时候，对于其中一些细节的思考和处理以及关于提升效率的设计感到折服，难以遏制想要分享这份心得的念头，因此在这里写一篇小文章作为记录。

本片文章建立在已经了解并发相关基础概念的基础上，可能不会涉及很多源码，以思路为主。
如果文章有什么纰漏或者错误，还请务必指正，预谢。

首先我们需要知道独占锁（RenentractLock）这种基础的锁，在juc中是如何实现的：
它基于java.util.concurrent.locks.AbstractQueuedSynchronizer（AQS）这个抽象类所搭建的同步框架，利用AQS中的一个volatile int类型变量的CAS操作来表示锁的占用情况，以及一个双向链表的数据结构来存储排队中的线程。

简单地说：一个线程如果想要获取锁，就需要尝试对AQS中的这个volatile int变量（下面简称state）执行类似comapre 0 and swap 1的操作，如果不成功就进入同步队列排队（自旋和重入之类的细节不展开说了）同时休眠自己（LockSupport.park），等待占有锁的线程释放锁时再唤醒它。

那么，如果不考虑重入，state就是一个简单的状态标识：0表示锁未被占用，1表示锁被占用，同步性由volatile和CAS来保证。

上面说的是独占锁，state可以不严谨地认为只有两个状态位。

但是如果是读写锁，那这个锁的基本逻辑应该是：读和读共享、读和写互斥、写和写互斥

如何实现锁的共享呢？
如果我们不再把state当做一个状态，而是当做一个计数器，那仿佛就可以说得通了：获取锁时compare n and swap n+1，释放锁时compare n and swap n-1，这样就可以让锁不被独占了。

因此，要实现读写锁，我们可能需要两个锁，一个共享锁（读锁），一个独占锁（写锁），而且这两个锁还需要协作，写锁需要知道读锁的占用情况，读锁需要知道写锁的占用情况。

设想中的简单流程大概如下：

设想中的流程很简单，然而存在一些问题：

关于读写互斥：

对于某个线程，当它先获得读锁，然后在执行代码的过程中，又想获得写锁，这种情况是否应该允许？

如果允许，会有什么问题？如果不允许，当真的存在这种需求时怎么办？

关于写写互斥是否也存在上面两条提到的情况和问题呢？

我们知道一般而言，读的操作数量要远远大于写的操作，那么很有可能读锁一旦被获取就长时间处于被占有的情况（因为新来的读操作只需要进去+1就好了，不需要等待state回到0，这样的话state可能永远不会有回到0的一天），这会导致极端情况下写锁根本没有机会上位，该如何解决这种情况？

对于上面用计数器来实现共享锁的假设，当任意一个线程想要释放锁（即使它并未获取锁，因为解锁的方法是开放的，任何获取锁对象的线程都可以执行lock.unlock()）时，如何判断它是否有权限执行compare n and swap n-1？

是否应该使用ThreadLocal来实现这种权限控制？

如果使用ThreadLocal来控制，如何保证性能和效率？

在开始研究ReentrantReadWriteLock之前，应当先了解两个概念：

重入性
一个很现实的问题是：我们时常需要在锁中加锁。这可能是由代码复用产生的需求，也可能业务的逻辑就是这样。
但是不管怎样，在一个线程已经获取锁后，在释放前再次获取锁是一个合理的需求，而且并不生硬。
上文在说独占锁时说到如果不考虑重入的情况，state会像boolean一样只有两个状态位。那么如果考虑重入，也很简单，在加锁时将state的值累加即可，表示同一个线程重入此锁的次数，当state归零，即表示释放完毕。

公平、非公平
这里的公平和非公平是指线程在获取锁时的机会是否公平。
我们知道AQS中有一个FIFO的线程排队队列，那么如果所有线程想要获取锁时都来排队，大家先来后到井然有序，这就是公平；而如果每个线程都不守秩序，选择插队，而且还插队成功了，那这就是不公平。
但是为什么需要不公平呢？
因为效率。
有两个因素会制约公平机制下的效率：

上下文切换带来的消耗

依赖同步队列造成的消耗

我们之所以会使用锁、使用并发，可能很大一部分原因是想要挖掘程序的效率，那么相应的，对于性能和效率的影响需要更加敏感。
简单地说，上述的两点由于公平带来的性能损耗很可能让你的并发失去高效的初衷。
当然这也是和场景密切关联的，比如说你非常需要避免ABA问题，那么公平模式很适合你。
具体的不再展开，可以参考这篇文章：深入剖析ReentrantLock公平锁与非公平锁源码实现

回到我们之前提的问题：

我们先考虑这种情况是否实际存在：假设我们有一个对象，它有两个实例变量a和b，我们需要在实现：if a = 1 then set b = 2，或者换个例子，如果有个用户名字叫张三就给他打钱。

这看上去仿佛是个CAS操作，然而它并不是，因为它涉及了两个变量，CAS操作并不支持这种针对多个变量的疑似CAS的操作。
为什么不支持呢？因为cpu不提供这种针对多个变量的CAS操作指令（至少x86不提供），代码层面的CAS只是对cpu指令的封装而已。
为什么cpu不支持呢？可以，但没必要鄙人也不是特别清楚（逃）。

总而言之这种情况是存在的，但是在并发情况下如果不加锁就会有问题：比如先判断得到这个用户确实名叫张三，正准备打钱，突然中途有人把他的名字改了，那再打这笔钱就有问题了，我们判断名字和打钱这两个行为中间应当是没有空隙的。
那么为了保证这个操作的正确性，我们或许可以在读之前加一个读锁，在我释放锁之前，其他人不得改变任何内容，这就是之前所说的读写互斥：读的期间不准写。
但是如果照着这个想法，就产生了自相矛盾的地方：都说了读期间不能写，那你自己怎么在写（打钱）呢？

如果我们顺着这个思路去尝试解释“自己读的期间还在写”的行为的正当性，我们也许可以设立一个规则：如果读锁是我自己持有，则我可以写。
然而这会出现其他的问题：因为读锁是共享的，你加了读锁，其他人仍然可以读，这是否会有问题呢？假如我们的打钱操作涉及更多的值的改变，只有这些值全部改变完毕，才能说此时的整体状态正确，否则在改变完毕之前，读到的东西都有可能是错的。
再去延伸这个思路似乎会变得非常艰难，也许会陷入耦合的地狱。

但是实际上我们不需要这样做，我们只需要反过来使用读写互斥的概念即可：因为写写互斥（写锁是独占锁），所以我们在执行这个先读后写的行为之前，加一个写锁，这同样能防止其他人来写，同时还可以阻止其他人来读，从而实现我们在单线程中读写并存的需求。

这就是ReentrantReadWriteLock中一个重要的概念：锁降级

对于另一个子问题：如果在已经获取写锁的期间还要再获取写锁的时候怎么办？
这种情况还是很常见的，多数是由于代码的复用导致，不过相应的处理也很简单：对写锁这个独占锁增加允许单线程重入的规则即可。

如果我们有两把锁，一把读锁，一把写锁，它们之间想要互通各自加锁的情况很简单——只要去get对方的state就行了。
但是只知道state是不够的，对于读的操作来说，它如果只看到写锁没被占用，也不管有多少个写操作还在排队，就去在读锁上+1，那很可能发展成为问题所说的场景：写操作永远没机会上位。

那么我们理想的情况应该是：读操作如果发现写锁空闲，最好再看看写操作的排队情况如何，酌情考虑放弃这一次竞争，让写操作有机会上位。
这也是我理解的，为什么ReentrantReadWriteLock不设计成两个互相沟通的、独立的锁，而是公用一个锁（class Sync extends AbstractQueuedSynchronizer）——因为它们看似独立，实际上对于耦合的需求很大，它们不仅需要沟通锁的情况，还要沟通队列的情况。

公用一个锁的具体实现是：使用int state的高16位表示读锁的state，低16位表示写锁的state，而队列公用的方式是给每个节点增加一个标记，表明该节点是一个共享锁的节点（读操作）还是一个独占锁的节点（写操作）。

上面说到的“酌情放弃这一次竞争”，ReentrantReadWriteLock中体现在boolean readerShouldBlock()这个方法里，这个方法有两个模式：公平和非公平，我们来稍微看一点源码
先看公平模式的实现：

final boolean readerShouldBlock() {
    return hasQueuedPredecessors();
}

当线程发现自己可以获取读锁时（写锁未被占用），会调用这个方法，来判断自己是否应该放弃此次获取。
hasQueuedPredecessors()这个方法我们不去看源码，因为它的意思很显而易见（实际代码也是）：是否存在排队中的线程（Predecessor先驱者可以理解为先来的）。
如果有，那就放弃竞争去排队。
在公平模式下，无论读写操作，只需要大家都遵守FIFO的秩序，就不会出现问题描述的情况

再来看看非公平模式下的实现代码：

final boolean readerShouldBlock() {  
    return apparentlyFirstQueuedIsExclusive();
}

final boolean apparentlyFirstQueuedIsExclusive() {
    // Node表示同步队列中的一个节点
    Node h, s;
    // head是当前队列的头节点的一个公共引用，它是一个没有实际意义的节点，null or not只能标识队列是否初始化过
    // next是当前节点的下一个节点的引用
    // isShared()方法表明这个节点是一个共享锁（读锁）的节点还是独占锁（写锁）的节点
    return (h = head) != null &&
        (s = h.next)  != null &&
        !s.isShared()         &&
        s.thread != null;
}

总结一下语义：如果队列不为空，且队列最前面的节点是个独占锁的节点，则放弃竞争。也就是我们上面说的“根据队列情况酌情放弃”。

一般而言的读操作的线程对于state的操作可能只是+1然后-1，而如果发生重入，那就会是n次+1然后n次-1。
但是不管怎样，每一个线程都应当有一份记录自己持有共享锁数量的信息，这样释放锁的时候才能知道自己可不可以去-1。
这也许很简单，我们可以在锁里增加一个Map对象，用类似tid(k)-count(v)的数据结构来记录每个线程的持有数量；也可以为每个线程创建一个ThreadLocal，让它们自己拿着。

现在我们面前有两条路比较直观：将所有线程的小计数器维护在一个Map中，或是每个线程在ThreadLocal中维护自己的小计数器。
就这两条途径而言，应该是Map的这一条路比较高效，因为如果选择ThreadLocal也许会频繁进行其内部的ThreadLocalMap对象的创建和销毁，这很消耗资源。

然而事实是，ReentranctReadWriteLock选择的实现方式是后者，即使用ThreadLocal来实现，但是为什么选择这种方式正是我十分好奇的地方，因为根据经验，一定是利用Map统一管理小计数器的方式较为高效，且单个线程针对单个key的value进行+1或者-1的操作应该是满足as-if-serial原则的，也不存在安全问题。
因此针对两种不同的实现方式进行了一些测试：四线程并行情况下一千万次加解锁时间测试

Map统一管理实现

public static void main(String[] args) {
    long total = 0;
    for (int i = 0; i < 30; i++) {
        total += execute();
    }
    System.out.println(total / 30);
}

private static long execute() {

    var map = new HashMap();
    var readerPool = new ThreadPoolExecutor(
            4,
            4,
            5L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("readerPool"));

    var countDown = new CountDownLatch(10000000);
    for (int i = 0; i < 10000000; i++) {
        readerPool.execute(() -> {
            try {
                countDown.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
                return;
            }
            mapImplement(map);
        });
        countDown.countDown();
    }

    long startTime = System.currentTimeMillis();
    while (readerPool.getCompletedTaskCount() < 10000000) {
        LockSupport.parkNanos(100);
    }
    long total = System.currentTimeMillis() - startTime;

    System.out.println(readerPool.getCompletedTaskCount() + ", time: " + total);
    return total;
}

private static void mapImplement(HashMap map) {
    // lock
    var tid = Thread.currentThread().getId();
    Integer count;
    if ((count = map.get(tid)) != null) {
        map.put(tid, count + 1);
    } else {
        map.put(tid, 1);
    }

    // unlock
    int afterDecrement = -999;
    if ((count = map.get(tid)) == null ||
            (afterDecrement = (count - 1)) < 0) {
        System.out.println("error, count: " + count + ", afterDecrement: " + afterDecrement);
        return;
    }
    map.put(tid, afterDecrement);
}

三十次测试过程的平均执行时间为：2378毫秒，个人认为这个结果还是比较乐观的。

ThreadLocal各自持有实现

// 小计数器实体
static final class HoldCounter {
    int count;
}

// threadLocal
static final class ThreadLocalHoldCounter extends ThreadLocal {
    @Override
    public HoldCounter initialValue() {
        return new HoldCounter();
    }
}

public static void main(String[] args) {
    long total = 0;
    for (int i = 0; i < 30; i++) {
        total += execute();
    }
    System.out.println(total / 30);
}

private static long execute() {
    
    var readHolds = new ThreadLocalHoldCounter();
    var readerPool = new ThreadPoolExecutor(
            4,
            4,
            5L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new CustomizableThreadFactory("readerPool"));

    var countDown = new CountDownLatch(10000000);
    for (int i = 0; i < 10000000; i++) {
        readerPool.execute(() -> {
            try {
                countDown.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
                return;
            }
            threadLocalImplement(readHolds);
        });
        countDown.countDown();
    }

    long startTime = System.currentTimeMillis();
    while (readerPool.getCompletedTaskCount() < 10000000) {
        LockSupport.parkNanos(100);
    }
    long total = System.currentTimeMillis() - startTime;

    System.out.println(readerPool.getCompletedTaskCount() + ", time: " + total);
    return total;
}

private static void threadLocalImplement(ThreadLocalHoldCounter readHolds) {
    // lock
    var hc = readHolds.get();
    ++hc.count;

    // unlock
    hc = readHolds.get();
    --hc.count;
    if (hc.count == 0) {
        readHolds.remove();
    }
}

三十次测试过程的平均执行时间为：3079毫秒

可以看到，使用Map集中管理小计数器的实现方式的执行效率要比ThreadLocal的实现方式快20%以上。

难道我作为一个Java萌新都能想到的性能差距，Doug Lea这样的大神会想不到吗？

当然不会

实际上，ReentranctReadWriteLock在针对小计数器的具体实现上，增加了类似两层缓存的设计，大概如下：

// ThreadLocal对象本体
private transient ThreadLocalHoldCounter readHolds;

// 二级缓存：最近一次获取锁的线程所持有的小计数器对象的引用
private transient HoldCounter cachedHoldCounter;

// 一级缓存：首次获取锁的线程的线程对象引用以及它的计数
private transient Thread firstReader;
private transient int firstReaderHoldCount;

当线程尝试获取锁时，会执行如下的流程：

判断当前共享锁总计数器是否为0（当前锁处于空闲状态） 或 firstReader == Thread.currentThread()
是： 则直接在firstReaderHoldCount上进行+1（以及执行firstReader = Thread.currentThread()）
否： 前往2

判断cachedHoldCounter.tid == Thread.currentThread().getId()
是： 则直接在cachedHoldCounter.count上进行+1
否： 前往3

执行readHolds.get()进行获取或初始化，然后再对小计数器进行操作

当线程释放锁时，执行流程也大致相似，都是先对两级缓存进行尝试，逼不得已再去对ThreadLocal进行操作。
由于读操作的实际执行内容一般相当简单（类似return a），所以在绝大多数情况下，线程的加解锁行为都会命中一级缓存。

我尝试在ReentranctReadWriteLock的加解锁行为内埋了几个计数点来测试两级缓存的命中率，四线程并行1000万次加解锁操作，结果是：

一级缓存命中率大概为90~95%

二级缓存命中率大概为5~10%

ThreadLocal本体命中率大概为1~5%

而执行效率，进行1000万次加解锁，循环三十次得到的平均执行时间是：2027毫秒。
比上面提到的使用Map实现的方式更要快了15%左右。

虽说上面的小测试的编码也好，测试环境也好，都不算特别严谨，但是还是能非常直观地说明问题的吧。

如图，ReentranctReadWriteLock中有五个内部类：

Sync
Sync继承自AbstractQueuedSynchronizer，上文提到的volatile int state以及同步队列的实际实现都是由AbstractQueuedSynchronizer这个抽象类提供的，它还提供了一些在锁的性质不同时实现也会不同的可重写方法，Sync需要做的事情就是将这些通用的方法和规则加以实现和扩充，形成自己想要实现的锁。
Sync也是我们上文提到的，同时实现了读写两种性质的锁的根本。
另外，上文提到的关于分段使用state、利用公平性避免机会不均衡的问题、分级缓存共享锁小计数器等特性，均在此类中实现，需要特别关注。

NonfairSync与FairSync
这两个类都继承自Sync，它们提供了由Sync定义的两个用于进行公平性判断的方法：boolean writerShouldBlock()与boolean readerShouldBlock()。实际使用ReentranctReadWriteLock时，我们会通过构造方法选择需要构造公平还是非公平的锁，相应的会通过这两个子类构造实际的Sync类的对象，从而影响到加解锁过程中的一些判断。

ReadLock与WriteLock
这两个类都会持有上面提到的Sync类的对象的引用，并向用户（使用者）提供包装好但实现不同的操作，比如：

读锁获取

public void lock() {
    sync.acquireShared(1);
}

写锁获取

public void lock() {
    sync.acquire(1);
}

更多的源码就不再赘述了，搜索一下就会有非常多的文章解读源码并非作者懒得贴了。

juc这套并发框架的设计者和创始人Doug Lea，可以说是java开发者金字塔顶端的巨佬之一了，他所编写的juc包的代码，无论是代码结构的合理性、各种设计模式的使用、代码的优雅程度都令人叹为观止。看完源码觉得整个人都升华了

笔者学习了源码之后，觉得在面对这些充满了考虑的设计细节时产生的思考，才是真正可以使人得到长远的提升的东西。

因此整理出来，作为心得体会的记录。如果可以对其他小伙伴带来启发，那就更好了。
最后，如果文章内有什么纰漏或是错误，还请务必指正，再次感谢。