10 读写锁ReentrantReadWriteLock

1 介绍

为什么要使用读写锁？

        需要高并发读取和较低并发写入的应用程序，降低锁的粒度，提高系统性能

使用场景：

        读多写少的共享资源

        缓存管理：读 >> 写，控制多个线程同时读缓存，需要刷新or修改操作时才使用写锁

        数据库连接池：多个线程从池中获取连接（读操作），只有一个线程可以设置连接到池中（写操作）

        文件读写

        数据结构的并发访问

2 使用

import java.util.concurrent.locks.ReentrantReadWriteLock;
 
public class SharedResource {
    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
    private final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
 
    public void readFromResource() {
        readLock.lock(); // 获取读锁
        try {
            // 执行读取共享资源的操作
        } finally {
            readLock.unlock(); // 释放读锁
        }
    }
 
    public void writeToResource() {
        writeLock.lock(); // 获取写锁
        try {
            // 执行写入共享资源的操作
        } finally {
            writeLock.unlock(); // 释放写锁
        }
    }
}

3 原理分析

读写锁两个状态，读状态、写状态

但AQS中只有一个state，如何记录两种状态？

        高低位；int4个字节，共32位，采用高16位控制读，低16位控制写

00000000 00000000 00000000 00000000

加锁时如何判断读锁、写锁？

        高16位>0，表示有读锁（sharedCount()）

        低16位>0，表示有写锁（exclusiveCount()）

如何实现可重入？

        写锁只有一个线程独占，重入则低16位+1即可

        写锁有多个线程持有，如何记录？ThreadLocal线程私有

4 读锁源码

读锁：tryAcquireShared()、tryReleaseShared()；读读共享

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);
            //获取共享锁
            //1 未阻塞
            //2 不超最大读计数
            //3 设置共享锁成功
            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;
                    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);
        }

final int fullTryAcquireShared(Thread current) {
            HoldCounter rh = null;
            //1 自旋 获取共享锁or失败
            for (;;) {
                int c = getState();
                //2 若存在写锁，且不是当前线程持有的，不允许获取共享锁
                if (exclusiveCount(c) != 0) {
                    if (getExclusiveOwnerThread() != current)
                        return -1;
                //3 读线程阻塞
                } else if (readerShouldBlock()) {
                    if (firstReader == current) {
                        
                    } else {
                        if (rh == null) {
                            rh = cachedHoldCounter;
                            if (rh == null || rh.tid != getThreadId(current)) {
                                rh = readHolds.get();
                                if (rh.count == 0)
                                    readHolds.remove();
                            }
                        }
                        if (rh.count == 0)
                            return -1;
                    }
                }
                // 4 不允许再申请共享锁
                if (sharedCount(c) == MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                // 5 尝试获取锁
                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;
                }
            }
        }

5 写锁源码

写锁：tryAcquire()、tryRelease()；写写互斥，读写互斥

public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}
 
protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            int c = getState();
            //获取低16位（写锁）的值，sharedCount()方法获取高16位值
            int w = exclusiveCount(c);
            //代表有加锁
            if (c != 0) {
                //1 是读锁，读写互斥，不能加写锁
                //2 是写锁，但不是设置重入的写锁，线程之间写写互斥
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                //重入写锁
                setState(c + acquires);
                return true;
            }
            //公平锁需要排队，非公平直接让当前线程尝试获取锁
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            setExclusiveOwnerThread(current);
            return true;
        }

6 存在问题

        读的过程不允许写，悲观锁，可能会造成锁饥饿

        为了进一步提升并发执行效率，Java 8引入了新的读写锁：StampedLock。StampedLock和ReentrantReadWriteLock相比，改进之处在于：读的过程中也允许获取写锁后写入！在原先读写锁的基础上新增了一种叫乐观读（Optimistic Reading）的模式。该模式并不会加锁，所以不会阻塞线程，会有更高的吞吐量和更高的性能