java读写锁—ReentrantReadWriteLock

前言

 

ReentrantReadWriteLock从字面上直接翻译过来应该是“重入读写锁”，首先需要指明的是读写锁本质上是“一个锁”的不同视图。ReentrantReadWriteLock是基于AQS实现的（对AQS的理解可以点击这里，文中提到在jdk1.8中基于AQS直接实现的API有CountDownLatch、Semaphore、ReentrantLock、ThreadPoolExecutor

、ReentrantReadWriteLock，前面4个已经总结过），这里提到的“一个锁”是指：不论是读线程、还是写线程都会同“一个锁”阻塞，并在同一个AQS队列中排队。

 

ReentrantReadWriteLock主要是利用读写分离的思想，读取数据使用“读锁” 实现多个读线程可以并行读；写数据使用“写锁” 实现只能由一个线程写入。相对于ReentrantLock，在读多写少的情况下，使用ReentrantReadWriteLock会有更好的性能表现。在《java并发编程实战》中，jdk的大神们用ReentrantReadWriteLock实现了读写线程安全的Map，这里以此为模板实现一个读写线程安全的list：

 

/**
 * Created by gantianxing on 2018/1/6.
 */
public class ReadWriteList<E> {
    private final List<E> list;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    private Lock r = lock.readLock();
    private Lock w = lock.writeLock();
 
    public ReadWriteList(List<E> list){
        this.list = list;
    }
 
    //获取数据，用读锁
    public E get(int index){
        r.lock();
        try{
            return list.get(index);
        }finally {
            r.unlock();
        }
    }
 
    //写入数据，用写锁
    public void add(E e){
        w.lock();
        try{
            list.add(e);
        }finally {
            w.unlock();
        }
    }
}

 

 

该工具类，可以将任意类型的List转换成“读写线程安全”的ReadWriteList，当然有人说我们可以直接用CopyOnWriteArrayList。但如果如果程序中已经大量使用的ArrayList、LinkedList，这时为了尽量少的改动原有代码就可以使用自己定义的 ReadWriteList，在读多写少的场景中比直接使用synchronized或者ReentrantLock性能会好很多。当然你也可以实现一个“读写线程安全”的Map，把这两个工具类放到你的代码中。

 

通过对自己实现ReadWriteList的讲解，相信大家对读写锁ReentrantReadWriteLock的用法已经有了基本的了解。下面来看ReentrantReadWriteLock的具体实现原理。

 

ReentrantReadWriteLock实现原理

 

构造方法

文章开头就提到ReentrantReadWriteLock中的读写锁，本质上是一个锁，那具体是怎么实现的呢？首先从它的两个构造方法就可以看出：

public ReentrantReadWriteLock(boolean fair) {
        //根据条件创建公平、或非公锁
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);//创建读锁
        writerLock = new WriteLock(this);//创建写锁
}
 
//默认依然是非公平锁
public ReentrantReadWriteLock() {
        this(false);
}
 

 

其中不论FairSync和NonfairSync是都对AQS的实现，这里构造方法会根据条件创建“一个”公平或者非公平锁，接着以此锁为参数同时创建 ReadLock（读锁）和WriteLock（写锁）两个视图，也就是说本质上多个排队线程会被同一个锁阻塞在同一个AQS队列上。接着来看是如何对AQS实现的。

 

对AQS的实现

与ReentrantLock的实现一样，ReentrantReadWriteLock也有公平锁和非公平锁的实现，其内部定义了三个内部类：Sync、FairSync、NonfairSync。其中Sync是抽象类 并且直接基础自AQS，FairSync和NonfairSync都继承自Sync分别实现“公平锁”和“非公平锁”。

 

我们都知道AQS的state字段只有1个值，如何表述两种锁的状态呢？ReentrantReadWriteLock的具体做法是用一个32位的二进制标识状态state,其中高16位表示持有“读锁”的线程数（多线程共享），低16位表示持有“写锁”线程重入次数（只有一个线程），在内部抽象类Sync中定义了如下几个字段：

        

        /*  //这段注释就是说明高16位，和低16位
         * Read vs write count extraction constants and functions.
         * Lock state is logically divided into two unsigned shorts:
         * The lower one representing the exclusive (writer) lock hold count,
         * and the upper the shared (reader) hold count.
         */
        //以16位为界分割
        static final int SHARED_SHIFT   = 16;
        //相当于对高16位加1，也就是持有读锁的线程数加1
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        //写锁最大重入次数，或者持有读锁最大线程数
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
       
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
 
        /** 返回持有读锁的线程数*/
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        /** 返回写锁重入次数  */
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

 

其中sharedCount方法取高16位，返回持有“读锁”的线程数；

