Java并发编程之锁机制之ReentrantReadWriteLock（读写锁）

时间  2019-11-07
标签 java 并发 编程 机制 reentrantreadwritelock 读写 栏目 Java 繁體版
原文   原文链接

前言

在前面的文章中，咱们讲到了ReentrantLock(重入锁)，接下来咱们讲ReentrantReadWriteLock（读写锁），该锁具有重入锁的可重入性、可中断获取锁等特征，可是与ReentrantLock不同的是，在ReentrantReadWriteLock中，维护了一对锁，一个读锁一个写锁，而读写锁在同一时刻容许多个读线程访问。可是在写线程访问时，全部的读线程和其余的写线程均被阻塞。在阅读本片文章以前，但愿你已阅读过如下几篇文章：编程

基本结构

在具体了解ReentrantReadWriteLock以前，咱们先看一下其总体结构，具体结构以下图所示： 缓存

ReentrantReadWriteLock.png

从总体图上来看，ReentrantReadWriteLock实现了ReadWriteLock接口，其中在ReentrantReadWriteLock中分别声明了如下几个静态内部类：bash

  • WriteLock与ReadLock（维护的一对读写锁）：单从类名咱们能够看出这两个类的做用，就是控制读写线程的锁
  • Sync及其子类NofairSync与FairSync：若是你阅读过 Java并发编程之锁机制之重入锁中公平锁与非公平锁的介绍，那么咱们也能够猜想出ReentrantReadWriteLock（读写锁）是支持公平锁与非公平锁的。
  • ThreadLoclHoldCounter及HoldCounter：涉及到锁的重进入，在下文中咱们会具体进行描述。

基本使用

在使用某些种类的Collection时，可使用ReentrantReadWriteLock 来提升并发性。一般，在预期Collection 很大，且读取线程访问它的次数多于写入线程的状况下，且所承担的操做开销高于同步开销时，这很值得一试。例如，如下是一个使用 TreeMap（咱们假设预期它很大，而且能被同时访问） 的字典类。数据结构

class RWDictionary {
    private final Map<String, Data> m = new TreeMap<String, Data>();
    private final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    private final Lock r = rwl.readLock();//获取读锁
    private final Lock w = rwl.writeLock();//获取写锁
    
	//读取Map中的对应key的数据
    public Data get(String key) {
        r.lock();
        try { return m.get(key); }
        finally { r.unlock(); }
    }
    //读取Map中全部的key
    public String[] allKeys() {
        r.lock();
        try { return m.keySet().toArray(); }
        finally { r.unlock(); }
    }
    //往Map中写数据
    public Data put(String key, Data value) {
        w.lock();
        try { return m.put(key, value); }
        finally { w.unlock(); }
    }
    //清空数据
    public void clear() {
        w.lock();
        try { m.clear(); }
        finally { w.unlock(); }
    }
 }
复制代码

在上述例子中，咱们分别对TreeMap中的读取操做进行了加锁的操做。当咱们调用get(String key)方法，去获取TreeMap中对应key值的数据时，须要先获取读锁。那么其余线程对于写锁的获取将会被阻塞，而对获取读锁的线程不会阻塞。同理，当咱们调用put(String key, Data value)方法，去更新数据时，咱们须要获取写锁。那么其余线程对于写锁与读锁的获取都将会被阻塞。只有当获取写锁的线程释放了锁以后。其余读写操做才能进行。并发

这里可能会有小伙伴会有疑问，为何当获取写锁成功后，会阻塞其余的读写操做？，这里实际上是为了保证数据可见性。若是不阻塞其余读写操做，假如读操做优先与写操做，那么在数据更新以前，读操做获取的数据与写操做更新后的数据就会产生不一致的状况。app

须要注意的是：ReentrantReadWriteLock最多支持 65535 个递归写入锁和65535个读取锁。试图超出这些限制将致使锁方法抛出 Error。具体缘由会在下文进行描述。函数

实现原理

到如今为止，咱们已经基本了解了ReentrantReadWriteLock的基本结构与基本使用。我相信你们确定对其内部原理感到好奇，下面我会带着你们一块儿去了解其内部实现。这里我会对总体的一个原理进行分析，内部更深的细节会在下文进行描述。由于我以为只有理解总体原理后，再去理解其中的细节。那么对整个ReentrantReadWriteLock（读写锁）的学习来讲，要容易一点。高并发

总体原理

在前文中，咱们介绍了ReentrantReadWriteLock的基本使用，咱们发现整个读写锁对线程的控制是交给了WriteLock与ReadLock。当咱们调用读写锁的lock()方法去获取相应的锁时，咱们会执行如下代码：工具

public void lock() { sync.acquireShared(1);}
复制代码

也就是会调用sync.acquireShared(1)，而sync又是什么呢？从其构造函数中咱们也能够看出:post

