【源码解析】聊聊ReentrantReadWriteLock是如何实现的读写锁

2023-12-21 23:35:05

为什么需要读写锁

在并发编程领域,有多线程进行提升整体性能,但是却引入了共享数据安全性问题。基本就是无锁编程下的单线程操作,有互斥同步锁操作,但是性能不高,并且同一时刻只有一个线程可以操作资源类。但是对于大多数常见下,都是读操作多,写操作少,那么可以利用将锁的粒度进行细化,进而分化出读锁/写锁。也就是syn/ReentrantLock的升级版本ReentrantReadWriteLock。

之前一篇文章已经简单介绍过 ,本篇主要从源码角度剖析具体原理如何实现的。
聊聊ReentrantReadWriteLock锁降级和StampedLock邮戳锁

源码解析

带着三个问题去梳理

  • 读写锁是怎样实现分别记录读写锁的状态?
  • 读锁如何获取和释放锁
  • 写锁如何获取和释放锁

在这里插入图片描述
可以看到顶层通过接口定义规范,内部持有Sync实现AQS,分别实现不同的公平锁和非公平锁。
在这里插入图片描述

//读写锁的接口规范
public interface ReadWriteLock {

    Lock readLock();

    Lock writeLock();
}
// 内部持有读写锁 
public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    private static final long serialVersionUID = -6992448646407690164L;
   
    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);
    }

锁状态

看到这里其实有点懵逼,什么 这都是什么操作,其实在AQS内部通过一个变量state进行控制是否可以获取资源,但是读写锁如何要用两个变量的话,其实不太好,所以就通过高16位代表读锁的状态、低16位代表写锁的状态。

对于低16来说,值等于0没有加写锁,值等于1 加了写锁,大于1 标识写锁的重入次数。
高16来说,0 :没有加读锁, 1: 加读锁。 值大于1 不表示读锁的重入次数,表示读锁总共被获取了多少次。读锁的重入次数存储在和线程相关的地方,通过threadLocal进行存储。
在这里插入图片描述

  abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 6317671515068378041L;

          // 偏移位数
        static final int SHARED_SHIFT   = 16;
        // 共享锁基本单位  左移16位 state+= shared_unit
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        // 读锁、写锁 可重入最大数量
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
        // 获取低16位的条件
        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  */
        // 写锁 是否持有 1 为一个线程持有 2 1次冲入 1次获取写锁
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

写状态,等于 S & 0x0000FFFF(将高 16 位全部抹去)。 当写状态加1,等于S+1.
读状态,等于 S >>> 16 (无符号补 0 右移 16 位)。当读状态加1,等于 S+(1<<16),也就是S+0x00010000。

这样 我们就完成了一个state值可以同时表示两种状态的。

写锁

写锁加锁

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