exclusiveCount方法取低16位，返回“写锁”重入次数。到这里就解释了，为什么说读锁和写锁其实是同一个锁。

 

通过前几次对CountDownLatch、Semaphore、ReentrantLock的总结，相信大家对如何使用AQS实现锁都有了一定的认识：要实现排它锁，就重写AQS的tryAcquire和tryRelease方法，比如ReentrantLock的实现；要实现共享锁，就重写AQS的tryAcquireShared和tryReleaseShared方法，比如CountDownLatch和Semaphore的实现。有了这点基础的认识，我们再来看ReentrantReadWriteLock，本质上我们可以把ReentrantReadWriteLock的“读锁”理解为排它锁，ReentrantReadWriteLock的“写锁”理解为共享锁。

 

ReentrantReadWriteLock中对AQS的实现也确实如此，内部抽象类Sync同时实现了tryAcquire、tryRelease、tryAcquireShared、tryReleaseShared 4个方法（注意文中提到的排它锁就是“写锁”，共享锁就是“读锁”，这种称呼仅在ReentrantReadWriteLock的实现中有效）：

 

排它锁（写锁）实现

tryAcquire（排它获取方法）

protected final boolean tryAcquire(int acquires) {
        Thread current = Thread.currentThread();
        int c = getState();//获取当前AQS状态（一共32位）
        int w = exclusiveCount(c);//返回排它锁重入次数，也就是低16位
        if (c != 0) {//c不为0，表示有可能是重入独占锁（对应写锁），也有可能是共享锁（对应读锁）
            // (Note: if c != 0 and w == 0 then shared count != 0)
            //如果重入独占锁状态不为0，并且持有锁的线程不是当前线程，说明独占锁已经被其他线程占用
            if (w == 0 || current != getExclusiveOwnerThread())
                return false;
 
            //重入锁次数是否超过上限
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // 重入获取锁
            setState(c + acquires);
            return true;
        }
        //如果前AQS状态为0，表示可以获取锁
        if (writerShouldBlock() || //writerShouldBlock仅仅用于表示公平和非公平
                !compareAndSetState(c, c + acquires))
            return false;
        setExclusiveOwnerThread(current);//设置排它锁的持有者为当前线程
        return true;
    }

 

结合注释很好理解，始终只有一个线程能获取到排它锁（写锁），本质上就是对AQS的state字段的低16位+1。WriteLock写锁加锁的完整过程：WriteLock的lock方法-->AQS的acquire--> Sync的tryAcquire获取锁方法-->获取失败调用AQS的加入队列方法acquireQueued-->失败后阻塞 调用AQS的selfInterrupt()方法。

 

tryRelease（排它释放方法）

    

protected final boolean tryRelease(int releases) {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        int nextc = getState() - releases;//释放锁，
        boolean free = exclusiveCount(nextc) == 0;
        if (free) //因为有重入锁的情况，只有该线程全部释放，才能释放锁
            setExclusiveOwnerThread(null);
        setState(nextc);
        return free;
    }

 

排它锁的释放过程本质上是对AQS的state字段的低16位-1，直到为0时释放锁。WriteLock写锁释放的完整过程：WriteLock的unlock方法-->AQS的release方法--> Sync的tryRelease释放锁方法-->释放成功后调用AQS的unparkSuccessor方法，唤醒下一个线程获取锁。

 

共享锁（读锁）实现

tryAcquireShared（共享获取方法）
protected final int tryAcquireShared(int unused) {
           
        Thread current = Thread.currentThread();
        int c = getState();
        //如果其他线程持有写锁，获取读锁失败；
        //反正如果是当前线程持有写锁，可以再次获得读锁
        if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
            return -1;
        int r = sharedCount(c);//正在共享“读锁”的线程数
        if (!readerShouldBlock() && //公平锁和非公平锁实现的差别
                r < MAX_COUNT && //如果正在共享“读锁”的线程数超过最大值，获取锁失败
                compareAndSetState(c, c + SHARED_UNIT)) {//相当于对高16位+1
            if (r == 0) {
                firstReader = current; //设置第一个获取读锁线程
                firstReaderHoldCount = 1;
            } else if (firstReader == current) {
                firstReaderHoldCount++;//设置第一个读锁线程获取次数（重入）
            } else {//设置其他读锁线程获取次数（重入）
                HoldCounter rh = cachedHoldCounter;
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);//把重入次数放到每个读线程自己的ThreadLocal中
                rh.count++;
            }
            return 1;//获取成功
        }
        //完整版获取共享锁，主要用于compareAndSetState这步cas处理失败后，进行轮询获取
        return fullTryAcquireShared(current);
    }

 

