JUC锁框架源码阅读-ReentrantReadWriteLock

说明

我们都知道ArrayList不是线程安全的，在读的时候同时并发在写，在写的时候同时在读，会出现索引越界问题，解决这个问题醉简单的方式在写和读的地方都加上锁。都加上锁的话并发读也会产生互斥

但是为了性能 读的频繁写的时候比较少。我们都会允许读读共享，读写互斥，写写互斥,ReentrantReadWriteLock 就是读写锁 读读共享，读写互斥，写写互

为什么ArrayList线程不安全可以参考《ArrayList在多线程的线程不安全的几种体现》

ReentrantReadWriteLock 是基于AQS实现的 AQS源码阅读参考《JUC锁框架源码阅读-AbstractQueuedSynchronizer》

关于位运算可以参考:《java位运算符,&,|,>>>,>>,<<,<<<的区别》

初始化

main

 //<1>
 ReadWriteLock readWriteLock = new ReentrantReadWriteLock();

1.通过位运算符& 通过AQS state&EXCLUSIVE_MASK 进行运算，如果大于0则表示写锁以及被持有，则获取读锁(AQS共享锁)失败 返回-1加入CLH队列

2.如果小于0则表示没有写锁。读读共享(共享锁可以多线程持有)则尝试获取，通过state从高位右移可以获取到写锁数量。

3.先因为int是32位 16位是临界，则65536是最大支持个线程可获取读锁数量(一般够用了)，需要校验是否超过65536如果超过了直接获取失败 加入CLH队列

4.没有超过如果限制没有任何线程获取读锁，则直接CAS尝试获取

5.获取失败则自旋获取，每次获取都需要判断是否写锁被持有了，自旋直到获取成功或者因为写锁被持有，或者超出最大可获取数量返回对应结果(注获取失败都是返回-1 AQS将阻塞线程并加入CLH队列)

6.获取成功 因为可重入将更新ThreadLocal的HoldCounter count+1 

<1>

    public ReentrantReadWriteLock() {
        this(false);
    }
    public ReentrantReadWriteLock(boolean fair) {
        //可以sync是ReentrantReadWriteLock 的2个内部类。可以发现ReentrantReadWriteLock 也是支持公平和非公平的
        sync = fair ? new ReentrantReadWriteLock.FairSync() : new ReentrantReadWriteLock.NonfairSync();
        //读锁也是内部内 内部保存了ReentrantReadWriteLock的引用 意味着可以拿到sync
        readerLock = new ReentrantReadWriteLock.ReadLock(this);
        //写锁也是内部内 内部保存了ReentrantReadWriteLock的引用 意味着可以拿到sync
        writerLock = new ReentrantReadWriteLock.WriteLock(this);
    }

Sync

ReentrantReadWriteLock 的公平锁和非公平锁都继承了Sync

继续看类结构 他们只重写了部分方法 主要是为了实现公平和非公平锁,具体加锁都在父类Sync

 /**
     * Nonfair version of Sync
     */
    static final class NonfairSync extends ReentrantReadWriteLock.Sync {
        private static final long serialVersionUID = -8159625535654395037L;
        //重写了writerShouldBlock方法
        final boolean writerShouldBlock() {
            return false; // writers can always barge
        }
        final boolean readerShouldBlock() {
            return apparentlyFirstQueuedIsExclusive();
        }
    }

    /**
     * Fair version of Sync
     */
    static final class FairSync extends ReentrantReadWriteLock.Sync {
        private static final long serialVersionUID = -2274990926593161451L;
        //重写了writerShouldBlock
        final boolean writerShouldBlock() {
            return hasQueuedPredecessors();
        }
        //重写了readerShouldBlock
        final boolean readerShouldBlock() {
            return hasQueuedPredecessors();
        }
    }

