多线程高并发编程(4) -- ReentrantReadWriteLock读写锁源码分析

public interface ReadWriteLock {
    Lock readLock();
    Lock writeLock();
}

public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    private final ReentrantReadWriteLock.ReadLock readerLock;//读锁内部类
    private final ReentrantReadWriteLock.WriteLock writerLock;//写锁内部类
    final Sync sync;//同步器
    public ReentrantReadWriteLock(boolean fair) {//默认使用非公平锁
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }
    /** 外部获得WriteLock、ReadLock的方法，由ReentrantReadWriteLock构造方法创建/
    * 获取实例：
    * ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    * ReentrantReadWriteLock.ReadLock readLock = rw.readLock();
    * ReentrantReadWriteLock.WriteLock writeLock = rw.writeLock();
    */
    public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
    public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }

    //ReadLock读锁内部类源码
    public static class ReadLock implements Lock, java.io.Serializable {
        private final Sync sync;
        public void lock() {//共享锁获取
            sync.acquireShared(1);//调用AQS的acquireShared
        }
        public void unlock() {//共享锁释放
            sync.releaseShared(1);//调用AQS的releaseShared
        }
    }

    //WriteLock写锁内部类源码
    public static class WriteLock implements Lock, java.io.Serializable {private final Sync sync;
        public void lock() {//排他锁获取
            sync.acquire(1);//调用AQS的acquire
        }
        public void unlock() {//排他锁释放
            sync.release(1);//调用AQS的release
        }
    }

    //内部类，同步器Sync，实现tryAcquireShared、tryAcquire、tryRelease，即具体实现锁的获取和释放
    abstract static class Sync extends AbstractQueuedSynchronizer{
            //.....................
    }

}