public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }
复制代码

其中关于FairSync与NonfairSync的声明以下所示：

//同步队列
abstract static class Sync extends AbstractQueuedSynchronizer {省略部分代码...}
//非公平锁
static final class NonfairSync extends Sync{省略部分代码...}
//公平锁
static final class FairSync extends Sync {省略部分代码...}
复制代码

这里咱们又看到了咱们熟悉的AQS，也就是说WriteLock与ReadLock这两个锁，实际上是经过AQS中的同步队列来对线程的进行控制的。那么结合咱们以前的AQS的知识，咱们能够获得下图：

（若是你对AQS不熟，那么你能够阅读该篇文章---->Java并发编程之锁机制之AQS(AbstractQueuedSynchronizer)

读写锁状态关系图.png
这里我省略了 为何维护的是同一个同步队列的缘由，这个问题留给你们。

读写状态设计

虽然如今咱们已经知道了，WriteLock与ReadLock这两个锁维护了同一个同步队列，可是我相信你们都会有个疑问，同步队列中只有一个int类型的state变量来表示当前的同步状态。那么其内部是怎么将两个读写状态分开，而且达到控制线程的目的的呢？

在ReentrantReadWriteLock中的同步队列，实际上是将同步状态分为了两个部分，其中高16位表示读状态，低16位表示写状态，具体状况以下图所示：

读写锁状态划分.png

在上图中，咱们能得知，读写状态能表示的最大值为65535(排除负数)，也就是说容许锁重进入的次数为65535次。

接下来 咱们单看高16位，这里表示当前线程已经获取了写锁，且重进入了七次。一样的这里若是咱们也只但看低16位，那么就表示当前线程获取了读锁，且重进入了七次。这里你们须要注意的是，在实际的状况中，读状态与写状态是不能被不一样线程同时赋值的。由于根据ReentrantReadWriteLock的设计来讲，读写操做线程是互斥的。上图中这样表示，只是为了帮助你们理解同步状态的划分。

到如今为止咱们已经知道同步状态的划分，那接下来又有新的问题了。如何快速的区分及获取读写状态呢？其实也很是简单。

  • 读状态：想要获取读状态，只须要将当前同步变量无符号右移16位
  • 写状态：咱们只须要将当前同步状态（这里用S表示）进行这样的操做S&0x0000FFFF)，也就是S&(1<<16-1)。

也就是以下图所示（可能图片不是很清楚，建议在pc端上观看）：

读写锁状态原理.png

细节分析

在了解了ReentrantReadWriteLock的总体原理及读写状态的划分后，咱们再来理解其内部的读写线程控制就容易的多了，下面的文章中，我会对读锁与写锁的获取分别进行讨论。

读锁的获取

由于当调用ReentrantReadWriteLock中的ReadLock的lock()方法时，最终会走Sync中的tryAcquireShared(int unused)方法，来判断可否获取写锁。那如今咱们就来看看该方法的具体实现。具体代码以下所示：

