聊聊高并发(二十九)解析java.util.concurrent各个组件(十一) 再看看ReentrantReadWriteLock可重入读-写锁

上一篇聊聊高并发(二十八)解析java.util.concurrent各个组件(十) 理解ReentrantReadWriteLock可重入读-写锁 讲了可重入读写锁的基本情况和主要的方法,显示了如何实现的锁降级。但是下面几个问题没说清楚,这篇补充一下

1. 释放锁时的优先级问题,是让写锁先获得还是先让读锁先获得

2. 是否允许读线程插队

3. 是否允许写线程插队,因为读写锁一般用在大量读,少量写的情况,如果写线程没有优先级,那么可能造成写线程的饥饿

关于释放锁后是让写锁先获得还是让读锁先获得,这里有两种情况

1. 释放锁后,请求获取写锁的线程不在AQS队列

2. 释放锁后,请求获取写锁的线程已经AQS队列

如果是第一种情况,那么非公平锁的实现下,获取写锁的线程直接尝试竞争锁也不用管AQS里面先来的线程。获取读锁的线程只判断是否已经有线程获得写锁(既Head节点是独占模式的节点),如果没有,那么就不用管AQS里面先来的准备获取读锁的线程。

static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -8159625535654395037L;
        final boolean writerShouldBlock() {
            return false; // writers can always barge
        }
        final boolean readerShouldBlock() {
            return apparentlyFirstQueuedIsExclusive();
        }
    }

在公平锁的情况下,获取读锁和写锁的线程都判断是否已经或先来的线程再等待了,如果有,就进入AQS队列等待。

static final class FairSync extends Sync {
        private static final long serialVersionUID = -2274990926593161451L;
        final boolean writerShouldBlock() {
            return hasQueuedPredecessors();
        }
        final boolean readerShouldBlock() {
            return hasQueuedPredecessors();
        }
    }

对于第二种情况,如果准备获取写锁的线程在AQS队列里面等待,那么实际是遵循先来先服务的公平性的,因为AQS的队列是FIFO的队列。所以获取锁的线程的顺序是跟它在AQS同步队列里的位置有关系。

下面这张图模拟了AQS队列中等待的线程节点的情况

1. Head节点始终是当前获得了锁的线程

2. 非Head节点在竞争锁失败后,acquire方法会不断地轮询,于自旋不同的是,AQS轮询过程中的线程是阻塞等待。

所以要理解AQS的release释放动作并不是让后续节点直接获取锁,而是唤醒后续节点unparkSuccessor()。真正获取锁的地方还是在acquire方法,被release唤醒的线程继续轮询状态,如果它的前驱是head,并且tryAcquire获取资源成功了,那么它就获得锁

public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

3. 图中Head之后有3个准备获取读锁的线程,最后是1个准备获取写锁的线程。

那么如果是AQS队列中的节点获取锁

情况是第一个读锁节点先获得锁,它获取锁的时候就会尝试释放共享模式下的一个读锁,如果释放成功了,下一个读锁节点就也会被unparkSuccessor唤醒,然后也会获得锁。

如果释放失败了,那就把它的状态标记了PROPAGATE,当它释放的时候,会再次取尝试唤醒下一个读锁节点

如果后继节点是写锁,那么就不唤醒

private void doAcquireShared(int arg) {
        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) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

 private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
       
        if (propagate > 0 || h == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }

private void doReleaseShared() {
        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
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

AQS的FIFO队列保证了在大量读锁和少量写锁的情况下,写锁也不会饥饿。

关于读锁能不能插队的问题,非公平性的Sync提供了插队的可能,但是前提是它在tryAcquire就成功获得了,如果tryAcquire失败了,它就得进入AQS队列排队,也不会出现让写锁饥饿的情况。

关于写锁能不能插队的情况,也是和读锁一样,非公平的Sync提供了插队的可能,如果tryAcquire获取失败,就得进入AQS等待。

最后说说为什么Semaphore和ReentrantLock在tryAcquireXX方法就实现了非公平性和公平性,而ReentrantReadWriteLock却要抽象出readerShouldBlock和writerShouldBlock的方法来单独处理公平性。

abstract boolean readerShouldBlock();
abstract boolean writerShouldBlock();

原因是Semaphore只支持共享模式,所以它只需要在NonfairSync和FairSync里面实现tryAcquireShared方法就能实现公平性和非公平性。

ReentrantLock只支持独占模式,所以它只需要在NonfairSync和FairSync里面实现tryAcquire方法就能实现公平性和非公平性。

而ReentrantReadWriteLock即要支持共享和独占模式,又要支持公平性和非公平性,所以它在基类的Sync里面用tryAcquire和tryAcquireShared方法来区分独占和共享模式,

在NonfairSync和FairSync的readerShouldBlock和writerShouldBlock里面实现非公平性和公平性。

时间: 2025-01-01 06:26:00

聊聊高并发(二十九)解析java.util.concurrent各个组件(十一) 再看看ReentrantReadWriteLock可重入读-写锁的相关文章

聊聊高并发(二十)解析java.util.concurrent各个组件(二) 12个原子变量相关类

这篇说说java.util.concurrent.atomic包里的类,总共12个.网上有非常多文章解析这几个类.这里挑些重点说说. 这12个类能够分为三组: 1. 普通类型的原子变量 2. 数组类型的原子变量 3. 域更新器 普通类型的原子变量的6个, 1. 当中AtomicBoolean, AtomicInteger, AtomicLong, AtomicReference分别相应boolean, int,  long, object完毕主要的原子操作 2. AtomicMarkableRe

聊聊高并发(二十七)解析java.util.concurrent各个组件(九) 理解ReentrantLock可重入锁

这篇讲讲ReentrantLock可重入锁,JUC里提供的可重入锁是基于AQS实现的阻塞式可重入锁.这篇 聊聊高并发(十六)实现一个简单的可重入锁 模拟了可重入锁的实现.可重入锁的特点是: 1. 是互斥锁,基于AQS的互斥模式实现,也就是说同时只有一个线程进入临界区,唤醒下一个线程时也只能释放一个等待线程 2. 可重入,通过设置了一个字段exclusiveOwnerThread来标示当前获得锁的线程.获取锁操作是,如果当前线程是已经获得锁的线程,那么获取操作成功.把当前状态作为获得锁次数的计数器

聊聊高并发(三十)解析java.util.concurrent各个组件(十二) 理解CyclicBarrier栅栏

这篇讲讲CyclicBarrier栅栏,从它的名字可以看出,它是可循环使用的.它的功能和CountDownLatch类似,也是让一组线程等待,然后一起开始往下执行.但是两者还是有几个区别 1. 等待的对象不同.CountDownLatch的一组线程等待的是一个事件,或者说是一个计数器归0的事件.而CyclicBarrier等待的对象是线程,只有线程都到齐了才往下执行 2. 使用方式不同,这个也是由等待的对象不同引起的,CountDownLatch需要调用await()来让线程等待,调用count

谈论高并发(三十)解析java.util.concurrent各种组件(十二) 认识CyclicBarrier栅栏

这次谈话CyclicBarrier栅栏,如可以从它的名字可以看出,它是可重复使用. 它的功能和CountDownLatch类别似,也让一组线程等待,然后开始往下跑起来.但也有在两者之间有一些差别 1. 不同的对象等.CountDownLatch组线程等待的是一个事件.或者说是一个计数器归0的事件.而CyclicBarrier等待的对象是线程,仅仅有线程都到齐了才往下运行 2. 使用方式不同,这个也是由等待的对象不同引起的,CountDownLatch须要调用await()来让线程等待.调用cou

聊聊高并发(四十)解析java.util.concurrent各个组件(十六) ThreadPoolExecutor源代码分析

ThreadPoolExecutor是Executor运行框架最重要的一个实现类.提供了线程池管理和任务管理是两个最主要的能力.这篇通过分析ThreadPoolExecutor的源代码来看看怎样设计和实现一个基于生产者消费者模型的运行器. 生产者消费者模型 生产者消费者模型包括三个角色:生产者,工作队列,消费者.对于ThreadPoolExecutor来说, 1. 生产者是任务的提交者,是外部调用ThreadPoolExecutor的线程 2. 工作队列是一个堵塞队列的接口,详细的实现类能够有非

聊聊高并发(四十一)解析java.util.concurrent各个组件(十七) 任务的异步执行和状态控制

聊聊高并发(三十九)解析java.util.concurrent各个组件(十五) 理解ExecutorService接口的设计这篇说了ExecutorService接口扩展了Executor接口,在执行任务的基础上,提供了执行框架生命周期的管理,任务的异步执行,批量任务的执行的能力.AbstractExecutorService抽象类实现了ExecutorService接口,提供了任务异步执行和批量执行的默认实现.这篇说说任务的异步执行和状态控制 说明一点,使用Executor框架执行任务的方式

聊聊高并发(二十五)解析java.util.concurrent各个组件(七) 理解Semaphore

前几篇分析了一下AQS的原理和实现.这篇拿Semaphore信号量做样例看看AQS实际是怎样使用的. Semaphore表示了一种能够同一时候有多个线程进入临界区的同步器,它维护了一个状态表示可用的票据,仅仅有拿到了票据的线程尽能够进入临界区,否则就等待.直到获得释放出的票据. Semaphore经常使用在资源池中来管理资源.当状态仅仅有1个0两个值时,它退化成了一个相互排斥的同步器.类似锁. 以下来看看Semaphore的代码. 它维护了一个内部类Sync来继承AQS,定制tryXXX方法来使

聊聊高并发(二十六)解析java.util.concurrent各个组件(八) 理解CountDownLatch闭锁

CountDownLatch闭锁也是基于AQS实现的一种同步器,它表示了"所有线程都等待,直到锁打开才继续执行"的含义.它和Semaphore的语意不同, Semaphore的获取和释放操作都会修改状态,都可能让自己或者其他线程立刻拿到锁.而闭锁的获取操作只判断状态是否为0,不修改状态本身,闭锁的释放操作会修改状态,每次递减1,直到状态为0. 所以正常情况下,闭锁的获取操作只是等待,不会立刻让自己获得锁,直到释放操作把状态变为0. 闭锁可以用来实现很多场景,比如: 1. 某个服务依赖于

聊聊高并发(三十九)解析java.util.concurrent各个组件(十五) 理解ExecutorService接口的设计

上一篇讲了Executor接口的设计,目的是将任务的运行和任务的提交解耦.能够隐藏任务的运行策略.这篇说说ExecutorService接口.它扩展了Executor接口,对Executor的生命周期进行管理.并进行了进一步的扩展. Executor负责执行任务. 它的生命周期有3个:执行,关闭和已终止. 在执行的不论什么时刻,有些 任务可能已经完毕,有些可能正在执行,有些可能正在队列中等待执行.所以假设要关闭Executor的话.就有多种方式,比方优雅平滑的关闭,当执行关闭时就不在接受新的任务