J.U.C|读-写锁ReentrantReadWriteLock

摘要：所以就有了读写锁。只要没有，读取锁可以由多个线程同时保持。其读写锁为两个内部类都实现了接口。读写锁同样依赖自定义同步器来实现同步状态的，而读写状态就是其自定义同步器的状态。判断申请写锁数量是否超标超标则直接异常，反之则设置共享状态。

在上篇我们聊到了可重入锁（排它锁）ReentrantLcok ,具体参见《J.U.C|可重入锁ReentrantLock》

ReentrantLcok 属于排它锁，本章我们再来一起聊聊另一个我们工作中出镜率很高的读-写锁。

重入锁ReentrantLock是排他锁（互斥锁），排他锁在同一时刻仅有一个线程可访问，但是在大多数场景下，大部分时间都是提供读服务的，而写服务占用极少的时间，然而读服务不存在数据竞争的问题，如果一个线程在读时禁止其他线程读势必会降低性能。所以就有了读写锁。

读写锁内部维护着一对锁，一个读锁和一个写锁。通过分离读锁和写锁，使得并发性比一般排他锁有着显著的提升。

读写锁在同一时间可以允许多个读线程同时访问，但是写线程访问时，所有的读线程和写线程都会阻塞。

主要有以下特征：

公平性选择：支持非公平（默认）和公平的锁获取方式，吞吐量还是非公平优于公平。

重进入：该锁支持重进入，以读写线程为列，读线程在获取到读锁之后，能再次获取读锁。而写线程在获取写锁后能够再次获取写锁，同时也可以获取读锁。

锁降级：遵循获取写锁、读锁再释放写锁的次序，写锁能够降级成为读锁。

读写锁最多支持65535个递归写入锁和65535个递归读取锁。 锁降级：遵循获取写锁、获取读锁在释放写锁的次序，写锁能够降级成为读锁 读写锁ReentrantReadWriteLock实现接口ReadWriteLock，该接口维护了一对相关的锁，一个用于只读操作，另一个用于写入操作。只要没有 writer，读取锁可以由多个 reader 线程同时保持。

读写锁ReentrantReadWriteLock 实现了ReadWriteLock 接口，该接口维护一对相关的锁即读锁和写锁。

public interface ReadWriteLock {
    /**
     * Returns the lock used for reading.
     *
     * @return the lock used for reading
     */
    Lock readLock();

    /**
     * Returns the lock used for writing.
     *
     * @return the lock used for writing
     */
    Lock writeLock();
}

ReadWriteLock定义了两个方法。readLock()返回用于读操作的锁，writeLock()返回用于写操作的锁。ReentrantReadWriteLock定义如下：

/** 内部类 读锁*/
private final ReentrantReadWriteLock.ReadLock readerLock;
 /** 内部类 写锁*/
private final ReentrantReadWriteLock.WriteLock writerLock;
/** 执行所有同步机制 */
final Sync sync;

// 默认实现非公平锁
public ReentrantReadWriteLock() {
    this(false);
}
// 利用给定的公平策略初始化ReentrantReadWriteLock
public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
 }
 
 // 返回写锁
 public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
 //返回读锁
 public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }
 
 // 实现同步器，也是实现锁的核心
 abstract static class Sync extends AbstractQueuedSynchronizer {
    // 省略实现代码
 }
 
 // 公平锁的实现
 static final class FairSync extends Sync {
    // 省略实现代码
 }
 // 非公平锁的实现
 static final class NonfairSync extends Sync {
    // 省略实现代码
 }
 
 // 读锁实现
 public static class ReadLock implements Lock, java.io.Serializable {
    // 省略实现代码
 }
 // 写锁实现
 public static class WriteLock implements Lock, java.io.Serializable {
    // 省略实现代码
 }

ReentrantReadWriteLock 和 ReentrantLock 其实都一样，锁核心都是Sync， 读锁和写锁都是基于Sync来实现的。从这分析其实ReentrantReadWriteLock就是一个锁，只不过内部根据不同的场景设计了两个不同的实现方式。其读写锁为两个内部类： ReadLock、WriteLock 都实现了Lock 接口。