protected final int tryAcquireShared(int unused) {
            Thread current = Thread.currentThread();
            int c = getState();
            //（1）判断当前是否有写锁，有直接返回
            if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
                return -1;
               
            int r = sharedCount(c);
             //(2)获取当前读锁的状态，判断是否小于最大值，
             //同时根据公平锁，仍是非公平锁的模式，判断当前线程是否须要阻塞，
            if (!readerShouldBlock() &&
                r < MAX_COUNT &&
	                compareAndSetState(c, c + SHARED_UNIT)) {
                //（3)若是是不要阻塞，且写状态小于最大值，则设置当前线程重进入的次数
                if (r == 0) {
			        //若是当前读状态为0，则设置当前读线程为，当前线程为第一个读线程。
                    firstReader = current;
                    firstReaderHoldCount = 1;
                } else if (firstReader == current) {
		            //计算第一个读线程，重进入的次数
                    firstReaderHoldCount++;
                } else {
	                //经过ThreadLocl获取读线程中进入的锁
                    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;//获取共享同步状态成功
            }
            //(4)当获取读状态失败后，继续尝试获取读锁，
            return fullTryAcquireShared(current);
        }
复制代码
  • （1）根据当前的同步状态，判断是否存在写锁，且当前拥有写锁的线程不是当前线程，那么直接返回-1，须要注意的是若是该方法返回值为负数，那么会将该请求线程加入到AQS的同步队列中。（对该方法不是很熟的小伙伴，建议查看 Java并发编程之锁机制之AQS(AbstractQueuedSynchronizer)
  • （2）获取当前读锁的状态，判断是否小于最大值，同时根据公平锁，仍是非公平锁的模式，判断当前线程是否须要阻塞
  • （3）若是条件（2）知足，则设置分别第一个读取线程重进入的次数及后续线程重进入的次数
  • （4）若是条件（2）不知足，在再次尝试获取读锁。

在读锁的获取中，涉及到的方法较为复杂，因此下面会对每一个步骤中涉及到的方法，进行介绍。

步骤（1）中如何判断是否有写锁？

在读锁的获取中的步骤（1）中，代码中会调用exclusiveCount(int c)方法来判当前是否存在写锁。而该方法是属于Sync中的方法，具体代码以下所示：

abstract static class Sync extends AbstractQueuedSynchronizer {

        static final int SHARED_SHIFT   = 16;
        static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
        static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;//最大状态数为2的16次方-1
        static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

        /*返回当前的读状态*/
        static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
        /*返回当前的写状态 */
        static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
       }
复制代码

从代码中咱们能够看出，只是简单的执行了c & EXCLUSIVE_MASK，也就是S&0x0000FFFF，结合咱们上文中咱们所讲的读写状态的区分，我相信exclusiveCount(int c)与sharedCount(int c)方法是不难理解的。

步骤（2）中如何判断是公平锁与非公平锁。

在步骤（2）中，咱们发现调用了readerShouldBlock()方法，而该方法是Sync类中的抽象方法。在ReentrantReadWriteLock类中，公平锁与非公平锁进行了相应的实现，具体代码以下图所示：

//公平锁
    static final class FairSync extends Sync {
        private static final long serialVersionUID = -2274990926593161451L;
        final boolean writerShouldBlock(){return hasQueuedPredecessors();}
        final boolean readerShouldBlock(){return hasQueuedPredecessors();
        }
    }
    //非公平锁
    static final class NonfairSync extends Sync {
        final boolean writerShouldBlock() { return false;}
        final boolean readerShouldBlock() {return apparentlyFirstQueuedIsExclusive();}
    }
复制代码

这里就再也不对公平锁与非公平锁进行分析了。在文章 Java并发编程之锁机制之重入锁中已经对这个知识点进行了分析。有兴趣的小伙伴能够参考该文章。

步骤（3）中为毛要记录第一个获取写锁的线程？线程的重进入是如何实现的？

在ReentrantReadWriteLock类中分别定义了Thread firstReader与int firstReaderHoldCount变量来记录当前第一个获取写锁的线程以及其重进入的次数。官方的给的解释是便于跟踪与记录线程且这种记录是很是廉价的。也就是说，之因此单独定义一个变量来记录第一个获取获取写锁的线程，是为了在众多的读线程中区分线程，也是为了之后的调试与跟踪。

当咱们解决了第一个问题后，如今咱们来解决第二个问题。这里我就不在对第一个线程如何记录重进入次数进行分析了。咱们直接看其余读线程的重进入次数设置。这里由于篇幅的限制，我就直接讲原理，其余线程的重进入的次数判断是经过ThreadLocal来实现的。经过在每一个线程中的内存空间保存HodlerCount类（用于记录当前线程获取锁的次数），来获取相应的次数。具体代码以下所示：

static final class HoldCounter {
            int count;//记录当前线程进入的次数
            final long tid = getThreadId(Thread.currentThread());
        }
    
   static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {
            public HoldCounter initialValue() {
                return new HoldCounter();
            }
        }
     
   private transient ThreadLocalHoldCounter readHolds;
复制代码

若是有小伙伴不熟悉ThreadLocal，能够参看该篇文章《Android Handler机制之ThreadLocal》

步骤（4）中继续尝试获取读锁？

当第一次获取读锁失败的时候，会调用fullTryAcquireShared(Thread current)方法会继续尝试获取锁。该函数返回的三个条件为：

  • 当前已经存在写锁了。直接加入AQS同步队列中。
  • 当前写锁的次数超过最大值，直接抛出异常
  • 获取读锁成功。直接返回

具体代码以下所示：

final int fullTryAcquireShared(Thread current) {
            HoldCounter rh = null;
            for (;;) {//注意这里的for循环
                int c = getState();
                if (exclusiveCount(c) != 0) {//（1）存在写锁直接返回
                    if (getExclusiveOwnerThread() != current)
                        return -1;
                } else if (readerShouldBlock()) {
                    // Make sure we're not acquiring read lock reentrantly if (firstReader == current) { // assert firstReaderHoldCount > 0; } 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; } } if (sharedCount(c) == MAX_COUNT)//(2)锁迭代次数超过最大值。抛出异常 throw new Error("Maximum lock count exceeded"); if (compareAndSetState(c, c + SHARED_UNIT)) {//(3)获取锁成功，记录次数 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; } } } 复制代码

由于该方法和上文提到的tryAcquireShared(int unused)方法较为相似。因此这里就再也不对其中的逻辑再次讲解。你们须要注意的是该方法会自旋式的获取锁。

写锁的获取

了解了读锁的获取，再来了解写锁的获取就很是简单了。写锁的获取最终会走Sync中的tryAcquire(int acquires)方法。具体代码以下所示：

protected final boolean tryAcquire(int acquires) {
            Thread current = Thread.currentThread();
            //（1）获取同步状态 = 写状态+读状态，单独获取写状态
            int c = getState();
            int w = exclusiveCount(c);
            //（2）若是c!=0则表示有线程操做
            if (c != 0) {
                // （2.1）没有写锁线程，则表示有读线程，则直接获取失败，并返回
                if (w == 0 || current != getExclusiveOwnerThread())
                    return false;
                    
                 //（2.2）若是w>0则，表示当前线程为写线程，则计算当前重进入的次数，若是已经饱和，则抛出异常
                if (w + exclusiveCount(acquires) > MAX_COUNT)
                    throw new Error("Maximum lock count exceeded");
                    
                // （2.3）获取成功，直接记录当前写状态
                setState(c + acquires);
                return true;
            }
            //（3）没有线程获取读写锁，根据当前锁的模式与设置写状态是否成功，判断是否须要阻塞线程
            if (writerShouldBlock() ||
                !compareAndSetState(c, c + acquires))
                return false;
            //(4)第一次进入，获取成功   
            setExclusiveOwnerThread(current);
            return true;
        }
复制代码

为了帮助你们理解，我这里将该方法分为了一下几个步骤：

  • （1）获取同步状态 c（写状态+读状态），并单独获取写状态w。
  • （2）若是c!=0则表示有线程操做。
  • （2.1）没有写锁线程，则表示有读线程，则直接获取失败，并返回。
  • （2.2）若是w>0则，表示当前线程为写线程，则计算当前重进入的次数，若是已经饱和，则抛出异常。
  • （2.3）获取成功，直接记录当前写状态。
  • （3）在（2）条件不知足的条件下，没有线程获取读写锁，根据当前锁的模式与设置写状态是否成功，判断是否须要阻塞线程
  • （4）在（2）（3）条件都不知足的状况下，则为第一次进入，那么就获取成功 。

相信结合以上步骤。再来理解代码就很是容易了。

锁降级

读写锁除了保证写操做对读操做的可见性以及并发性的提高以外，读写锁也能简化读写交互的编程方式，试想一种状况，在程序中咱们须要定义一个共享的用做缓存数据结构，而且其大部分时间提供读服务（例如查询和搜索），而写操做占有的时间不多，可是咱们又但愿写操做完成以后的更新须要对后续的读操做可见。那么该怎么实现呢？参看以下例子：

public class CachedData {
    Object data;
    volatile boolean cacheValid;
    final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
    
    void processCachedData() {
        rwl.readLock().lock();
        if (!cacheValid) {
            //若是缓存过时，释放读锁，并获取写锁
            rwl.readLock().unlock();
            rwl.writeLock().lock();（1）
            try {
                //从新检查缓存是否过时，由于有可能在当前线程操做以前，其余写线程有可能改变缓存状态
                if (!cacheValid) {
                    data = ...//从新写入数据
                    cacheValid = true;
                }
                // 获取读锁
                rwl.readLock().lock();（2）
            } finally {
	            //释放写锁
                rwl.writeLock().unlock(); （3）
            }
        }

        try {
            use(data);//操做使用数据
        } finally {
            rwl.readLock().unlock();//最后释放读锁
        }
    }
}
复制代码

在上述例子中，若是数据缓存过时，也就是cacheValid变量（volatile 修饰的布尔类型）被设置为false，那么全部调用processCachedData（）方法的线程都能感知到变化，可是只有一个线程能过获取到写锁。其余线程会被阻塞在读锁和写锁的lock()方法上。当前线程获取写锁完成数据准备以后，再获取读锁，随后释放写锁（上述代码的（1）（2）（3）三个步骤），这种在拥有写锁的状况下，在获取读锁。随后释放写锁的过程，称之为锁降级（在读写锁内部实现中，是支持锁锁降级的）。

那接下来，我个问题想问你们，为何当线程获取写锁，修改数据完成后，要先获取读锁呢，而不直接释放写锁呢？，其实缘由很简单，若是当前线程直接释放写锁，那么这个时候若是有其余线程获取了写锁，并修改了数据。那么对于当前释放写锁的线程是没法感知数据变化的。先获取读锁的目的，就是保证没有其余线程来修改数据啦。

总结

  • ReentrantReadWriteLock最多支持 65535 个递归写入锁和65535个读取锁。
  • ReentrantReadWriteLock中用同一int变量的高16位表示读状态，低16位表示写状态。
  • ReentrantReadWriteLock支持公平锁与非公平锁模式。
  • ReentrantReadWriteLock支持锁的降级。
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程