public abstract class AbstractQueuedSynchronizer{
    //ReadLock的lock调用
    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)//获取锁失败
            doAcquireShared(arg);//加入等待队列中
    }
    //acquireShared调用，由ReentrantReadWriteLock的Sync实现
    protected int tryAcquireShared(int arg) {
        throw new UnsupportedOperationException();
    }
}

    abstract static class Sync extends AbstractQueuedSynchronizer{
        /**
         * int为4字节，共有32位：1111 1111 1111 1111 - 1111 1111 1111 1111
         * 高16位(读状态)为读锁持有数量sharedCount；低16位(写状态)为写锁持有数量exclusiveCount；
         * 0000 0000 0000 0011  0000 0000 0000 0000表示有3个线程获取了读锁，0个线程获取了写锁
         */
        static final int SHARED_SHIFT   = 16;
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;//最多持有数量为65535
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }//返回某状态c中共享锁(读锁)使用的次数，相当于是获取高16位，忽略低位
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }//返回某状态c中独占锁(写锁)使用的次数，相当于是获取低16位，忽略高位
        //每个线程获取的读锁数量，用ThreadLocal来维护，数据缓存到cachedHoldCounter
        static final class HoldCounter {
            int count = 0;//初始值为0
            // 使用id而不是引用来避免垃圾保留【垃圾保留：无法回收的垃圾】
            final long tid = getThreadId(Thread.currentThread());
        }
        //维护HoldCounter
        static final class ThreadLocalHoldCounter
            extends ThreadLocal<HoldCounter> {
            public HoldCounter initialValue() {
                return new HoldCounter();//为每个线程创建一个HoldCounter，保存到ThreadLocal中
            }
        }
        //readHolds当前线程持有的数据对象，包含锁数量和线程id，当线程数量为0时会移除该对象
        private transient ThreadLocalHoldCounter readHolds;
        //缓存上一个成功获取读锁的线程的数据对象，减少ThreadLocal的查找操作
        private transient HoldCounter cachedHoldCounter;
        //第一个读线程的对象
        private transient Thread firstReader = null;
        //第一个读线程持有的锁数量
        private transient int firstReaderHoldCount;
        Sync() {
            readHolds = new ThreadLocalHoldCounter();//创建当前线程数据对象
            setState(getState()); // 更新数据，确保readHolds的可见性
        }
}

        protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();//获得当前线程
            int c = getState();//得到状态
            /**
            * 当前有写线程且本线程不是写线程，不符合重入，失败
            * 此处判断逻辑隐含了一个条件，就是当有写锁获取并且是获取写锁的是当前线程，那么不返回-1，允许此写锁获取读锁。【锁降级】
            * 锁降级指的是先获取到写锁，然后获取到读锁，然后释放了写锁的过程。
            * 流程：writeLock.lock();  readLock.lock();  writeLock.unlock(); readLock.unlock();
　　　　　　  * 锁降级的应用场景: 对于数据比较敏感, 需要在对数据修改以后, 获取到修改后的值, 并进行接下来的其它操作。
            * 锁降级中读锁的获取是否必要呢？
            * 答案是必要的。主要是为了保证数据的可见性，如果当前线程不获取读锁而是直接释放写锁， 假设此刻另一个线程（记作线程T）获取了写锁并修改了数据，那么当前线程无法感知线程T的数据更新。如果当前线程获取读锁，即遵循锁降级的步骤，则线程T将会被阻塞，直到当前线程使用数据并释放读锁之后，线程T才能获取写锁进行数据更新。
            * 保证数据的可见性可以这样理解：假设线程A修改了数据，释放了写锁，这个时候线程T获得了写锁，修改了数据，然后也释放了写锁，线程A读取数据的时候，读到的是线程T修改的，并不是线程A自己修改的，那么在使用修改后的数据时，就会忽略线程A之前的修改结果。当前线程无法感知线程T的数据更新，是说线程A使用数据时，并不知道别的线程已经更改了数据，所以使用的是线程T的修改结果。因此通过锁降级来保证数据每次修改后的可见性。
            * 锁是不支持锁升级的（先获取写锁，再获取读锁然后释放读锁），
            * 因为第一步获取读锁的时候可能有多个线程获取了读锁，这样如果锁升级的话将会导致写操作对其他已经获取了读锁的线程不可见。
            */
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            //得到读锁个数
            int r = sharedCount(c);
            //如果读不应该阻塞并且当前持有的锁个数小于最大值65535，且可以更新状态值，成功
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
                if (r == 0) {//读锁个数为0，即第一个读
                    firstReader = current;//第一个读线程就是当前线程
                    firstReaderHoldCount = 1;//第一个读线程持有的锁数量为1
                } else if (firstReader == current) {//当前线程重入
                    firstReaderHoldCount++;//锁数量+1
                } else {//当前线程和第一个读线程不一致(不是重入），从ThreadLocal中获取当前线程重入读锁的次数，然后自增下。
                    HoldCounter rh = cachedHoldCounter;//获取最后一次成功获取读锁的线程数据对象
                    //rh == null(当前线程是第二个获取的),或者当前线程和rh不是同一个
                    if (rh == null || rh.tid != getThreadId(current))
                        //把最后一次成功获取读锁的线程数据对象更新为当前线程的数据对象
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)//最后一次成功获取读锁的锁数量为0
                        readHolds.set(rh);//当前线程设置为HoldCounter，即把当前线程数据存放到ThreadLocal中进行维护
                    rh.count++;//获取到的读锁数量+1
                }
                return 1;
            }
            //如果读锁获取失败，调用该方法进行CAS循环获取
            return fullTryAcquireShared(current);
        }

        final int fullTryAcquireShared(Thread current) {
            //这段代码在一定程度上与tryacquirered中的代码是冗余的，但总的来说更简单，
            //因为它没有使tryacquirered在重试和延迟读锁计数之间的交互复杂化。
            HoldCounter rh = null;
            //自旋
            for (;;) {
                int c = getState();
                //已经有写锁被获取
                if (exclusiveCount(c) != 0) {
                    //当前线程不是重入，获取失败
                    if (getExclusiveOwnerThread() != current)
                        return -1;
                } else if (readerShouldBlock()) {//有写锁，读锁被阻塞，可能会造成死锁
                    if (firstReader == current) {
                        // assert firstReaderHoldCount > 0;
                    } else {
                        //第一次循环
                        if (rh == null) {
                            rh = cachedHoldCounter;//最后一次成功获取读锁的数据对象
                            if (rh == null || rh.tid != getThreadId(current)) {
                                rh = readHolds.get(); //把最后一次成功获取读锁的线程数据对象更新为当前线程的数据对象
                                if (rh.count == 0)//如果当前线程的读锁为0就remove
                                    readHolds.remove();
                            }
                        }
                        //不是第一次循环
                        if (rh.count == 0)
                            return -1;
                    }
                }
                if (sharedCount(c) == MAX_COUNT)//读锁数量达到临界值抛出异常
                    throw new Error("Maximum lock count exceeded");
                //尝试CAS设置同步状态
                //后续操作和tryAquireShared基本一致
                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;
                    }
                    return 1;
                }
            }
        }