读写锁同样依赖自定义同步器来实现同步状态的， 而读写状态就是其自定义同步器的状态。回想ReentantLock 中自定义同步器的实现，同步状态表示锁被一个线程重复获取的次数，而读写锁中的自定义同步器需要在一个同步状态（一个整型变量）上维护多个读线程和写线程的状况，而该状态的设计成为关键。

如何在一个整型上维护多种状态，那就需要‘按位切割使用’这个变量，读写锁将变量切割成两部分，高16位表示读，低16位表示写。

分割之后，读写锁是如何迅速确定读锁和写锁的状态呢？通过位运算，假如当前同步状态为S，那么写状态等于 S & 0x0000FFFF（将高16位全部抹去），读状态等于S >>> 16(无符号补0右移16位)。代码如下：

static final int SHARED_SHIFT   = 16;
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;

/** Returns the number of shared holds represented in count  */
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
/** Returns the number of exclusive holds represented in count  */
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

写锁是一个支持重入的排他锁，如果当前线程已经获取了写锁，则增加写状态。如果当前线程获取写锁时读锁已经被获取或者该线程不是已经获取写锁的线程，则当前线程进入等待状态。

写锁的获取

写锁的获取入口通过WriteLock的lock方法

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

Sync的acquire(1)方法 定义在AQS中

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}

tryAcquire(arg) 方法除了重入方法外，还增加了是否存在读锁的判断，如果读锁存在、则不能获取写锁。原因在于写操作要对所有的读操作的可见性。

protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            // 获取同步状态
            int c = getState();
            // 获取写锁的获取次数
            int w = exclusiveCount(c);
             // 已有线程获取锁
            if (c != 0) {
                // (Note: if c != 0 and w == 0 then shared count != 0)
                /**
                 * w == 0 表示存在读锁（同步状态不等于0说明已有线程获取到锁（读/写）
                 * 而写锁状态为0则说明不存在写锁，所以只能是读锁了）
                 * current != getExclusiveOwnerThread()) 不是自己获取的写锁
                 * 
                 * 如果存在读锁或者持有写锁的线程不是自己，直接返回false
                 */
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                    
                 // 如果获取写锁的数量超过最大值65535 ，直接异常
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                // 设置共享状态
                setState(c + acquires);
                return true;
            }
             
             /**
             * writerShouldBlock() 是否需要阻塞写锁，这里直接返回的是false
             * compareAndSetState(c, c + acquires) 设置写锁的状态
             */
            
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;   
    }

小结

写锁的获取基本和RenntrantLock 类似
判断当前是否有线程持有写锁（写锁的状态是否为0）
写锁的状态不为0，如果存在读锁或者写锁不是自己持有则直接返回fasle。
判断申请写锁数量是否超标（> 65535）,超标则直接异常，反之则设置共享状态。
写锁状态为0，如果写锁需要阻塞或者CAS设置共享状态失败，则直接返回false，否则获取锁成功，设置持锁线程为自己。

来张图加深下理解

写锁的释放

写锁的释放和ReentrantLock 极为相似， 每次释放就是状态减1 ，当状态为0表示释放成功。

写锁释放的入口WriteLock中的unlock方法

public void unlock() {
      sync.release(1)
 }

Sync 中release方法由AQS中实现的

public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

tryRelease(arg) 方法释放共享状态，非常简单就是共享状态减1，为0表示释放成功

protected final boolean tryRelease(int releases) {
            // 判断锁持有者是否是自己
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
             // 共享状态值 - release
            int nextc = getState() - releases;
            // 判断写锁数量是否为0
            boolean free = exclusiveCount(nextc) == 0;
            if (free)
                setExclusiveOwnerThread(null);
            setState(nextc);
            return free;
        }

小结

写锁的释放很简单

首先判断锁持有者不是自己则直接异常

是自己则将共享状态 -1

判断写锁数量是否为0，如果为0将持有锁线程变量设为null

设置共享状态

来张图加深下理解

读锁为一个可重入的共享锁，它能够被多个线程同时持有，在没有其他写线程访问时，读锁总是获取成功，所需要的也就是（线程安全的）增加读状态。

读锁的获取

读锁的获取可以通过ReadLock.lock()方法。

public void lock() {
    //读锁是一个可重入共享锁，委托给内部类Sync实现
     sync.acquireShared(1);
}

Sync的acquireShared(1)方法定义在AQS中

