[Java知识库]JUC-读写锁ReentrantReadWriteLock

1、ReentrantReadWriteLock

当读操作远远高于写操作时，这时候使用读写锁让读-读可以并发，提高性能。类似于数据库中的select ... from ... lock in share mode

提供一个数据容器类分别使用读锁保护数据的read()方法，写锁保护数据的write()方法

public class DataContainer {

    private Object data;
    private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    private ReentrantReadWriteLock.ReadLock r = rw.readLock();
    private ReentrantReadWriteLock.WriteLock w = rw.writeLock();

    public Object read() {
        log.debug("获取读锁");
        r.lock();
        try {
            log.debug("读取");
            sleep(1);
            return data;
        } finally {
            log.debug("释放读锁");
            r.unlock();
        }
    }

    public Object write() {
        log.debug("获取写锁");
        w.lock();
        try {
            log.debug("写入");
            sleep(1);
            return data;
        } finally {
            log.debug("释放写锁");
            w.unlock();
        }
    }

    public void sleep(long second) {
        try {
            Thread.sleep(second * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

测试读锁-读锁可以并发

DataContainer dataContainer = new DataContainer();
        new Thread(() -> {
            dataContainer.read();
        }, "t1").start();

        new Thread(() -> {
            dataContainer.read();
        }, "t2").start();

输出结果，从这里可以看到Thread-0锁定期间，Thread-1的操作不受影响

2022/05/04-14:10:23.505 [t2] c.DataContainer - 获取读锁
2022/05/04-14:10:23.505 [t1] c.DataContainer - 获取读锁
2022/05/04-14:10:23.507 [t1] c.DataContainer - 读取
2022/05/04-14:10:23.507 [t2] c.DataContainer - 读取
2022/05/04-14:10:24.516 [t2] c.DataContainer - 释放读锁
2022/05/04-14:10:24.516 [t1] c.DataContainer - 释放读锁

测试读锁-写锁相互阻塞

DataContainer dataContainer = new DataContainer();
        new Thread(() -> {
            dataContainer.read();
        }, "t1").start();

        new Thread(() -> {
            dataContainer.write();
        }, "t2").start();

输出结果

2022/05/04-14:11:07.301 [t2] c.DataContainer - 获取写锁
2022/05/04-14:11:07.301 [t1] c.DataContainer - 获取读锁
2022/05/04-14:11:07.305 [t2] c.DataContainer - 写入
2022/05/04-14:11:08.315 [t2] c.DataContainer - 释放写锁
2022/05/04-14:11:08.315 [t1] c.DataContainer - 读取
2022/05/04-14:11:09.326 [t1] c.DataContainer - 释放读锁

写锁-写锁也是相互阻塞的，这里就不测试了

注意事项

• 读锁不支持条件变量
• 重入时升级不支持：即持有读锁的情况下去获取写锁，会导致回去写锁永久等待

r.lock();
        try {
            // ...
            w.lock();
            try {
                // ...
            } finally {
                w.unlock();
            }
        } finally {
            r.unlock();
        }

• 重入时降级支持：即持有写锁的情况下去获取读锁

class CachedData {
    Object data;
    // 是否有效，如果失效，需要重新计算 data
    volatile boolean cacheValid;
    final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();

    void processCachedData() {
        rwl.readLock().lock();
        if (!cacheValid) {
            // 获取写锁前必须释放读锁
            rwl.readLock().unlock();
            rwl.writeLock().lock();
            try {
                // 判断是否有其它线程已经获取了写锁、更新了缓存, 避免重复更新
                if (!cacheValid) {
                    data = ...
                    cacheValid = true;
                }
                // 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存
                rwl.readLock().lock();
            } finally {

                rwl.writeLock().unlock();
            }
        }
        // 自己用完数据, 释放读锁
        try {
            use(data);
        } finally {
            rwl.readLock().unlock();
        }
    }
}

2、应用之缓存

1、缓存更新策略

更新时，是先清除缓存还是先更新数据库

先清缓存

先更新数据库

补充一种情况，假设线程A查询数据时恰好缓存数据由于时间到期失效，或是第一次查询

这种情况出现的几率非常小，见facebook论文

2、读写锁实现一致性缓存

class GenericCachedDao<T> {
    // HashMap 作为缓存非线程安全, 需要保护
    HashMap<SqlPair, T> map = new HashMap<>();

    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    GenericDao genericDao = new GenericDao();

    public int update(String sql, Object... params) {
        SqlPair key = new SqlPair(sql, params);
        // 加写锁, 防止其它线程对缓存读取和更改
        lock.writeLock().lock();
        try {
            int rows = genericDao.update(sql, params);
            map.clear();
            return rows;
        } finally {
            lock.writeLock().unlock();
        }
    }

    public T queryOne(Class<T> beanClass, String sql, Object... params) {
        SqlPair key = new SqlPair(sql, params);
        // 加读锁, 防止其它线程对缓存更改
        lock.readLock().lock();
        try {
            T value = map.get(key);
            if (value != null) {
                return value;
            }
        } finally {
            lock.readLock().unlock();
        }
        // 加写锁, 防止其它线程对缓存读取和更改
        lock.writeLock().lock();
        try {
            // get 方法上面部分是可能多个线程进来的, 可能已经向缓存填充了数据
            // 为防止重复查询数据库, 再次验证
            T value = map.get(key);
            if (value == null) {
                // 如果没有, 查询数据库
                value = genericDao.queryOne(beanClass, sql, params);
                map.put(key, value);
            }
            return value;
        } finally {
            lock.writeLock().unlock();
        }
    }
}

// 作为 key 保证其是不可变的
class SqlPair {
    private String sql;
    private Object[] params;

    public SqlPair(String sql, Object[] params) {
        this.sql = sql;
        this.params = params;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        SqlPair sqlPair = (SqlPair) o;
        return sql.equals(sqlPair.sql) &&
                Arrays.equals(params, sqlPair.params);
    }

    @Override
    public int hashCode() {
        int result = Objects.hash(sql);
        result = 31 * result + Arrays.hashCode(params);
        return result;
    }
}

注意：

• 以上实现体现的是读写锁的应用，保证缓存和数据库的一致性，但有下面的问题没有考虑
  • 适合读多写少，如果写操作比较频繁，以上实现性能较低
  • 没有考虑缓存容量
  • 没有考虑缓存过期
  • 只适合单机
  • 并发性还是低，目前只会用一把锁
  • 更新方法过于简单粗暴，清空了所有key（考虑按类型分区或重新设计key）
• 乐观锁实现：用CAS去更新

3、原理

1、图解流程

读写锁用的是同一个Sync同步器，因此等待队列，state等也是同一个

t1 w.lock, t2 r.lock

1）t1成功上锁，流程与ReentrantLock加锁相比没有特殊之处，不同的是写状态占了state的低16位，而读锁使用的是state的高16位

2）t2执行r.lock，这时进入读锁的sync.acquireShared(1)流程，首先会进入tryAcquireShared流程，如果有写锁占用，那么tryAcquireShared返回-1表示失败

tryAcquireShared返回值表示

• -1 表示失败
• 0 表示成功，单后继结点不会继续唤醒
• 正数表示成功，而且数值是还有几个后继结点需要唤醒

3）这时会进入sync.doAcquireShared(1)流程，首先也是调用addWaiter添加结点，不同之处在于结点被设置为Node.SHARED模式而非Node.EXCLUSIVE模式，注意此时t2仍处于活跃状态

4）t2会看看自己的结点是不是老二，如果是，还会再次调用tryAcquireShared(1)来尝试获取锁
5）如果没有成功，在doAcquireShared内for(;;)循环一次，把waitStatus改为-1，再for(;;)循环一次尝试tryAcquireShared(1)，如果还不成功，那么在parkAndCheckInterrupt()处park

t3 r.lock，t4 w.lock

这种状态下，假设又有t3加读锁和t4加写锁，这期间t1仍然持有锁，就就变成了下面的样子

t1 w.unlock
这是会走到写锁的sync.release(1)流程，调用sync.tryRelease(1)成功，变成下面的样子

接下来回字形唤醒流程sync.unparkSuccessor，即让老二恢复运行，这时t2在doAcquireShared内parkAndCheckInterrupt()处恢复运行

这回再来一次for(;;)执行tryAcquireShared成功，则让读锁计数加一

这时t2已经恢复运行，接下来t2调用setHeadAndPropagate(node, 1)，他原本所在结点被设置为头结点

事情还没完，在setHeadAndPropagate方法内还会检查下一个结点是否是shared，如果是则调用
doReleaseShared()将head的状态从-1改为0并唤醒老二，这时t2在doAcquireShared内，parkAndCheckInterrupt()处恢复运行

这回再来一次for(;;)执行tryAcquireShared成功则让读锁计数加一

这时t3已经恢复运行，接下来t3调用setHeadAndPropagate(node, 1)，他原本所在的结点被设置为头结点

下一个结点不是shared了，因此不会继续唤醒t4所在结点

t2 r.unlock， t3 r.unlock
t2进入sync.releaseShared(1)中，调用tryReleaseShared(1)让计数减一，但由于计数还不为零

t3进入sync.releaseShared(1)中，调用tryReleaseShared(1)让计数减一，这回计数为零了，进入doReleaseShared()将头结点从-1改为0，并唤醒老二，即

之后t4在acquireQueued中parkAndCheckInterrupt处恢复运行，再次for(;;)这次自己是老二，并且没有其他竞争，修改头结点，流程结束

源码分析

写锁上锁流程

static final class NonfairSync extends Sync {
    // ... 省略无关代码

    // 外部类 WriteLock 方法, 方便阅读, 放在此处
    public void lock() {
        sync.acquire(1);
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final void acquire(int arg) {
        if (
            // 尝试获得写锁失败
                !tryAcquire(arg) &&
                        // 将当前线程关联到一个 Node 对象上, 模式为独占模式
                        // 进入 AQS 队列阻塞
                        acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
        ) {
            selfInterrupt();
        }
    }

    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryAcquire(int acquires) {
        // 获得低 16 位, 代表写锁的 state 计数
        Thread current = Thread.currentThread();
        int c = getState();
        int w = exclusiveCount(c);

        if (c != 0) {
            if (
                // c != 0 and w == 0 表示有读锁, 或者
                    w == 0 ||
                            // 如果 exclusiveOwnerThread 不是自己
                            current != getExclusiveOwnerThread()
            ) {
                // 获得锁失败
                return false;
            }
            // 写锁计数超过低 16 位, 报异常
            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;
    }

    // 非公平锁 writerShouldBlock 总是返回 false, 无需阻塞
    final boolean writerShouldBlock() {
        return false;
    }
}

写锁释放流程

static final class NonfairSync extends Sync {
    // ... 省略无关代码

    // WriteLock 方法, 方便阅读, 放在此处
    public void unlock() {
        sync.release(1);
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final boolean release(int arg) {
        // 尝试释放写锁成功
        if (tryRelease(arg)) {
            // unpark AQS 中等待的线程
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryRelease(int releases) {
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        int nextc = getState() - releases;
        // 因为可重入的原因, 写锁计数为 0, 才算释放成功
        boolean free = exclusiveCount(nextc) == 0;
        if (free) {
            setExclusiveOwnerThread(null);
        }
        setState(nextc);
        return free;
    }
}

读锁上锁流程

static final class NonfairSync extends Sync {

    // ReadLock 方法, 方便阅读, 放在此处
    public void lock() {
        sync.acquireShared(1);
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final void acquireShared(int arg) {
        // tryAcquireShared 返回负数, 表示获取读锁失败
        if (tryAcquireShared(arg) < 0) {
            doAcquireShared(arg);
        }
    }

    // Sync 继承过来的方法, 方便阅读, 放在此处
    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 &&
                        // 尝试增加计数成功
                        compareAndSetState(c, c + SHARED_UNIT)
        ) {
            // ... 省略不重要的代码
            return 1;
        }
        return fullTryAcquireShared(current);
    }

    // 非公平锁 readerShouldBlock 看 AQS 队列中第一个节点是否是写锁
    // true 则该阻塞, false 则不阻塞
    final boolean readerShouldBlock() {
        return apparentlyFirstQueuedIsExclusive();
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    // 与 tryAcquireShared 功能类似, 但会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞
    final int fullTryAcquireShared(Thread current) {
        HoldCounter rh = null;
        for (; ; ) {
            int c = getState();
            if (exclusiveCount(c) != 0) {
                if (getExclusiveOwnerThread() != current)
                    return -1;
            } else if (readerShouldBlock()) {
                // ... 省略不重要的代码
            }
            if (sharedCount(c) == MAX_COUNT)
                throw new Error("Maximum lock count exceeded");
            if (compareAndSetState(c, c + SHARED_UNIT)) {
                // ... 省略不重要的代码
                return 1;
            }
        }
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    private void doAcquireShared(int arg) {
        // 将当前线程关联到一个 Node 对象上, 模式为共享模式
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (; ; ) {
                final Node p = node.predecessor();
                if (p == head) {
// 再一次尝试获取读锁
                    int r = tryAcquireShared(arg);
                    // 成功
                    if (r >= 0) {
                        // ㈠
// r 表示可用资源数, 在这里总是 1 允许传播
                        //（唤醒 AQS 中下一个 Share 节点）
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (
                    // 是否在获取读锁失败时阻塞（前一个阶段 waitStatus == Node.SIGNAL）
                        shouldParkAfterFailedAcquire(p, node) &&
                                // park 当前线程
                                parkAndCheckInterrupt()
                ) {
                    interrupted = true;
                }
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

    // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        // 设置自己为 head
        setHead(node);

        // propagate 表示有共享资源（例如共享读锁或信号量）
        // 原 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
        // 现在 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            // 如果是最后一个节点或者是等待共享读锁的节点
            if (s == null || s.isShared()) {
                // 进入 ㈡
                doReleaseShared();
            }
        }
    }

    // ㈡ AQS 继承过来的方法, 方便阅读, 放在此处
    private void doReleaseShared() {
        // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
// 如果 head.waitStatus == 0 ==> Node.PROPAGATE, 为了解决 bug, 见后面分析
        for (; ; ) {
            Node h = head;
            // 队列还有节点
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue; // loop to recheck cases
                    // 下一个节点 unpark 如果成功获取读锁
                    // 并且下下个节点还是 shared, 继续 doReleaseShared
                    unparkSuccessor(h);
                } else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue; // loop on failed CAS
            }
            if (h == head) // loop if head changed
                break;
        }
    }
}

读锁释放流程

static final class NonfairSync extends Sync {

    // ReadLock 方法, 方便阅读, 放在此处
    public void unlock() {
        sync.releaseShared(1);
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

    // Sync 继承过来的方法, 方便阅读, 放在此处
    protected final boolean tryReleaseShared(int unused) {
        // ... 省略不重要的代码
        for (; ; ) {
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc)) {
                // 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程
                // 计数为 0 才是真正释放
                return nextc == 0;
            }
        }
    }

    // AQS 继承过来的方法, 方便阅读, 放在此处
    private void doReleaseShared() {
        // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
        // 如果 head.waitStatus == 0 ==> Node.PROPAGATE 
        for (; ; ) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                // 如果有其它线程也在释放读锁，那么需要将 waitStatus 先改为 0
                // 防止 unparkSuccessor 被多次执行
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue; // loop to recheck cases
                    unparkSuccessor(h);
                }
                // 如果已经是 0 了，改为 -3，用来解决传播性，见后文信号量 bug 分析
                else if (ws == 0 &&
                        !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue; // loop on failed CAS
            }
            if (h == head) // loop if head changed
                break;
        }
    }
}

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