/**
*  非公平锁的读锁获取策略
*/ 
final boolean readerShouldBlock() {
    //如果当前线程的后续节点为独占式写线程，则返回true(表示当前线程在tryAcquireShared方法中不能立刻获取读锁，需要后续通过fullTryAcquireShared方法取判断是否需要阻塞线程)
    //在fullTryAcquireShared方法中会通过判断当前获取读锁线程的读锁数量来判断当前尝试获取读锁的线程是否持有写锁，如果持有写锁则锁降级，需要将当前锁降级的线程添加到阻塞队列中重新获取读锁
    //这么做是为了让后续的写线程有抢占写锁的机会，不会因为一直有读线程或者锁降级情况的存在而造成后续写线程的饥饿等待
    return apparentlyFirstQueuedIsExclusive();
}

final boolean apparentlyFirstQueuedIsExclusive() {
    Node h, s;
    return (h = head) != null &&
        (s = h.next)  != null &&
        !s.isShared()         &&
        s.thread != null;
}

/**
*  公平锁的读锁获取策略
*/  
final boolean readerShouldBlock() {
    //如果当前线程不是同步队列头结点的next节点（head.next) （判断是否有前驱节点，如果有则返回false，否则返回true。遵循FIFO）
    //则阻塞当前线程 
    return hasQueuedPredecessors();
}