ReentrantReadWriteLock 是通过将AQS state int 32转2进制通过高16位表示读锁 低16位表示写锁

 

    //因为int 是32位，通过16将他们一分为二 高位加读锁 低位加写锁
    static final int SHARED_SHIFT   = 16;
    //1转为二进制向左边移动16位
    static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
    //1转为二进制向左边移动16位 -1表示高的最小值
    static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
    //1转为二进制向左边移动16位 -1表示低的最大值
    static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

    //将state向右移动16位 则可以得到持有锁的数量 我的理解 就是高位的移动到后面当成16进制处理
    static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }

    /**
     * 位运算 将&左右2边转为二进制 同时为1 则为1否则为0 得出写锁数量
     * 00000000 00000001 00000000 00000000  高位读锁与EXCLUSIVE_MASK计算后 =0
     * 00000000 00000001 00000000 00000001  低位写锁与EXCLUSIVE_MASK计算后  =1
     * 00000000 00000000 11111111 11111111 EXCLUSIVE_MASK 
     * @param c
     * @return
     */
    static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

HoldCounter

    //Sync内部类 锁是可重入的,保存每个线程持有数量
    static final class HoldCounter {
        //计数器
        int count = 0;
        //线程id
        // Use id, not reference, to avoid garbage retention
        final long tid = getThreadId(Thread.currentThread());
    }

    /**
     *Sync内部类 重新了ThreadLocal的 initialValue方法，为ThreadLocal设置默认值
     */
    static final class ThreadLocalHoldCounter
            extends ThreadLocal<ReentrantReadWriteLock.Sync.HoldCounter> {
        public ReentrantReadWriteLock.Sync.HoldCounter initialValue() {
            return new ReentrantReadWriteLock.Sync.HoldCounter();
        }
    }
    //sync成员变量
    private transient ReentrantReadWriteLock.Sync.ThreadLocalHoldCounter readHolds;

读锁加锁

main

 public static void main(String[] args)  {
        //<L>初始化
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        //获得读锁 内部return readLock
        Lock rlock = readWriteLock.readLock();
        //<1>加锁
        rlock.lock();
    }

<1>lock

public void lock() {
        //<2>调用的继承父类的AQS方法
        sync.acquireShared(1);
    }

<2>acquireShared

可以参考:《AQS源码阅读》

// 获取共享锁
    public final void acquireShared(int arg) {
        //<3>模板模式 抽象方法 由子类实现 尝试去获取共享锁，如果返回值小于0表示获取共享锁失败 这里调用的是Snyc
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

<3>tryAcquireShared

java.util.concurrent.locks.ReentrantReadWriteLock.Sync#tryAcquireShared

 protected final int tryAcquireShared(int unused) {
        //获得当前线程
        Thread current = Thread.currentThread();
        //获得AQS状态
        int c = getState();
        /**
         * exclusiveCount判断是否已经持有了写锁
         * getExclusiveOwnerThread 获得持有写锁的线程，如果不是当前线程持有直接返回-1 获取读锁失败 这里说明 同一个线程加了写锁是可以继续获取读锁的
         */
        if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
            return -1;
        //获取获取读锁的数量
        int r = sharedCount(c);
        /**
         * readerShouldBlock 公平锁的实现 NonfairSync FairSync都重新了此方法
         * FairSync(公平锁):直接如果队列不为空就返回false
         * NonfairSync（非公平锁）遍历CLH队列如果已经在队列中表示轮到他获取锁了返回true否则返回false 需要重新加入CLH队列
         * r < MAX_COUNT是否超了 读锁的获取临界，因为state划分了高位和低位所以不能超过最大值 1<<16-1=65536 所以同时支持65536 获取这么多度锁。。理论上够用了
         * compareAndSetState(c, c + SHARED_UNIT)
         */
        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 {
                //获取最近一个获取成功获取读锁的计数器 这省却了ThreadLocal查找，
                ReentrantReadWriteLock.Sync.HoldCounter rh = cachedHoldCounter;
                //判断是否是当前线程的 如果不是则从ThreadLocal查找 并设置到最近一个获取
                if (rh == null || rh.tid != getThreadId(current))
                    cachedHoldCounter = rh = readHolds.get();
                else if (rh.count == 0)
                    readHolds.set(rh);
                //计数器+1
                rh.count++;
            }
            return 1;
        }
        //<4>并发获取锁的情况 尝试获取读锁失败，自旋重试。
        return fullTryAcquireShared(current);
    }

