【Java并发编程】16、ReentrantReadWriteLock源码分析

一、前言

　　在分析了锁框架的其他类之后，下面进入锁框架中最后一个类ReentrantReadWriteLock的分析，它表示可重入读写锁，ReentrantReadWriteLock中包含了两种锁，读锁ReadLock和写锁WriteLock，可以通过这两种锁实现线程间的同步，下面开始进行分析。

二、ReentrantReadWriteLock数据结构

　　分析源码可以知道，ReentrantReadWriteLock底层是基于ReentrantLock和AbstractQueuedSynchronizer来实现的，所以，ReentrantReadWriteLock的数据结构也依托于AQS的数据结构，在前面对AQS的分析中已经指出了其数据结构，在这里不再累赘。

三、ReentrantReadWriteLock源码分析

　　3.1. 类的继承关系　　

public class ReentrantReadWriteLockimplements ReadWriteLock, java.io.Serializable {}

　　说明：可以看到，ReentrantReadWriteLock实现了ReadWriteLock接口，ReadWriteLock接口定义了获取读锁和写锁的规范，具体需要实现类去实现；同时其还实现了Serializable接口，表示可以进行序列化，在源代码中可以看到ReentrantReadWriteLock实现了自己的序列化逻辑。

　　3.2. 类的内部类

　　ReentrantReadWriteLock有五个内部类，五个内部类之间也是相互关联的。内部类的关系如下图所示。

　　说明：如上图所示，Sync继承自AQS、NonfairSync继承自Sync类、FairSync继承自Sync类；ReadLock实现了Lock接口、WriteLock也实现了Lock接口。

　　① Sync类

　　1. 类的继承关系　

abstract static class Sync extends AbstractQueuedSynchronizer {}

　　说明：Sync抽象类继承自AQS抽象类，Sync类提供了对ReentrantReadWriteLock的支持。

　　2. 类的内部类

　　Sync类内部存在两个内部类，分别为HoldCounter和ThreadLocalHoldCounter，其中HoldCounter主要与读锁配套使用，其中，HoldCounter源码如下。

        // 计数器static final class HoldCounter {// 计数int count = 0;// Use id, not reference, to avoid garbage retention// 获取当前线程的TID属性的值final long tid = getThreadId(Thread.currentThread());}

说明：HoldCounter主要有两个属性，count和tid，其中count表示某个读线程重入的次数，tid表示该线程的tid字段的值，该字段可以用来唯一标识一个线程。ThreadLocalHoldCounter的源码如下

        // 本地线程计数器static final class ThreadLocalHoldCounterextends ThreadLocal<HoldCounter> {// 重写初始化方法，在没有进行set的情况下，获取的都是该HoldCounter值public HoldCounter initialValue() {return new HoldCounter();}}

说明：ThreadLocalHoldCounter重写了ThreadLocal的initialValue方法，ThreadLocal类可以将线程与对象相关联。在没有进行set的情况下，get到的均是initialValue方法里面生成的那个HolderCounter对象。

