Java多线程并发ReentrantReadWriteLock源码分析

蜗牛 互联网技术资讯 2022-06-16 17 0

本篇内容主要讲解“Java多线程并发ReentrantReadWriteLock源码分析”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Java多线程并发ReentrantReadWriteLock源码分析”吧!

    ReadWriteLock

    ReentrantReadWriteLock 是基于 AbstractQueuedSynchronizer 并实现了 ReadWriteLock 接口实现的一个锁机制。ReadWriteLock 定义了读写锁的特性:

    public interface ReadWriteLock {
        /**
         * Returns the lock used for reading.
         */
        Lock readLock();
        /**
         * Returns the lock used for writing.
         */
        Lock writeLock();
    }

    ReadWriteLock 中定义了获取两种锁的方式,一个用于获取读锁、一个用于获取写锁。只要没有持有写锁的线程在执行,读锁可以同时被多个尝试读操作的线程持有,而写锁是排他锁。

    与互斥锁相比,读写锁在访问共享数据时允许更高级的并发特性,即每次只有一个线程可以执行写操作,并且在没有写操作时其他线程可以并发读取共享数据。从读操作的效率来看,如果是互斥锁每次只能一个线程执行读写操作,而读写锁可以多个线程读,写操作时才互斥,所以读写锁的执行效率更高。

    ReentrantReadWriteLock 源码分析

    前面的内容介绍了读写锁的含义和优势,接下来分析 Java 并发包中对它的实现 ReentrantReadWriteLock 。

    类关系

    public class ReentrantReadWriteLock implements ReadWriteLock, java.io.Serializable {
        abstract static class Sync extends AbstractQueuedSynchronizer {
          	static final class HoldCounter
            static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> 
        }
        static final class NonfairSync extends Sync
        static final class FairSync extends Sync
        public static class ReadLock implements Lock, java.io.Serializable
        public static class WriteLock implements Lock, java.io.Serializable
    }

    ReentrantReadWriteLock 实现了读写锁接口 ReadWriteLock 和序列化接口 Serializable 。

    它有一个抽象静态内部类 Sync ,Sync 是 AQS 的抽象子类,Sync 有两个静态实现 NonfairSync 和 FairSync ,这部分是锁逻辑的核心内容;Sync 还有两个内部数据结构类 HoldCounter 和 ThreadLocalHoldCounter 。

    ReadLock 和 WriteLock 分别对应了读锁和写锁,它们都实现了 Lock 接口和序列号接口 Serializable 。它们是 ReentrantReadWriteLock 中对不同操作的锁类型的实现,使用了装饰模式,本质上还是通过 Sync 的能力实现的。

    Sync

    核心逻辑是来自于 Sync 及其两个实现,Sync 继承自 AbstractQueuedSynchronizer ,自身有两个内部类 HoldCounter 和 ThreadLocalHoldCounter 。

    HoldCounter

    static final class HoldCounter {
        int count;          // initially 0
        // Use id, not reference, to avoid garbage retention
        final long tid = LockSupport.getThreadId(Thread.currentThread());
    }

    HoldCounter 是一个计数器,count 用来记录当前线程拥有读锁的数量,即读锁的重入次数;tid 用来记录当前线程唯一 ID 。

    Sync 有一个 cachedHoldCounter 属性,用来做缓存效果,避免每次都通过 ThreadLocal 去读取数据。

    ThreadLocalHoldCounter

    static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {
        public HoldCounter initialValue() {
            return new HoldCounter();
        }
    }

    ThreadLocalHoldCounter 重写了 ThreadLocal 的 initialValue() ,在 ThreadLocal 没有进行过 set 数据的情况下,默认读取到的值都来自于这个方法,也就是配合 ThreadLocal 使用,默认值返回一个新的 HoldCounter 实例。

    在 Sync 中,有一个属性 readHolds ,它的类型是 ThreadLocalHoldCounter ,用来做当前线程读锁重入计数器的 ThreadLocal 包装,便于线程读取自己的读锁重入计数器。

    属性

    Sync 中定义的属性包括:

    abstract static class Sync extends AbstractQueuedSynchronizer {
    		// 高16位为读锁,低16位为写锁
        static final int SHARED_SHIFT   = 16;
        // 读锁单位
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);  // 1 * 2^16 = 65536
        // 读锁最大数量
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;  // 2^16 - 1
        // 写锁最大数量
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;   // 2^16 - 1 独占标记
        // 当前线程读锁重入次数。当持有读锁的线程数量下降到0时删除。
        private transient ThreadLocalHoldCounter readHolds;
        // 缓存对象,避免每次都去从 ThreadLocal 查找。
        private transient HoldCounter cachedHoldCounter;
    		// 第一个获取读锁线程
        private transient Thread firstReader;
        // 第一个读锁线程重入读锁的计数
        private transient int firstReaderHoldCount;
        // ...
    }

    构造方法

    Sync() {
    		readHolds = new ThreadLocalHoldCounter(); 
        setState(getState()); // ensures visibility of readHolds
    }

    Sync 初始化方法创建了 ThreadLocalHoldCounter 并重新设置了 State ,为什么要重新设置呢?因为这里要读取当前线程最新的同步状态并重新设置,获取实时的同步状态。

    核心方法

    Sync 的关键方法包括:

    abstract static class Sync extends AbstractQueuedSynchronizer {
      	// 并发计数
        static int sharedCount(int c)
        static int exclusiveCount(int c)
    		// 阻塞检查
        abstract boolean readerShouldBlock();
        abstract boolean writerShouldBlock();
    		// 获取和释放写锁
        @ReservedStackAccess
        protected final boolean tryRelease(int releases)
        @ReservedStackAccess
        protected final boolean tryAcquire(int acquires)
    		// 获取和释放读锁
        @ReservedStackAccess
        protected final boolean tryReleaseShared(int unused)
        @ReservedStackAccess
        protected final int tryAcquireShared(int unused)
        final int fullTryAcquireShared(Thread current)
    		// 尝试加读写锁
        @ReservedStackAccess
        final boolean tryWriteLock()
        @ReservedStackAccess
        final boolean tryReadLock()
    
        // ... 
    }

    锁的计数方法

    首先是两个静态方法 sharedCount(int c) 和 exclusiveCount(int c) :

    /** 表示共享持有的数量。 */
    static int sharedCount(int c)    { return c >>> SHARED_SHIFT; } // 无符号右移,高位补 0 
    /** 表示独占持有的数量。 */
    static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

    参数 c 是 AQS 中的 state,根据 state 进行位运算。这两个方法可以根据锁自身的状态解析出持有读写锁的数量。

    • sharedCount ,表示占有读锁的线程数量。直接将 AQS 中的 state 右移 16 位,高位补 0,就可以得到读锁的线程数量,因为 state 的高十六位表示读锁,对应的低十六位表示写锁数量。

    • exclusiveCount,表示占有写锁的线程数量。直接将 AQS 的 state 和 (2^16 - 1) 做与运算,其等效于将 state 模上 2^16 。写锁数量由 state 的低十六位表示。

    读写锁阻塞检查方法

    第二组方法是 readerShouldBlock 和 writerShouldBlock ,用来检查当前的读锁/写锁是否会造成当前线程阻塞。

    // 获取和释放对公平锁和非公平锁使用相同的代码,不同点在于但在队列非空时是否/如何允许碰撞。
    
    // 如果当前线程在尝试获取读锁时,并且在其他符合条件的线程也在尝试获取读锁,由于策略其他等待线程占用了读锁,当前线程应该阻塞,则返回true。
    abstract boolean readerShouldBlock();
    
    // 如果当前线程在尝试获取写锁时,并且在其他符合条件的线程也在尝试获取写锁,由于策略其他等待线程占用了写锁,当前线程应该阻塞,则返回true。
    abstract boolean writerShouldBlock();

    这两个方法的实现在 Sync 的子类中 -- 公平策略实现 FairSync 和非公平策略实现 NonfairSync。

    公平策略实现 FairSync 和非公平策略实现 NonfairSync

        // 非公平策略
    		static final class NonfairSync extends Sync {
            private static final long serialVersionUID = -8159625535654395037L;
            final boolean writerShouldBlock() {
                return false; // 正在持有写锁的线程永不阻塞
            }
            final boolean readerShouldBlock() {
                return apparentlyFirstQueuedIsExclusive(); 
            }
        }
    		// 公平策略
        static final class FairSync extends Sync {
            private static final long serialVersionUID = -2274990926593161451L;
          
            final boolean writerShouldBlock() {
                return hasQueuedPredecessors();
            }
            final boolean readerShouldBlock() {
                return hasQueuedPredecessors();
            }
        }

    公平锁策略和非公平锁策略的实现,本质上的不同是这两个方法的实现。

    NonfairSync 非公平策略

    NonfairSync 中,执行写操作的线程是否应该进入阻塞状态的判断,直接是 false ,这是因为非公平策略下,如果当前自身已经拥有了写锁,直接重入,以独占的方式继续运行(所以是不公平的)。

    执行读操作的线程是否会阻塞,是通过 apparentlyFirstQueuedIsExclusive() 判断的,这个方法是 AQS 中的方法:

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

    这个方法的作用是,CLH 队列中的头节点和它的的 next 都存在的情况下,如果 next 节点不是 SharedNode ,且它的关联线程不为空的情况(即下一个锁不是共享锁,共享锁在读写锁里就是读锁)的情况,会导致当前执行读操作的线程进入阻塞状态,确保写操作的互斥特性。

    FairSync 公平策略

    FairSync 中,读写执行线程是否应该进入阻塞状态都是根据 hasQueuedPredecessors() 方法判断的:

        public final boolean hasQueuedPredecessors() {
            Thread first = null; Node h = head, s = h.next;
            if (h != null && (s == null || (first = s.waiter) == null || s.prev == null))
                first = getFirstQueuedThread(); // retry via getFirstQueuedThread
            return first != null && first != Thread.currentThread();
        }
    
        public final Thread getFirstQueuedThread() {
            Thread first = null, w; Node h, s;
            if ((h = head) != null && ((s = h.next) == null || (first = s.waiter) == null || s.prev == null)) {
                // traverse from tail on stale reads
                for (Node p = tail, q; p != null && (q = p.prev) != null; p = q)
                    if ((w = p.waiter) != null)
                        first = w;
            }
            return first;
        }

    hasQueuedPredecessors() 对 head 节点和它的 next 节点进行空检查,并检查下一个节点的执行线程和 prev 指针是否有值,满足条件的情况下通过 getFirstQueuedThread() 方法获取到队列中第一个节点关联的线程。最终返回的结过是检查这个线程不等于当前线程。

    如果存在等待队列第一个等待执行的线程,那么就优先执行这个线程。也就是说,不管当前线程是拥有读锁还是写锁,都优先执行等待队列第一个未执行节点,这里就能体现出公平,即优先执行等待队列中头一个等待的节点所关联的线程。

    Release 和 Acquire 方法组

    这一组方法是整个 Sync 的核心逻辑,也是加解锁核心逻辑。

    tryRelease

    @ReservedStackAccess
    protected final boolean tryRelease(int releases) {
    		if (!isHeldExclusively()) // 不是独占持有锁的情况,直接抛出异常。
    				throw new IllegalMonitorStateException();
    		int nextc = getState() - releases; // AQS 当前锁状态 - releases = 新的锁状态
    		boolean free = exclusiveCount(nextc) == 0; // 根据新的锁状态获取到独占写锁的数量 == 0
    		if (free) 
    				setExclusiveOwnerThread(null); // 持有写锁的线程数为0,更新当前独占线程引用
    		setState(nextc); 	// 无论是不是解锁了,都要更新锁状态
    		return free;			// 最后返回锁是否已经可用了
    }

    tryRelease(int releases) 用来尝试释放写锁。

    它的逻辑如下图:

    tryAcquire

    @ReservedStackAccess
    protected final boolean tryAcquire(int acquires) {
                /*
                 * 工作流程:
                 * 1. 如果写锁计数非零或所有者是不同的线程,则失败。
                 * 2. 如果写锁计数超过最大数量,失败(这只发生在计数非 0 的情况)。
                 * 3. 否则,如果这个线程是可重入的获取方式或者队列策略允许的话,它就有资格获得锁。
                 *    如果是,更新状态并设置 owner。
                 */
    		Thread current = Thread.currentThread(); // 当前线程
    		int c = getState();											 // 当前锁状态
    		int w = exclusiveCount(c);							 // 计算拥有写锁的线程数量
    		if (c != 0) { // 0 是锁可用状态,当前状态表面锁状态为被持有。
    				if (w == 0 || current != getExclusiveOwnerThread()) // 对应 【1】 的情况,写线程数量为0或者当前线程没有占有独占资源
    						return false;
            if (w + exclusiveCount(acquires) > MAX_COUNT) // 对应【2】的情况, 判断是否超过最高写线程数量
                throw new Error("Maximum lock count exceeded");
            // 重入获取写锁
            setState(c + acquires);
            return true; 
    		}
    		if (writerShouldBlock() || !compareAndSetState(c, c + acquires)) // 是否应该阻塞或更新状态是否成功,失败直接 return false;
    				return false;
    		setExclusiveOwnerThread(current); // 设置当前为持有锁的线程。
    		return true;
    }

    此函数用于获取写锁,首先会获取 state ,判断 state 是否为0。

    若为0,表示此时没有读锁线程,再判断写线程是否应该被阻塞,而在非公平策略下总是不会被阻塞,在公平策略下会进行判断(判断同步队列中是否有等待时间更长的线程,若存在,则需要被阻塞,否则,无需阻塞),之后在设置状态state,然后返回true。若state不为0,则表示此时存在读锁或写锁线程,若写锁线程数量为0或者当前线程为独占锁线程,则返回false,表示不成功,否则,判断写锁线程的重入次数是否大于了最大值,若是,则抛出异常,否则,设置状态state,返回true,表示成功。

    其函数流程图如下:

    Java多线程并发ReentrantReadWriteLock源码分析  java 第1张

    tryReleaseShared:

            @ReservedStackAccess
            protected final boolean tryReleaseShared(int unused) {
                Thread current = Thread.currentThread(); // 当前线程
                if (firstReader == current) { // 当前线程是否是第一个读线程
                    // assert firstReaderHoldCount > 0;
                    if (firstReaderHoldCount == 1) 
                        firstReader = null;  // 释放线程引用
                    else
                        firstReaderHoldCount--;  // 当前线程重入次数自减
                } else {
                    HoldCounter rh = cachedHoldCounter; // 获取当前线程的重入读锁的次数
                    if (rh == null || rh.tid != LockSupport.getThreadId(current))
                        rh = readHolds.get();
                    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;
                }
            }

    Java多线程并发ReentrantReadWriteLock源码分析  java 第2张

    tryAcquireShared :

            @ReservedStackAccess
            protected final int tryAcquireShared(int unused) {
                /*
                 * Walkthrough:
                 * 1. If write lock held by another thread, fail.
                 * 2. Otherwise, this thread is eligible for
                 *    lock wrt state, so ask if it should block
                 *    because of queue policy. If not, try
                 *    to grant by CASing state and updating count.
                 *    Note that step does not check for reentrant
                 *    acquires, which is postponed to full version
                 *    to avoid having to check hold count in
                 *    the more typical non-reentrant case.
                 * 3. If step 2 fails either because thread
                 *    apparently not eligible or CAS fails or count
                 *    saturated, chain to version with full retry loop.
                 */
                Thread current = Thread.currentThread();
                int c = getState();
                if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current) // 当独占线程不是当前线程
                    return -1;
                int r = sharedCount(c); // 共享读锁的线程数量
              	// 检查读线程不应该阻塞 and 持有读锁的线程数量小于 MAX_COUNT and 更新锁状态成功
                if (!readerShouldBlock() && r < MAX_COUNT && compareAndSetState(c, c + SHARED_UNIT)) {
                    if (r == 0) {  // 第一个尝试获取读锁的线程
                        firstReader = current;
                        firstReaderHoldCount = 1;
                    } else if (firstReader == current) {  // 第一个线程重入
                        firstReaderHoldCount++;
                    } else {	
                        HoldCounter rh = cachedHoldCounter;
                      	// 无缓存 or 当前线程不是计数器所在线程
                        if (rh == null || rh.tid != LockSupport.getThreadId(current)) 
                            cachedHoldCounter = rh = readHolds.get(); // 从 ThreadLocal 中读取
                        else if (rh.count == 0) 
                            readHolds.set(rh);
                        rh.count++; // 当前线程获取读锁次数 + 1
                    }
                    return 1;
                }
                return fullTryAcquireShared(current);
            }

    Java多线程并发ReentrantReadWriteLock源码分析  java 第3张

    最后执行到了 fullTryAcquireShared :

            final int fullTryAcquireShared(Thread current) {
                /*
                 * 这段代码与 tryAcquireShared 中的部分代码是冗余的,但总体上更简单,因为它不会使
                 * tryAcquireShared 在重试和懒加载读锁计数之间的交互复杂化。
                 */
                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 {
                          	// 不是重入的情况下,更新 HoldCounter
                            if (rh == null) { 
                                rh = cachedHoldCounter;
                                if (rh == null || rh.tid != LockSupport.getThreadId(current)) {
                                    rh = readHolds.get();
                                    if (rh.count == 0)
                                        readHolds.remove();
                                }
                            }
                            if (rh.count == 0)
                                return -1;
                        }
                    }
                  	// 共享读锁 == 最大数量,抛出异常
                    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 != LockSupport.getThreadId(current))
                                rh = readHolds.get();
                            else if (rh.count == 0)
                                readHolds.set(rh);
                            rh.count++;
                            cachedHoldCounter = rh; // cache for release
                        }
                        return 1;
                    }
                }
            }

    这个方法的整体逻辑与 tryAcquireShared 基本相同。

    ReadLock

    public static class ReadLock implements Lock, java.io.Serializable {
        private static final long serialVersionUID = -5992448646407690164L;
        private final Sync sync;
        protected ReadLock(ReentrantReadWriteLock lock) {
            sync = lock.sync;
        }
        public void lock() {
            sync.acquireShared(1);
        }
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
        }
        public boolean tryLock() {
            return sync.tryReadLock();
        }
        public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
            return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
        }
        public void unlock() {
            sync.releaseShared(1);
        }
        public Condition newCondition() {
            throw new UnsupportedOperationException();
        }
        // ... 
    }

    ReadLock 实现了 Lock 接口,代理调用到逻辑都是 Sync 中 Shared 组的核心方法。ReadLock 可以通过 readLock(): ReadLock 方法获取到。

    还有一点值得注意,newCondition() 方法直接抛出了异常,这是因为读锁是一种共享锁,不会导致互斥,所以也就不支持使用 Condition 控制阻塞与唤醒。

    WriteLock

    public static class WriteLock implements Lock, java.io.Serializable {
        private static final long serialVersionUID = -4992448646407690164L;
        private final Sync sync;
        protected WriteLock(ReentrantReadWriteLock lock) {
            sync = lock.sync;
        }
        public void lock() {
            sync.acquire(1);
        }
        public void lockInterruptibly() throws InterruptedException {
            sync.acquireInterruptibly(1);
        }
        public boolean tryLock() {
            return sync.tryWriteLock();
        }
        public boolean tryLock(long timeout, TimeUnit unit)
                throws InterruptedException {
            return sync.tryAcquireNanos(1, unit.toNanos(timeout));
        }
        public void unlock() {
            sync.release(1);
        }
        public Condition newCondition() {
            return sync.newCondition();
        }
        public String toString() {
            Thread o = sync.getOwner();
            return super.toString() + ((o == null) ? "[Unlocked]" : "[Locked by thread " + o.getName() + "]");
        }
       public boolean isHeldByCurrentThread() {
            return sync.isHeldExclusively();
        }
        public int getHoldCount() {
            return sync.getWriteHoldCount();
        }
    }

    写锁本质上也是代理 Sync 中的核心方法。

    读写锁降级

    锁降级指的是写锁降级为读锁,如果当前线程拥有写锁,将其释放然后再获取读锁,这种操作过程不是锁降级。锁降级是指把线程当前持有写锁,再去获取读锁,随后释放写锁,这个流程称为锁降级。

    public void processData() {
        readLock.lock();
        if (!update) {
            // 必须先释放读锁
            readLock.unlock();
            // 锁降级从写锁获取到开始
            writeLock.lock();
            try {
                if (!update) {
                    // 准备数据的流程(略)
                    update = true;
                }
                readLock.lock();
            } finally {
                writeLock.unlock();
            }
            // 锁降级完成,写锁降级为读锁
        }
        try {
            // 使用数据的流程(略)
        } finally {
            readLock.unlock();
        }
    }

    锁降级可以保证数据的可见性,如果再持有写锁的情况下,不先去获取读锁,直接释放写锁,再尝试获取读锁,这一系列操作中会有短暂的无锁状态,此时如果有其他线程获取了写锁并修改数据,那么当前线程就无法感知到数据更新,如果当前线程先获取了读锁,那么其他线程就会阻塞,直到当前线程释放读锁后才能获取写锁进行更新。

    读写锁 ReentrantReadWriteLock 不支持锁升级,目的是保证数据的可见性,如果读锁已被多个线程获取,其中任意线程成功获取了写锁,并更新了数据,那么这个更新对其他线程是不可见的,容易造成数据不一致问题。

    到此,相信大家对“Java多线程并发ReentrantReadWriteLock源码分析”有了更深的了解,不妨来实际操作一番吧!这里是蜗牛博客网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!

    免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:niceseo99@gmail.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

    评论