调用AQS的获取

    public final void acquire(int arg) {
        //tryAcquire(arg) true 获取锁成功直接结束
        //如果没有获取到锁,acquireQueued 会将线程压入队列中
        //!tryAcquire(arg)  没有获取到锁,将当前线程挂起
        //addWaiter
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

ReentrantReadWriteLock内部实现了tryAcquire方法。
该方法主要的作用就是
1.获取当前线程
2.判断state的状态。 c = 0 说明当前没有读锁和写锁,通过CAS进行设置state的值 直接获取锁
3.state值不等于0,w == 0 说明当前有读锁 获取锁失败,返回
4.w != 0 说明 当前是写锁重入,所以判断是否最大值,设置state的值+1
writerShouldBlock() 方法会根据是否是公平锁进行排队处理

       protected final boolean tryAcquire(int acquires) {
            // 获取当前线程
            Thread current = Thread.currentThread();
            // 获取state的值
            int c = getState();
            int w = exclusiveCount(c);
            // c = 0 说明 当前没有读锁和写锁
            if (c != 0) {
                // w == 0 等于0 说明 说明当前有读锁  或者当前线程不等于持有锁的线程
                // 写读互斥
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                // 获取写锁 不大于最大值
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // Reentrant acquire
                // 设置当前值 说明可重入
                setState(c + acquires);
                return true;
            }
            // 是否需要阻塞 公平锁
            if (writerShouldBlock() ||
                    //CAS 设置c的值 c += 1
                !compareAndSetState(c, c + acquires))
                return false;
            // 设置为当前线程
            setExclusiveOwnerThread(current);
            return true;
        }

在这里插入图片描述

在这里插入图片描述

写锁释放锁

当当前线程执行完毕业务逻辑之后,就会释放锁。

        public void unlock() {
            sync.release(1);
        }
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                //唤醒阻塞等待的线程
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

释放锁的流程主要就是
1.判断持有锁的线程是否属于当前线程,不是直接异常
2.将state-1 ,state = 0的话,说明重入的锁释放完毕。清空
3.设置state的值,可能是-1 或者 为0。

        protected final boolean tryRelease(int releases) {
            // 持有锁的线程 是否等于当前线程
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            // 将当前state -= 1
            int nextc = getState() - releases;
            boolean free = exclusiveCount(nextc) == 0;
            // 如果写锁为0 说明当前没有锁持有了
            if (free)
                // 将当前线程释放
                setExclusiveOwnerThread(null);
            // 设置state的值
            setState(nextc);
            return free;
        }

在这里插入图片描述

读锁

读锁加锁

        public void lock() {
            sync.acquireShared(1);
        }
    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
        protected final int tryAcquireShared(int unused) {
            // 获取当前线程
            Thread current = Thread.currentThread();
            int c = getState();
            //判断是否有写锁,并且当前线程不是持有写锁线程
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
            // 获取读锁
            int r = sharedCount(c);
            // 是否需要阻塞
            if (!readerShouldBlock() &&
                    //是否小于最大值
                r < MAX_COUNT &&
                    //CAS 设置  高16位加1
                compareAndSetState(c, c + SHARED_UNIT)) {
                // 第一次获取读锁
                if (r == 0) {
                    //设置第一个获取读锁的线程
                    firstReader = current; // 当前线程
                    //设置第一个获取读锁线程的重入数
                    firstReaderHoldCount = 1; //
                } else if (firstReader == current) {
                    // 如果当前线程是第一个获取读锁的线程,重入数++
                    firstReaderHoldCount++;
                } else {
                    //刷新除获取锁的第一个读线程的重入数
                    // threadLocal进行记录线程重入次数
                    HoldCounter rh = cachedHoldCounter;
                    if (rh == null || rh.tid != getThreadId(current))
                        cachedHoldCounter = rh = readHolds.get();
                    else if (rh.count == 0)
                        readHolds.set(rh);
                    rh.count++;
                }
                return 1;
            }
            // 再次尝试获取读锁,
            return fullTryAcquireShared(current);
        }

在这里插入图片描述
从这里可以看到,支持锁降级,持有写锁的线程,可以获取读锁,但是后续要记得把读锁和写锁读释放

读锁释放锁

        public void unlock() {
            sync.releaseShared(1);
        }
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
        protected final boolean tryReleaseShared(int unused) {
            Thread current = Thread.currentThread();
            // 如果当前线程是第一个获取读锁的线程
            if (firstReader == current) {
                // 第一个获取读锁的线程 重入次数等于=1
                // assert firstReaderHoldCount > 0;
                if (firstReaderHoldCount == 1)
                    //第一个获取读锁的线程设置为null
                    firstReader = null;
                else
                    // 当前线程重入多次 -1
                    firstReaderHoldCount--;
                //如果不是第一个获取读锁的线程,获取该线程的锁重入次数对象
            } else {
                // 获取线程持有共享锁的数量对象
                HoldCounter rh = cachedHoldCounter;
                // 如果rh==null 当前线程不是共享锁数量对象对应的线程id
                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;
            }
            //CAS同步更新
            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;
            }
        }

线程读锁的重入数与读锁数量是两个概念,线程读锁的重入数是每个线程获取同一个读锁的次数,读锁数量则是所有线程的读锁重入数总和。举一个例子就是 3个线程 分别获取了3次读锁,那么读锁数量就是9,每个线程的读锁重入数就是3。

锁升级&锁降级

锁升级就是线程持有读锁的前提下,去升级为写锁,显然这是违背读写互斥的。
在这里插入图片描述
锁降级,线程持有写锁的前提下,降级为读锁。
在这里插入图片描述
好了我们来看为什么需要锁降级,如果说针对一块临界区直接加一把大锁,那么其实并发读很低,那么可不可以在获取写锁的前提下 降级为读锁,这样既保证数据的一致性,又可以提升整体的并发度。锁降级就是为了结局这个问题。
在这里插入图片描述

设计思想

通过本篇的大概学习,我们了解到RRW中几个设计要点,通过一个变量去控制两个读写锁的状态,位运算的方式。值得我们借鉴,另一种就是锁降级的为了保证数据安全。以及在整体的代码实现上大量使用模板模式,AQS的子类都是相同的方式。

在这里插入图片描述

文章来源:https://blog.csdn.net/jia970426/article/details/135118986
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。