ReentrantReadWriteLock源码解析

超、凢脫俗 2023-02-26 02:05 11阅读 0赞

### 什么是读写锁 ###

读写锁是一对(`ReentrantReadWriteLock`下的`writeLock`和`readLock`), `写锁`是`独占锁`，`读锁`是`公共锁`，读读共存，只要有写就互斥。

### 类的继承体系 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h6YXVfaXRkb2c_size_16_color_FFFFFF_t_70]

ReentrantReadWriteLock中的三个主要内部类：

（1）ReentrantReadWriteLock本身实现了ReadWriteLock接口，这个接口只提供了两个方法readLock()和writeLock（）；

（2）同步器，包含一个继承了AQS的Sync内部类，以及其两个子类FairSync和NonfairSync；

（3）ReadLock和WriteLock两个内部类实现了Lock接口，它们具有锁的一些特性。

// 读锁
    private final ReentrantReadWriteLock.ReadLock readerLock;
    // 写锁
    private final ReentrantReadWriteLock.WriteLock writerLock;
    // 同步器
    final Sync sync;

### 构造方法 ###

public ReentrantReadWriteLock() { 
            this(false);
    }
    // 是否使用公平锁的构造方法
    public ReentrantReadWriteLock(boolean fair) { 
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }

###  ###

### 读锁 ###

**lock()**