abstract static class Sync extends AbstractQueuedSynchronizer {// 版本序列号private static final long serialVersionUID = 6317671515068378041L;        // 高16位为读锁，低16位为写锁static final int SHARED_SHIFT   = 16;// 读锁单位static final int SHARED_UNIT    = (1 << SHARED_SHIFT);// 读锁最大数量static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;// 写锁最大数量static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;// 本地线程计数器private transient ThreadLocalHoldCounter readHolds;// 缓存的计数器private transient HoldCounter cachedHoldCounter;// 第一个读线程private transient Thread firstReader = null;// 第一个读线程的计数private transient int firstReaderHoldCount;}

说明：该属性中包括了读锁、写锁线程的最大量。本地线程计数器等。

4. 类的构造函数

// 构造函数
        Sync() {// 本地线程计数器readHolds = new ThreadLocalHoldCounter();// 设置AQS的状态setState(getState()); // ensures visibility of readHolds}

说明：在Sync的构造函数中设置了本地线程计数器和AQS的状态state。

　　5. 核心函数分析

　　对ReentrantReadWriteLock对象的操作绝大多数都转发至Sync对象进行处理。下面对Sync类中的重点函数进行分析

　　I. sharedCount函数

　　表示占有读锁的线程数量，源码如下　

static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }

　　说明：直接将state右移16位，就可以得到读锁的线程数量，因为state的高16位表示读锁，对应的第十六位表示写锁数量。

　　II. exclusiveCount函数

　　表示占有写锁的线程数量，源码如下　　

static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

　　说明：直接将状态state和（2^16 - 1）做与运算，其等效于将state模上2^16。写锁数量由state的低十六位表示。

　　III. tryRelease函数　　

protected final boolean tryRelease(int releases) {if (!isHeldExclusively())throw new IllegalMonitorStateException();int nextc = getState() - releases;boolean free = exclusiveCount(nextc) == 0;if (free)setExclusiveOwnerThread(null);setState(nextc);return free;}

　说明：此函数用于释放写锁资源，首先会判断该线程是否为独占线程，若不为独占线程，则抛出异常，否则，计算释放资源后的写锁的数量，若为0，表示成功释放，资源不将被占用，否则，表示资源还被占用。其函数流程图如下。

　　IV. tryAcquire函数　　

protected final boolean tryAcquire(int acquires) {// 获取当前线程Thread current = Thread.currentThread();// 获取状态int c = getState();// 写线程数量int w = exclusiveCount(c);if (c != 0) { // 状态不为0// (Note: if c != 0 and w == 0 then shared count != 0)if (w == 0 || current != getExclusiveOwnerThread()) // 写线程数量为0或者当前线程没有占有独占资源return false;if (w + exclusiveCount(acquires) > MAX_COUNT) // 判断是否超过最高写线程数量throw new Error("Maximum lock count exceeded");// Reentrant acquire// 设置AQS状态setState(c + acquires);return true;}if (writerShouldBlock() ||!compareAndSetState(c, c + acquires)) // 写线程是否应该被阻塞return false;// 设置独占线程
            setExclusiveOwnerThread(current);return true;}

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

V. tryReleaseShared函数　

 protected final boolean tryReleaseShared(int unused) {// 获取当前线程Thread current = Thread.currentThread();if (firstReader == current) { // 当前线程为第一个读线程// assert firstReaderHoldCount > 0;if (firstReaderHoldCount == 1) // 读线程占用的资源数为1firstReader = null;else // 减少占用的资源firstReaderHoldCount--;} else { // 当前线程不为第一个读线程// 获取缓存的计数器HoldCounter rh = cachedHoldCounter;if (rh == null || rh.tid != getThreadId(current)) // 计数器为空或者计数器的tid不为当前正在运行的线程的tid// 获取当前线程对应的计数器rh = readHolds.get();// 获取计数int count = rh.count;if (count <= 1) { // 计数小于等于1// 移除
                    readHolds.remove();if (count <= 0) // 计数小于等于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;}}

说明：此函数表示读锁线程释放锁。首先判断当前线程是否为第一个读线程firstReader，若是，则判断第一个读线程占有的资源数firstReaderHoldCount是否为1，若是，则设置第一个读线程firstReader为空，否则，将第一个读线程占有的资源数firstReaderHoldCount减1；若当前线程不是第一个读线程，那么首先会获取缓存计数器（上一个读锁线程对应的计数器），若计数器为空或者tid不等于当前线程的tid值，则获取当前线程的计数器，如果计数器的计数count小于等于1，则移除当前线程对应的计数器，如果计数器的计数count小于等于0，则抛出异常，之后再减少计数即可。无论何种情况，都会进入无限循环，该循环可以确保成功设置状态state。其流程图如下

VI. tryAcquireShared函数　

private IllegalMonitorStateException unmatchedUnlockException() {return new IllegalMonitorStateException("attempt to unlock read lock, not locked by current thread");}// 共享模式下获取资源protected final int tryAcquireShared(int unused) {// 获取当前线程Thread current = Thread.currentThread();// 获取状态int c = getState();if (exclusiveCount(c) != 0 &&getExclusiveOwnerThread() != current) // 写线程数不为0并且占有资源的不是当前线程return -1;// 读锁数量int r = sharedCount(c);if (!readerShouldBlock() &&r < MAX_COUNT &&compareAndSetState(c, c + SHARED_UNIT)) { // 读线程是否应该被阻塞、并且小于最大值、并且比较设置成功if (r == 0) { // 读锁数量为0// 设置第一个读线程firstReader = current;// 读线程占用的资源数为1firstReaderHoldCount = 1;} else if (firstReader == current) { // 当前线程为第一个读线程// 占用资源数加1firstReaderHoldCount++;} else { // 读锁数量不为0并且不为当前线程// 获取计数器HoldCounter rh = cachedHoldCounter;if (rh == null || rh.tid != getThreadId(current)) // 计数器为空或者计数器的tid不为当前正在运行的线程的tid// 获取当前线程对应的计数器cachedHoldCounter = rh = readHolds.get();else if (rh.count == 0) // 计数为0// 设置
                        readHolds.set(rh);rh.count++;}return 1;}return fullTryAcquireShared(current);}