public final boolean hasQueuedPredecessors() {
    Node t = tail; // Read fields in reverse initialization order
    Node h = head;
    Node s;
    return h != t &&
        ((s = h.next) == null || s.thread != Thread.currentThread());
}

    private void doAcquireShared(int arg) {
    //把当前线程封装到一个SHARE类型Node中，添加到SyncQueue尾巴上，即加入等待队列中
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {//前继节点是head节点，下一个就到自己了
                    int r = tryAcquireShared(arg);//非公平锁实现，再尝试获取锁
                    if (r >= 0) {//获取成功
                        setHeadAndPropagate(node, r);//把当前节点更新为head节点并唤醒线程
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                //前继节点非head节点，将前继节点状态设置为SIGNAL，通过park挂起node节点的线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

    public final boolean releaseShared(int arg) {//ReadLock的unlock调用
        if (tryReleaseShared(arg)) {//锁释放成功
        //此处的doReleaseShared方法与setHeadAndPropagate方法中锁唤醒的节点有所差别
        //setHeadAndPropagate方法只唤醒head后继的共享锁节点
        //doReleaseShared方法则会唤醒head后继的独占锁或共享锁
            doReleaseShared();
            return true;
        }
        return false;
    }
    //tryReleaseShared调用，由ReentrantReadWriteLock的Sync实现
    protected boolean tryReleaseShared(int arg) {
        throw new UnsupportedOperationException();
    }

        //读锁释放，实现AQS的tryReleaseShared
        protected final boolean tryReleaseShared(int unused) {
            Thread current = Thread.currentThread();//获取当前线程
            if (firstReader == current) {//当前线程是第一个读线程
                // assert firstReaderHoldCount > 0;
                if (firstReaderHoldCount == 1)//第一个读线程的锁数量为1
                    firstReader = null;//可以让其他线程获取锁
                else
                    firstReaderHoldCount--;//锁数量-1
            } else {
                //获取最后一次成功获取锁的数据对象
                HoldCounter rh = cachedHoldCounter;
                //获取当前线程的数据对象
                if (rh == null || rh.tid != getThreadId(current))
                    rh = readHolds.get();
                //获取当前线程持有的读锁数量
                int count = rh.count;
                if (count <= 1) {//小于1
                    readHolds.remove();//移除当前线程的数据信息
                    if (count <= 0)//小于0抛出异常
                        throw unmatchedUnlockException();
                }
                --rh.count;//当前线程的读锁数量-1
            }
            for (;;) {//自旋
                int c = getState();//获取状态
                //释放后的同步状态
                int nextc = c - SHARED_UNIT;
                //CAS更新同步状态，成功则返回是否同步状态为0
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }

    /** releaseShared调用
     * 把当前结点设置为SIGNAL或者PROPAGATE
     * 唤醒head.next(B节点)，B节点唤醒后可以竞争锁，成功后head->B，然后又会唤醒B.next，一直重复直到共享节点都唤醒
     * head节点状态为SIGNAL，重置head.waitStatus->0，唤醒head节点线程，唤醒后线程去竞争共享锁
     * head节点状态为0，将head.waitStatus->Node.PROPAGATE传播状态，表示需要将状态向后继节点传播
     */
    private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {//head是SIGNAL状态
                   /* head状态是SIGNAL，重置head节点waitStatus为0，这里不直接设为Node.PROPAGATE,
                    * 是因为unparkSuccessor(h)中，如果ws < 0会设置为0，所以ws先设置为0，再设置为PROPAGATE
                    * 这里需要控制并发，因为入口有setHeadAndPropagate跟release两个，避免两次unpark
                    */
                    if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))
                        continue;//设置失败，重新循环
                        /* head状态为SIGNAL，且成功设置为0之后,唤醒head.next节点线程
                         * 此时head、head.next的线程都唤醒了，head.next会去竞争锁，成功后head会指向获取锁的节点，
                         * 也就是head发生了变化。看最底下一行代码可知，head发生变化后会重新循环，继续唤醒head的下一个节点
                         */
                        unparkSuccessor(h);
                /*
                 * 如果本身头节点的waitStatus是出于重置状态（waitStatus==0）的，将其设置为“传播”状态。
                 * 意味着需要将状态向后一个节点传播
                 */
                }
                else if (ws == 0 && !h.compareAndSetWaitStatus(0, Node.PROPAGATE))
                    continue;
            }
            if (h == head)//如果head变了，重新循环
            break;
        }
    }

    public final void acquire(int arg) {//WriteLock的lock调用
        //获取失败则会调用addWaiter方法将线程添加到CLH队列末尾，并调用acquireQueued方法阻塞当前线程
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }
    //acquire调用，由ReentrantReadWriteLock的Sync实现
    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }

        protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();//获得当前线程
            int c = getState();//获得状态
            //获取写锁数量（w>0表示已经有线程获取写锁）
            int w = exclusiveCount(c);
            if (c != 0) {//同步状态不为0，说明线程获取到了写锁或读锁
                 //写锁状态0（表示有线程已经获取读锁（共享锁获取时阻塞独占锁））或者当前线程不是已经获取写锁的线程（独占锁只允许自己持有锁）
                 //返回false
                 //此处的处理逻辑也间接验证了获取了读锁的线程不能同时获取写锁
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                //大于最大线程数则抛出错误
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                //如果写锁状态>0并且当前线程为写锁重入，更新写锁状态
                setState(c + acquires);
                return true;
            }
            //如果同步状态等于0
            //在尝试获取同步状态之前先调用writerShouldBlock()写等待策略
            //ReentrantReadWriteLock中通过FairSync（公平锁）和NonfairSync（非公平锁）重写writerShouldBlock()方法来达到公平与非公平的实现
            //NonfairSync（非公平锁）中直接返回false表示不进行阻塞直接获取
            //FairSync（公平锁）中需调用hasQueuedPredecessors()方法判断当前线程节点是否为等待队列的head结点的后置节点，是才可以获取锁
            //获取成功则将当前线程设置为持有锁线程，并返回true
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;
        }

   public final boolean release(int arg) {//WriteLock的unlock调用
    //tryRelease释放锁，直到写锁state等于0（所有重入锁都释放），唤醒后续节点线程
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    //release调用，由ReentrantReadWriteLock的Sync实现
    protected boolean tryRelease(int arg) {
        throw new UnsupportedOperationException();
    }

        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);//独占锁设置为null，让其他线程可以获取到锁
            setState(nextc);//更新状态
            return free;
        }