//java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock#lock
    public void lock() { 
        sync.acquireShared(1);
    }
    //java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireShared
    public final void acquireShared(int arg) { 
        // 尝试获取共享锁（返回1表示成功，返回-1表示失败）
        if (tryAcquireShared(arg) < 0)
             // 失败了就可能要排队等待获取
            doAcquireShared(arg);
    }
    
    protected final int tryAcquireShared(int unused) { 
        //当前线程
        Thread current = Thread.currentThread();
        //获取读写状态的int标量
        //int值按位切割，高16位表示读，低16位表示写
        int c = getState();
        //w=c&0x0000FFFF(将高16位全部抹去)
        //exclusiveCount获取写锁的获取次数，
        //不等于0代表写锁被获取，直接获取读锁失败，返回-1
        if (exclusiveCount(c) != 0 &&
            getExclusiveOwnerThread() != current)
            return -1;
        //到这写锁获取次数w=0,没有获取写锁
        //读锁r=c>>>16(无符号右移16位)
        int r = sharedCount(c);
        //读锁是否需要阻塞排队，读锁c+1<<<16，相当于读锁的标志位r+1
        if (!readerShouldBlock() &&
            r < MAX_COUNT &&
            compareAndSetState(c, c + SHARED_UNIT)) { 
            // 获取读锁成功
            //第一次获取读锁
            if (r == 0) { 
                //当前线程为第一个读锁拥有者
                firstReader = current;
                //记录，第一个线程获取读锁的次数为1
                firstReaderHoldCount = 1;
                //第一个读锁获取者再次获取锁，
            } else if (firstReader == current) { 
                //获取锁次数+1
                firstReaderHoldCount++;
            } else { 
                //线程不是第一个获取读锁
                // 则从缓存中获取重入次数保存器
                HoldCounter rh = cachedHoldCounter;
                 // 如果缓存不属性当前线程
                // 再从ThreadLocal中获取
                // readHolds本身是一个ThreadLocal，里面存储的是HoldCounter
                if (rh == null || rh.tid != getThreadId(current))
                    // get()的时候会初始化rh
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)
                     // 如果rh的次数为0，把它放到ThreadLocal中去
                    readHolds.set(rh);
                rh.count++;
            }
            return 1;
        }
         // 通过这个方法再去尝试获取读锁（如果之前其它线程获取了写锁，一样返回-1表示失败）
        return fullTryAcquireShared(current);
    }
    
    
    //java.util.concurrent.locks.AbstractQueuedSynchronizer#doAcquireShared
    private void doAcquireShared(int arg) { 
         // 加入AQS的队列中
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try { 
            boolean interrupted = false;
            for (;;) { 
                //前一个节点
                final Node p = node.predecessor();
                //前一个节点是头结点，不用排队
                if (p == head) { 
                    // 再次尝试获取读锁
                    int r = tryAcquireShared(arg);
                     // 如果成功了
                    if (r >= 0) { 
                        // 头节点后移并传播
                        // 传播即唤醒后面连续的读节点
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                  // 没获取到读锁，阻塞并等待被唤醒
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }
    
    
    //java.util.concurrent.locks.AbstractQueuedSynchronizer#setHeadAndPropagate
    private void setHeadAndPropagate(Node node, int propagate) { 
        // h为旧的头节点
        Node h = head; // Record old head for check below
        //设置新的头结点
        setHead(node);
     	// 如果旧的头节点或新的头节点为空或者其等待状态小于0（表示状态为SIGNAL/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();
        }
    }
    
    // AbstractQueuedSynchronizer.doReleaseShared()
    // 这个方法只会唤醒一个节点
    private void doReleaseShared() { 
        for (;;) { 
            Node h = head;
            if (h != null && h != tail) { 
                int ws = h.waitStatus;
                // 如果头节点状态为SIGNAL，说明要唤醒下一个节点
                if (ws == Node.SIGNAL) { 
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // 唤醒下一个节点
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         // 把头节点的状态改为PROPAGATE成功才会跳到下面的if
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            // 如果唤醒后head没变，则跳出循环
            if (h == head)                   // loop if head changed
                break;
        }
    }

我们来看看大致的逻辑：

（1）先尝试获取读锁；

（2）如果成功了直接结束；

（3）如果失败了，进入doAcquireShared()方法；

（4）doAcquireShared()方法中首先会生成一个新节点并进入AQS队列中；

（5）如果头节点正好是当前节点的上一个节点，再次尝试获取锁；

（6）如果成功了，则设置头节点为新节点，并传播；

（7）传播即唤醒下一个读节点（如果下一个节点是读节点的话）；

（8）如果头节点不是当前节点的上一个节点或者（5）失败，则阻塞当前线程等待被唤醒；

（9）唤醒之后继续走（5）的逻辑；

**unlock()**

// java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock.unlock
    public void unlock() { 
        sync.releaseShared(1);
    }
    // java.util.concurrent.locks.AbstractQueuedSynchronizer.releaseShared
    public final boolean releaseShared(int arg) { 
        // 如果尝试释放成功了，就唤醒下一个节点
        if (tryReleaseShared(arg)) { 
            // 这个方法实际是唤醒下一个节点
            doReleaseShared();
            return true;
        }
        return false;
    }
    // java.util.concurrent.locks.ReentrantReadWriteLock.Sync.tryReleaseShared
    protected final boolean tryReleaseShared(int unused) { 
        Thread current = Thread.currentThread();
        if (firstReader == current) { 
            // 如果第一个读锁获取线程是当前线程
            // 就把它重入的次数减1
            // 如果减到0了就把第一个读锁获取线程置为空
            if (firstReaderHoldCount == 1)
                firstReader = null;
            else
                firstReaderHoldCount--;
        } else { 
            // 如果第一个读锁获取线程不是当前线程
            // 一样地，把它重入的次数减1
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();
            int count = rh.count;
            if (count <= 1) { 
                readHolds.remove();
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --rh.count;
        }
        for (;;) { 
            // 共享锁获取的次数减1
            // 如果减为0了说明完全释放了，才返回true
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
    // java.util.concurrent.locks.AbstractQueuedSynchronizer.doReleaseShared
    // 行为跟方法名有点不符，实际是唤醒下一个节点
    private void doReleaseShared() { 
        for (;;) { 
            Node h = head;
            if (h != null && h != tail) { 
                int ws = h.waitStatus;
                // 如果头节点状态为SIGNAL，说明要唤醒下一个节点
                if (ws == Node.SIGNAL) { 
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // 唤醒下一个节点
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         // 把头节点的状态改为PROPAGATE成功才会跳到下面的if
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            // 如果唤醒后head没变，则跳出循环
            if (h == head)                   // loop if head changed
                break;
        }
    }

解锁的大致流程如下：

（1）将当前线程重入的次数减1；

（2）将共享锁总共被获取的次数减1；

（3）如果共享锁获取的次数减为0了，说明共享锁完全释放了，那就唤醒下一个节点；

### 写锁 ###

**lock()**

// java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock.lock()
    public void lock() { 
        sync.acquire(1);
    }
    // java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire()
    public final void acquire(int arg) { 
        // 先尝试获取锁
        // 如果失败，则会进入队列中排队，后面的逻辑跟ReentrantLock一模一样了
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    // java.util.concurrent.locks.ReentrantReadWriteLock.Sync.tryAcquire()
    protected final boolean tryAcquire(int acquires) { 
        Thread current = Thread.currentThread();
        // 状态变量state的值
        int c = getState();
        // 写锁，互斥锁被获取的次数
        int w = exclusiveCount(c);
        if (c != 0) { 
            // 如果c!=0且w==0，说明共享锁被获取的次数不为0
            // 这句话整个的意思就是
            // 如果共享锁被获取的次数不为0，或者被其它线程获取了互斥锁（写锁）
            // 那么就返回false，获取写锁失败
            if (w == 0 || current != getExclusiveOwnerThread())
                return false;
            // 溢出检测
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            // 到这里说明当前线程已经获取过写锁，这里是重入了，直接把state加1即可
            setState(c + acquires);
            // 获取写锁成功
            return true;
        }
        // 如果c等于0，就尝试更新state的值（非公平模式writerShouldBlock()返回false）
        // 如果失败了，说明获取写锁失败，返回false
        // 如果成功了，说明获取写锁成功，把自己设置为占有者，并返回true
        if (writerShouldBlock() ||
            !compareAndSetState(c, c + acquires))
            return false;
        setExclusiveOwnerThread(current);
        return true;
    }

写锁获取的过程大致如下：

（1）尝试获取锁；

（2）如果有读者占有着读锁，尝试获取写锁失败；

（3）如果有其它线程占有着写锁，尝试获取写锁失败；

（4）如果是当前线程占有着写锁，尝试获取写锁成功，state值加1；

（5）如果没有线程占有着锁（state==0），当前线程尝试更新state的值，成功了表示尝试获取锁成功，否则失败；

（6）尝试获取锁失败以后，进入队列排队，等待被唤醒；  
（7）尝试获取锁失败，再调用addWaiter()构建新节点并尝试把新节点入队；

（8）入队失败则CAS循化更新尾结点直到入队成功。

（9）然后调用acquireQueued()再次尝试获取锁，如果成功了，直接返回；

（10）如果再次失败，再调用shouldParkAfterFailedAcquire()将节点的等待状态置为等待唤醒（SIGNAL）；

（11）调用parkAndCheckInterrupt()阻塞当前线程；

（12）如果被唤醒了，会继续在acquireQueued()的for()循环再次尝试获取锁，如果成功了就返回；

（13）如果不成功，再次阻塞，重复（9）（10）（11）直到成功获取到锁。

**unlock**

// java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock.unlock()
    public void unlock() { 
        sync.release(1);
    }
    //java.util.concurrent.locks.AbstractQueuedSynchronizer.release()
    public final boolean release(int arg) { 
        // 如果尝试释放锁成功（完全释放锁）
        // 就尝试唤醒下一个节点
        if (tryRelease(arg)) { 
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    // java.util.concurrent.locks.ReentrantReadWriteLock.Sync.tryRelease()
    protected final boolean tryRelease(int releases) { 
        // 如果写锁不是当前线程占有着，抛出异常
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        // 状态变量的值减1
        int nextc = getState() - releases;
        // 是否完全释放锁
        boolean free = exclusiveCount(nextc) == 0;
        if (free)
            setExclusiveOwnerThread(null);
        // 设置状态变量的值
        setState(nextc);
        // 如果完全释放了写锁，返回true
        return free;
    }

写锁释放的过程大致为：

（1）先尝试释放锁，即状态变量state的值减1；

（2）如果减为0了，说明完全释放了锁；

（3）完全释放了锁才唤醒下一个等待的节点；

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h6YXVfaXRkb2c_size_16_color_FFFFFF_t_70]: /images/20230209/a3989d59a10d484e9b7c9084a94c3663.png