<4>fullTryAcquireShared

 final int fullTryAcquireShared(Thread current) {
        ReentrantReadWriteLock.Sync.HoldCounter rh = null;
        //自旋
        for (;;) {
            //获取AQS 锁状态
            int c = getState();
            /**
             * exclusiveCount判断是否已经持有了写锁
             * getExclusiveOwnerThread 获得持有写锁的线程，如果不是当前线程持有直接返回-1 获取读锁失败 这里说明 同一个线程加了写锁是可以继续获取读锁的
             */
            if (exclusiveCount(c) != 0) {
                //如果写锁以及被获取 同时不是当前线程直接返回-1 不自旋
                if (getExclusiveOwnerThread() != current)
                    return -1;
                /**
                 *          * readerShouldBlock 公平锁的实现 NonfairSync FairSync都重新了此方法
                 *          * FairSync(公平锁):直接如果队列不为空就返回false
                 *          * NonfairSync（非公平锁）遍历CLH队列如果已经在队列中表示轮到他获取锁了返回true否则返回false 需要重新加入CLH队列
                 */
            } else if (readerShouldBlock()) {
                // 第一个获取读锁的是当前线程不做处理
                if (firstReader == current) {
                    // assert firstReaderHoldCount > 0;
                } else {
                    //更新锁计数（可重入的体现）
                    if (rh == null) {
                        //获取最近一个获取读锁线程的Counter
                        rh = cachedHoldCounter;
                        if (rh == null || rh.tid != getThreadId(current)) {
                            //如果不是最近的从ThreadLocal获取
                            rh = readHolds.get();
                            if (rh.count == 0)
                                //如果当前线程的持有读锁数为0，那么就没必要使用计数器，直接移除
                                readHolds.remove();
                        }
                    }
                    if (rh.count == 0)
                        //结束自旋
                        return -1;
                }
            }
            //超过最大可获取数量
            if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            //自旋获取共享锁 获取成功Countter++
            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;
            }
        }
    }

读锁释放锁

1.每次释放通过HoldCounter 的count -1直到减为0则返回AQS父类释放成功,将AQS的state -1,并将计数器从ThreadLcal清除

2.AQS则会唤醒CLH阻塞队列重新尝试重新获取锁

main

   public static void main(String[] args)  {
        //<L>初始化
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        //获得读锁 内部return readLock
        Lock rlock = readWriteLock.readLock();
        rlock.lock();
        //<1>
        rlock.unlock();

    }

<1>unlock

   public void unlock() {
        //<2>调用Sync继承AQC方法
        sync.release(1);
    }

<2>releaseShared

可以参考《AQS源码》

// 释放共享锁
    public final boolean releaseShared(int arg) {
        //<3>模板模式 抽象方法尝试释放共享锁
        if (tryReleaseShared(arg)) {
            // 唤醒等待共享锁的线程
            doReleaseShared();
            return true;
        }
        return false;
    }

<3>tryReleaseShared

 protected final boolean tryReleaseShared(int unused) {
        Thread current = Thread.currentThread();
        //是不第一个获取锁
        if (firstReader == current) {
            //可重入的特点 直接置空
            if (firstReaderHoldCount == 1)
                firstReader = null;
            else
                firstReaderHoldCount--;//计数器-1
        } else {
            //判断是不是最近获取读锁的线程
            ReentrantReadWriteLock.Sync.HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();//如果不是 从ThreadLocal获取
            int count = rh.count;
            //可重入特点 判断数量 觉得
            if (count <= 1) {
                readHolds.remove();
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --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;
        }
    }

写锁加锁

main

 //<L>初始化
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        //获得读锁 内部return readLock
        Lock writeLock = readWriteLock.writeLock();
        //<1>加锁
        writeLock.lock();

1.如果AQS state不等于0,表示有可能获取的写锁或者读锁,根据位运算符&获取写锁数量，如果数量=0那么表示是读锁。直接获取失败(加入CLH队列阻塞)

2.如果获取写锁数量大于0 根据getExclusiveOwnerThread() 判断是否不是当前线程持有，如果是的话直接获取成功(可重入) state+1并通过CAS修改

<1>lock

public void lock() {
        //<2>调用继承自AQS加锁方法
        sync.acquire(1);
    }

<2>acquire

可以参考《AQS源码阅读》

public final void acquire(int arg) {
        /**
         *<3>tryAcquire  模板模式 是抽象方法由子类实现获取锁步骤
         *addWaiter   如果加锁失败 创建节点并放入队列尾
         *<acquireQueued 通过新创建的节点 判断是重试获取锁。还是阻塞线程
         */
        if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(node.EXCLUSIVE), arg))
            selfInterrupt();
    }