获取共享锁的过程本质上是对AQS的state字段的高16位+1,如果在有排它锁的情况下（并且不是当前线程持有），获取共享锁的线程直接阻塞。如果当前线程持有排它锁，该线程还可以继续获取共享锁，这就是所谓的降级。但不能升级，也就是持有共享锁的线程，不一定能获取到排它锁。

 

ReadLock读锁的加锁过程：ReadLock的lock方法-->AQS的acquireShared方法-->Sync的tryAcquireShared获取锁方法-->如果获取失败调用AQS的doAcquireShared阻塞当前线程。

 

tryReleaseShared（共享释放方法）

 

//释放共享锁（读锁）
    protected final boolean tryReleaseShared(int unused) {
        Thread current = Thread.currentThread();
 
        //shep1：减少线程持有重入读锁的次数
        if (firstReader == current) {
            // assert firstReaderHoldCount > 0;
            if (firstReaderHoldCount == 1)
                firstReader = null;//该线程释放完所有的读锁
            else
                firstReaderHoldCount--;
        } else {
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();
            int count = rh.count;
            if (count <= 1) {//该线程释放完所有的读锁，
                readHolds.remove();//从自己的ThreadLocal中去掉
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --rh.count;
        }
 
        //step2：CAS设置AQS的state字段的高16位（读锁线程数）
        for (;;) {
            int c = getState();
            int nextc = c - SHARED_UNIT; //相当于对高16位-1
            if (compareAndSetState(c, nextc))
                // Releasing the read lock has no effect on readers,
                // but it may allow waiting writers to proceed if
                // both read and write locks are now free.
                return nextc == 0;
        }
    }

 

实现比较复杂，结合注释也不难理解，其本质就是对AQS的state字段的高16位-1。

 

ReadLock读锁的加锁过程：ReadLock的unlock方法-->AQS的releaseShared方法-->Sync的tryReleaseShared释放锁方法-->AQS的doReleaseShared方法。

 

共享锁（读锁）的分段转播性

 

在讲解读锁、写锁具体实现的时候，也许都注意到了：ReentrantReadWriteLock 实现了 “读-读”不冲突，也就是多读不阻塞；但“读-写”和“写-写”冲突。但这只是加锁部分，在释放锁后我们都知道需要唤醒后续线程，再次获取锁。如果AQS队列中的线程都是单纯的获取读锁或者写锁都很好实现：如果是写锁 只用唤醒队列中的头结点；如果是读锁 逐个唤醒所有的线程，这就是共享锁（读锁）的转播性。

 

但在ReentrantReadWriteLock的阻塞队列中同时包含了获取读锁以及写锁的线程，又是怎么实现的呢？AQS队列是一个双向链表，我们先来模拟这个队列：

 

头节点表示当前获取到锁的节点，这里线程1获取到“写锁”，后面的线程都会被阻塞；

当线程1释放“写锁”，线程2会获取到“读锁”，同时由于共享锁的传播性，线程3、线程4都会获取到“读锁”同时被唤醒。

 

那么问题来了，线程6是否会被唤醒呢？答案是否定的，共享锁的转播性是遇到“写锁”时会中止，具体实现代码见AQS中的doAcquireShared 方法，会遍历setHeadAndPropagate方法：

private void setHeadAndPropagate(Node node, int propagate) {
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            //当前节点为空，或者是共享节点时，继续传播。否则停止传播
            if (s == null || s.isShared())
                doReleaseShared();//唤醒线程
        }
    }

 

也许有人会问这里没有for循环怎么实现传播唤醒的呀，别忘了被阻塞的线程是在doAcquireShared方法中阻塞的，另外这里存在一个head头结点的移动，被唤醒后继续执行doAcquireShared方法会继续去头结点线程进行唤醒，直到s == null || s.isShared()这句为false。

 

这就是共享锁（读锁）的分段转播性，被写锁分开。

 

到这里ReentrantReadWriteLock读写锁的核心实现总结完毕，还有一些其他中断锁、延时锁、轮询锁的实现与ReentrantLock相同，不再累述。

 

总结

 

 

ReentrantReadWriteLock的实现相对ReentrantLock来说复杂了很多，加锁过程的核心实现原理：就是使用32二进制表示AQS的state（可以理解为 锁的状态），高16位表示持有读锁的线程数，低16位表示持有写锁线程的重入数。写锁的唤醒过程相对比较简单，读锁的唤醒过程，具备分段传播特性。

 

心灵鸡汤

基础：有同事问我怎么成为一名架构师，我会对他说做好程序员该做的事，很多人觉得我是一名架构师，其实我只是一名执着的程序员。

 

 

摘自--《天星老师语录》