说明：此函数表示读锁线程获取读锁。首先判断写锁是否为0并且当前线程不占有独占锁，直接返回；否则，判断读线程是否需要被阻塞并且读锁数量是否小于最大值并且比较设置状态成功，若当前没有读锁，则设置第一个读线程firstReader和firstReaderHoldCount；若当前线程线程为第一个读线程，则增加firstReaderHoldCount；否则，将设置当前线程对应的HoldCounter对象的值。流程图如下。

VII. fullTryAcquireShared函数　

final int fullTryAcquireShared(Thread current) {HoldCounter rh = null;for (;;) { // 无限循环// 获取状态int c = getState();if (exclusiveCount(c) != 0) { // 写线程数量不为0if (getExclusiveOwnerThread() != current) // 不为当前线程return -1;// else we hold the exclusive lock; blocking here// would cause deadlock.} else if (readerShouldBlock()) { // 写线程数量为0并且读线程被阻塞// Make sure we're not acquiring read lock reentrantlyif (firstReader == current) { // 当前线程为第一个读线程// assert firstReaderHoldCount > 0;} else { // 当前线程不为第一个读线程if (rh == null) { // 计数器不为空// rh = cachedHoldCounter;if (rh == null || rh.tid != getThreadId(current)) { // 计数器为空或者计数器的tid不为当前正在运行的线程的tidrh = 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) { // 读线程数量为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;}}}

说明：在tryAcquireShared函数中，如果下列三个条件不满足（读线程是否应该被阻塞、小于最大值、比较设置成功）则会进行fullTryAcquireShared函数中，它用来保证相关操作可以成功。其逻辑与tryAcquireShared逻辑类似，不再累赘。

　　而其他内部类的操作基本上都是转化到了对Sync对象的操作，在此不再累赘。

　　3.3. 类的属性　　

public class ReentrantReadWriteLockimplements ReadWriteLock, java.io.Serializable {// 版本序列号    private static final long serialVersionUID = -6992448646407690164L;    // 读锁private final ReentrantReadWriteLock.ReadLock readerLock;// 写锁private final ReentrantReadWriteLock.WriteLock writerLock;// 同步队列final Sync sync;private static final sun.misc.Unsafe UNSAFE;// 线程ID的偏移地址private static final long TID_OFFSET;static {try {UNSAFE = sun.misc.Unsafe.getUnsafe();Class<?> tk = Thread.class;// 获取线程的tid字段的内存地址TID_OFFSET = UNSAFE.objectFieldOffset(tk.getDeclaredField("tid"));} catch (Exception e) {throw new Error(e);}}
}

说明：可以看到ReentrantReadWriteLock属性包括了一个ReentrantReadWriteLock.ReadLock对象，表示读锁；一个ReentrantReadWriteLock.WriteLock对象，表示写锁；一个Sync对象，表示同步队列。

　　3.4. 类的构造函数

　　1. ReentrantReadWriteLock()型构造函数　　

    public ReentrantReadWriteLock() {this(false);}

　说明：此构造函数会调用另外一个有参构造函数。

　　2. ReentrantReadWriteLock(boolean)型构造函数　

 public ReentrantReadWriteLock(boolean fair) {// 公平策略或者是非公平策略sync = fair ? new FairSync() : new NonfairSync();// 读锁readerLock = new ReadLock(this);// 写锁writerLock = new WriteLock(this);}

说明：可以指定设置公平策略或者非公平策略，并且该构造函数中生成了读锁与写锁两个对象。

　　3.5 核心函数分析

　　对ReentrantReadWriteLock的操作基本上都转化为了对Sync对象的操作，而Sync的函数已经分析过，不再累赘。

四、示例

　　下面给出了一个使用ReentrantReadWriteLock的示例，源代码如下。

package com.hust.grid.leesf.reentrantreadwritelock;import java.util.concurrent.locks.ReentrantReadWriteLock;class ReadThread extends Thread {private ReentrantReadWriteLock rrwLock;public ReadThread(String name, ReentrantReadWriteLock rrwLock) {super(name);this.rrwLock = rrwLock;}public void run() {System.out.println(Thread.currentThread().getName() + " trying to lock");try {rrwLock.readLock().lock();System.out.println(Thread.currentThread().getName() + " lock successfully");Thread.sleep(5000);        } catch (InterruptedException e) {e.printStackTrace();} finally {rrwLock.readLock().unlock();System.out.println(Thread.currentThread().getName() + " unlock successfully");}}
}class WriteThread extends Thread {private ReentrantReadWriteLock rrwLock;public WriteThread(String name, ReentrantReadWriteLock rrwLock) {super(name);this.rrwLock = rrwLock;}public void run() {System.out.println(Thread.currentThread().getName() + " trying to lock");try {rrwLock.writeLock().lock();System.out.println(Thread.currentThread().getName() + " lock successfully");    } finally {rrwLock.writeLock().unlock();System.out.println(Thread.currentThread().getName() + " unlock successfully");}}
}public class ReentrantReadWriteLockDemo {public static void main(String[] args) {ReentrantReadWriteLock rrwLock = new ReentrantReadWriteLock();ReadThread rt1 = new ReadThread("rt1", rrwLock);ReadThread rt2 = new ReadThread("rt2", rrwLock);WriteThread wt1 = new WriteThread("wt1", rrwLock);rt1.start();rt2.start();wt1.start();}
}

运行结果（某一次）：

rt1 trying to lock
rt2 trying to lock
wt1 trying to lock
rt1 lock successfully
rt2 lock successfully
rt1 unlock successfully
rt2 unlock successfully
wt1 lock successfully
wt1 unlock successfully

说明：程序中生成了一个ReentrantReadWriteLock对象，并且设置了两个读线程，一个写线程。根据结果，可能存在如下的时序图。

① rt1线程执行rrwLock.readLock().lock操作，主要的函数调用如下。

说明：此时，AQS的状态state为2^16 次方，即表示此时读线程数量为1。

　　② rt2线程执行rrwLock.readLock().lock操作，主要的函数调用如下。

说明：此时，AQS的状态state为2 * 2^16次方，即表示此时读线程数量为2。

　　③ wt1线程执行rrwLock.writeLock().lock操作，主要的函数调用如下。

　　说明：此时，在同步队列Sync queue中存在两个结点，并且wt1线程会被禁止运行。

　　④ rt1线程执行rrwLock.readLock().unlock操作，主要的函数调用如下。

说明：此时，AQS的state为2^16次方，表示还有一个读线程。

　　⑤ rt2线程执行rrwLock.readLock().unlock操作，主要的函数调用如下。

说明：当rt2线程执行unlock操作后，AQS的state为0，并且wt1线程将会被unpark，其获得CPU资源就可以运行。

　　⑥ wt1线程获得CPU资源，继续运行，需要恢复。由于之前acquireQueued函数中的parkAndCheckInterrupt函数中被禁止的，所以，恢复到parkAndCheckInterrupt函数中，主要的函数调用如下

　　说明：最后，sync queue队列中只有一个结点，并且头结点尾节点均指向它，AQS的state值为1，表示此时有一个写线程。

　　⑦ wt1执行rrwLock.writeLock().unlock操作，主要的函数调用如下。

　　说明：此时，AQS的state为0，表示没有任何读线程或者写线程了。并且Sync queue结构与上一个状态的结构相同，没有变化。

出处：https://www.cnblogs.com/leesf456/p/5419132.html

转载于:https://www.cnblogs.com/wangzhongqiu/p/8422925.html

【Java并发编程】16、ReentrantReadWriteLock源码分析相关推荐

Java并发编程：从源码分析几道必问线程池的面试题？
引言上一篇文章我们有介绍过线程池的一个基本执行流程<[Java并发编程]面试必备之线程池>以及它的7个核心参数,以及每个参数的作用.以及如何去使用线程池还留了几个小问题..建议看这篇文 ...
并发编程之——读锁源码分析(解释关于锁降级的争议)
1. 前言在前面的文章并发编程之--写锁源码分析中,我们分析了 1.8 JUC 中读写锁中的写锁的获取和释放过程,今天来分析一下读锁的获取和释放过程,读锁相比较写锁要稍微复杂一点,其中还有一点有争 ...
Java 并发编程 -- 线程池源码实战
一.概述小编在网上看了好多的关于线程池原理.源码分析相关的文章,但是说实话,没有一篇让我觉得读完之后豁然开朗,完完全全的明白线程池,要么写的太简单,只写了一点皮毛,要么就是是晦涩难懂,看完之后几乎都 ...
并发编程之——写锁源码分析
1.前言 Java 中的读写锁实现是 ReentrantReadWriteLock ,是一种锁分离策略.能有效提高读比写多的场景下的程序性能. 关于如何使用参见并发编程之 Java 三把锁. 由于读 ...
多线程高并发编程(10) -- ConcurrentHashMap源码分析
一.背景前文讲了HashMap的源码分析,从中可以看到下面的问题: HashMap的put/remove方法不是线程安全的,如果在多线程并发环境下,使用synchronized进行加锁,会导致效率低 ...
Java并发编程之ThreadLocal源码分析
1 一句话概括ThreadLocal 什么是ThreadLocal?顾名思义:线程本地变量,它为每个使用该对象的线程创建了一个独立的变量副本. 2 ThreadLocal使用场景用一句话总结 ...
libevent c++高并发网络编程_【多线程高并发编程】Callable源码分析
程序猿学社的GitHub,欢迎Starhttps://github.com/ITfqyd/cxyxs 本文已记录到github,形成对应专题. 前言通过上一章实现多线程有几种方式,我们已经了解多线程 ...
【java并发编程实践】源码
作者:(美)Brian Goetz,Tim Peierls 等著译者:童云兰等译 https://github.com/DoingLee/jcip-examples-src
Java并发-ReentrantReadWriteLock源码分析
ReentrantLock实现了标准的互斥重入锁,任一时刻只有一个线程能获得锁.考虑这样一个场景:大部分时间都是读操作,写操作很少发生:我们知道,读操作是不会修改共享数据的,如果实现互斥锁,那么即使都 ...

【Java并发编程】16、ReentrantReadWriteLock源码分析

【Java并发编程】16、ReentrantReadWriteLock源码分析相关推荐

最新文章

热门文章