<3>tryAcquire

 protected final boolean tryAcquire(int acquires) {
        Thread current = Thread.currentThread();
        //获得AQS锁状态
        int c = getState();
        //获得写锁数量
        int w = exclusiveCount(c);
        //>0表示写锁或者读锁被获取
        if (c != 0) {
            //==0表示读锁被持有 直接返回false 加入CLH队列 如果是写锁判断是是否是当前线程持有(可重入)
            if (w == 0 || current != getExclusiveOwnerThread())
                return false;
            //判断是否有超过可获取写锁数量
            if (w + exclusiveCount(acquires) > MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            //则state++ 可重入
            setState(c + acquires);
            return true;
        }
        /**
         * writerShouldBlock 公平锁的实现 NonfairSync FairSync都重新了此方法
         * FairSync(公平锁): 不在队列返回true 在队列返回false
         * NonfairSync（非公平锁）直接返回false 尝试直接获取锁
         * r < MAX_COUNT是否超了 读锁的获取临界，因为state划分了高位和低位所以不能超过最大值 1<<16-1=65536 所以同时支持65536 获取这么多度锁。。理论上够用了
         * compareAndSetState(c, c + SHARED_UNIT)
         */
        if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
            return false;
        //设置当前持有锁线程
        setExclusiveOwnerThread(current);
        return true;
    }

写锁释放锁

main 

  public static void main(String[] args)  {
        //<L>初始化
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        //获得读锁 内部return readLock
        Lock writeLock = readWriteLock.writeLock();
        //<1>释放锁
        writeLock.unlock();

    }

1.getExclusiveOwnerThread() == Thread.currentThread() 判断是否是当前线程持有锁 如不过不是 直接抛错

2.AQS sate-- 更新到AQS state 如果大于0返回释放失败 如果==0  清空exclusiveOwnerThread 返回释放成功 后续交给AQS处理 唤醒CLH等待线程重新获取锁

<1>unlock

    public void unlock() {
        //<2>调用的继承AQS的共享锁释放方法
        sync.release(1);
    }

<2>relase

可以参考 《AQS源码阅读》

// 在独占锁模式下，释放锁的操作
    public final boolean release(int arg) {
        // <3>调用tryRelease子类方法，尝试去释放锁，由子类具体实现
        if (tryRelease(arg)) {
            Node h = head;
            // <如果队列头节点的状态不是0，那么队列中就可能存在需要唤醒的等待节点。
            // 还记得我们在acquireQueued(final Node node, int arg)获取锁的方法中，如果节点node没有获取到锁，
            // 那么我们会将节点node的前一个节点状态设置为Node.SIGNAL，然后调用parkAndCheckInterrupt方法
            // 将节点node所在线程阻塞。
            // 在这里就是通过unparkSuccessor方法，进而调用LockSupport.unpark(s.thread)方法，唤醒被阻塞的线程
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

<3>tryRelease

 protected final boolean tryRelease(int releases) {
        //判断是不是当前线程持有getExclusiveOwnerThread() == Thread.currentThread(); 如果不是则抛异常 避免别的线程调用此方法释放了锁
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        //获取状态--
        int nextc = getState() - releases;
        //判断是否释放完了(可重入) 如果调用2次lock 则需要调用2次unlock才会真正释放锁
        boolean free = exclusiveCount(nextc) == 0;
        if (free)
            setExclusiveOwnerThread(null);//置空持有线程
        //设置新的锁状态 这里不用CAS是 读锁 只会有有一个线程持有
        setState(nextc);
        return free;
    }