public final void acquireShared(int arg) {
        // AQS 中 尝试获取共享状态，如果共享状态大于等于0则说明获取锁成功，否则加入同步队列。
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

tryAcquireShared(int unused)方法中，如果其他线程获取了写锁，则读锁获取失败线程将进入等待状态，如果当前线程获取写锁或者写锁未被获取则利用CAS(线程安全的)增加同步状态，成功则获取锁。

protected final int tryAcquireShared(int unused) {
            
            Thread current = Thread.currentThread();
            // 获取共享状态
            int c = getState();
            // 判断是否有写锁 && 持有写锁的线程是否是自己，为true直接返回-1 
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
             //获取共享资源的数量
            int r = sharedCount(c);
           
           /**
              * readerShouldBlock()：判断锁是否需要等待（公平锁原则）
              * r < MAX_COUNT：判断锁的数量是否超过最大值65535
              * compareAndSetState(c, c + SHARED_UNIT)： 设置共享状态（读锁状态）
              */
              
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                // r==0 :当前没有任何线程获取读锁
                if (r == 0) {
                    // 设置当前线程为第一个获取读锁的线程
                    firstReader = current;
                    // 计数设置为1
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
                    // 表示重入锁，在计数其上+1
                    firstReaderHoldCount++;
                } else {
                    
                    /**
                     *  HoldCounter 主要是一个类来记录线程获取锁的数量
                     *  cachedHoldCounter 缓存的是最后一个获取锁线程的HoldCounter对象
                     */
                     
                    HoldCounter rh = cachedHoldCounter;
                    // 如果缓存不存在，或者线程不是自己
                    if (rh == null || rh.tid != getThreadId(current))
                        // 从当前线程本地变量ThreadLocalHoldCounter 中获取HoldCounter 并赋值给 cachedHoldCounter， rh
                        cachedHoldCounter = rh = readHolds.get();
                     // 如果缓存的HoldCounter 是当前的线程的，且计数为0 
                    else if (rh.count == 0)
                        // 将rh 存到ThreadLocalHoldCounter 中，将计数+1 
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            /**
             * 进入fullTryAcquireShared(current) 条件
             * 1： readerShouldBlock()  = true  
             * 2: r < MAX_COUNT = false  读锁达到最大
             * 3: 设置共享状态失败
            return fullTryAcquireShared(current);
        }

NonfairSync 中的 readerShouldBlock() 方法判断当前申请读锁的线程是否需要阻塞

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

apparentlyFirstQueuedIsExclusive() 判断同步队列中老二节点是否是独占式（获取写锁请求）是返回ture 否则返回false

final boolean apparentlyFirstQueuedIsExclusive() {
        Node h, s;
        // 主要条件判断下一个节点是否是获取写锁线程在排队
        return (h = head) != null &&
            (s = h.next)  != null &&
            !s.isShared()         &&
            s.thread != null;
    }

自旋来获取读锁，个人感觉对tryAcquireShared(int unused) 方法获取读锁失败的一种补救，其实现逻辑基本相同。

final int fullTryAcquireShared(Thread current) {
            // 线程内部计数器
            HoldCounter rh = null;
            // 自旋
            for (;;) {
                // 获取共享状态
                int c = getState();
               
               /**
                 * exclusiveCount(c) ！=0：存在独占锁（写锁）
                 * getExclusiveOwnerThread() != current 判断是否是自己持有写锁
                 * 再次是写锁是否是自己
                 */
                 
                if (exclusiveCount(c) != 0) {
                    if (getExclusiveOwnerThread() != current)
                        return -1;
                   
                } else if (readerShouldBlock()) {//判断读锁是否需要阻塞
                    // 如果需要阻塞，表示除了当前线程持有写锁外，还有其他线程在等待获取写锁，故，即使申请读锁的线程已经持有写锁（写锁内部再次申请读锁，俗称锁降级）还是会失败，因为有其他线程也在申请写锁，此时，只能结束本次申请读锁的请求，转而去排队，否则，将造成死锁
                    if (firstReader == current) {
                        // assert firstReaderHoldCount > 0;
                    } else {
                        // 到这里其实就写锁的一个让步， 清楚HoldCounter 缓存
                        if (rh == null) {
                            rh = cachedHoldCounter;
                            if (rh == null || rh.tid != getThreadId(current)) {
                                rh = readHolds.get();
                                if (rh.count == 0)
                                    readHolds.remove();
                            }
                        }
                        if (rh.count == 0)
                            return -1;
                    }
                }
                // 下面逻辑和tryAcquireShared(int unused) 基本相同不再解释了
                if (sharedCount(c) == MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                if (compareAndSetState(c, c + SHARED_UNIT)) {
                    if (sharedCount(c) == 0) {
                        firstReader = current;
                        firstReaderHoldCount = 1;
                    } else if (firstReader == current) {
                        firstReaderHoldCount++;
                    } else {
                        if (rh == null)
                            rh = cachedHoldCounter;
                        if (rh == null || rh.tid != getThreadId(current))
                            rh = readHolds.get();
                        else if (rh.count == 0)
                            readHolds.set(rh);
                        rh.count++;
                        cachedHoldCounter = rh; // cache for release
                    }
                    return 1;
                }
            }
        }

小结

读锁的获取稍微有点复杂，整个过程如下

如果其他线程获取了写锁、则获取读锁失败。

如果当前线程获取到了写锁或者写锁未被获取则利用CAS（线程安全的）增加读锁状态

否则 fullTryAcquireShared(Thread current) 自旋方式再次来尝试获取。

读锁获取流程图如下

读锁的释放

读锁的释放通过ReadLock的unlock()方式释放的。

public void unlock() {
            sync.releaseShared(1);
        }

Sync的releaseShared(1)同样定义在AQS中

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

调用tryReleaseShared(int unused) 方法来释放共享状态。

protected final boolean tryReleaseShared(int unused) {
            Thread current = Thread.currentThread();
            //判断当前线程释放是第一个获取读锁的线程
            if (firstReader == current) {
                // assert firstReaderHoldCount > 0;
                // 判断获取锁的次数释放为1，如果为1说明没有重入情况，直接释放firstReader = null;否则将该线程持有锁的数量 -1
                if (firstReaderHoldCount == 1)
                    firstReader = null;
                else
                    firstReaderHoldCount--;
            } else {
                // 如果当前线程不是第一个获取读锁的线程。
                
                // 获取缓存中的HoldCounter
                HoldCounter rh = cachedHoldCounter;
                // 如果缓存中的HoldCounter 不属于当前线程则获取当前线程的HoldCounter。
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                int count = rh.count;
                if (count <= 1) {
                    // 如果线程持有锁的数量小于等1 直接删除HoldCounter
                    readHolds.remove();
                    if (count <= 0)
                        throw unmatchedUnlockException();
                }
                // 持有锁数量大于1 则执行 - 1操作
                --rh.count;
            }
            // 自旋释放同步状态
            for (;;) {
                int c = getState();
                int nextc = c - SHARED_UNIT;
                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;
            }
        }

小结

锁的释放比较简单，

首先看当前线程是否是第一个获取读锁的线程，如果是并且没有发生重入，则将首次获取读锁变量设为null， 如果发生重入，则将首次获取读锁计数器 -1

其次 查看缓存中计数器是否为空或者是否是当前线程，如果为空或者不是则获取当前线程的计数器，如果计数器个数小于等1， 从ThreadLocl 中删除计数器，并计数器值-1，如果小于等于0异常 。

最后自旋修改同步状态。

读锁释放流程图如下

通过上面的源码分析，我们来总结下：

在线程持有读锁的情况下，该线程不能取得写锁(为了保证写操作对后续所有的读操作保持可见性)。

在线程持有写锁的情况下，该线程可以继续获取读锁（获取读锁时如果发现写锁被占用，只有写锁没有被当前线程占用的情况才会获取失败）。

仔细想想，这个设计是合理的：因为当线程获取读锁的时候，可能有其他线程同时也在持有读锁，因此不能把获取读锁的线程“升级”为写锁；而对于获得写锁的线程，它一定独占了读写锁，因此可以继续让它获取读锁，当它同时获取了写锁和读锁后，还可以先释放写锁继续持有读锁，这样一个写锁就“降级”为了读

一个线程要想同时持有写锁和读锁，必须先获取写锁再获取读锁；写锁可以“降级”为读锁；读锁不能“升级”为写锁。

因技术水平有限，如有不对的地方，欢迎拍砖