澳门新葡萄京官网首页 4

Java并发——读写锁

1. 前言

在Java并发包中常用的锁(如:ReentrantLock),基本上都是排他锁,这些锁在同一时刻只允许一个线程进行访问,而读写锁在同一时
刻可以允许多个读线程访问,但是在写线程访问时,所有的读线程和其他写线程均被阻塞。读写锁维护了一对锁,一个读锁和一个写锁,通过分离读锁和写锁,使得
并发性相比一般的排他锁有了很大提升。

除了保证写操作对读操作的可见性以及并发性的提升之外,读写锁能够简化读写交互场景的编程方式。假设在程序中定义一个共享的数据结构用作缓存,它大部分时间提供读服务(例如:查询和搜索),而写操作占有的时间很少,但是写操作完成之后的更新需要对后续的读服务可见。

在没有读写锁支持的(Java 5
之前)时候,如果需要完成上述工作就要使用Java的等待通知机制,就是当写操作开始时,所有晚于写操作的读操作均会进入等待状态,只有写操作完成并进行
通知之后,所有等待的读操作才能继续执行(写操作之间依靠synchronized关键字进行同步),这样做的目的是使读操作都能读取到正确的数据,而不
会出现脏读。改用读写锁实现上述功能,只需要在读操作时获取读锁,而写操作时获取写锁即可,当写锁被获取到时,后续(非当前写操作线程)的读写操作都会被
阻塞,写锁释放之后,所有操作继续执行,编程方式相对于使用等待通知机制的实现方式而言,变得简单明了。

澳门新葡萄京官网首页,一般情况下,读写锁的性能都会比排它锁要好,因为大多数场景读是多于写的。在读多于写的情况下,读写锁能够提供比排它锁更好的并发性和吞吐量。Java并发包提供读写锁的实现是ReentrantReadWriteLock,它提供的特性如表1所示。

表1. ReentrantReadWriteLock的特性

特性 说明
公平性选择 支持非公平(默认)和公平的锁获取方式,吞吐量还是非公平优于公平
重进入 该锁支持重进入,以读写线程为例:读线程在获取了读锁之后,能够再次获取读锁。而写线程在获取了写锁之后能够再次获取写锁,同时也可以获取读锁
锁降级 遵循获取写锁、获取读锁再释放写锁的次序,写锁能够降级成为读锁

读写锁在同一时刻可以允许多个读线程访问,但是在写线程访问时,所有的读
线程和其他写线程均被阻塞。读写锁维护了一对锁,一个读锁和一个写锁,通过分离读锁和写锁,使得并发性相比一般的排他锁有了很大提升。

此篇博客所有源码均来自JDK 1.8

2. 读写锁的接口与示例

ReadWriteLock仅定义了获取读锁和写锁的两个方法,即readLock()和writeLock()方法,而其实现—
ReentrantReadWriteLock,除了接口方法之外,还提供了一些便于外界监控其内部工作状态的方法,这些方法以及描述如表2所示。

表2. ReentrantReadWriteLock展示内部工作状态的方法

方法名称 描述
int getReadLockCount() 返回当前读锁被获取的次数。该次数不等于获取读锁的线程数,比如:仅一个线程,它连续获取(重进入)了n次读锁,那么占据读锁的线程数是1,但该方法返回n
int getReadHoldCount() 返回当前线程获取读锁的次数。该方法在Java 6 中加入到ReentrantReadWriteLock中,使用ThreadLocal保存当前线程获取的次数,这也使得Java 6 的实现变得更加复杂
boolean isWriteLocked() 判断写锁是否被获取
int getWriteHoldCount() 返回当前写锁被获取的次数

接下来通过一个缓存示例说明读写锁的使用方式,示例代码如代码清单1所示。

代码清单1. Cache.java

public class Cache {
  static Map<String, Object> map = new HashMap<String, Object>();
  static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
  static Lock r = rwl.readLock();
  static Lock w = rwl.writeLock();
  // 获取一个key对应的value
  public static final Object get(String key) {
    r.lock();
    try {
      return map.get(key);
    } finally {
      r.unlock();
    }
  }
  // 设置key对应的value,并返回旧有的value
  public static final Object put(String key, Object value) {
    w.lock();
    try {
      return map.put(key, value);
    } finally {
      w.unlock();
    }
  }
  // 清空所有的内容
  public static final void clear() {
    w.lock();
    try {
      map.clear();
    } finally {
      w.unlock();
    }
  }
}

上述示例中,Cache组合了一个非线程安全的HashMap作为缓存的实现,同时使用读写锁的读锁和写锁来保证Cache是线程安全的。在读操作
get(String
key)方法中,需要获取读锁,这使得并发访问该方法时不会被阻塞。写操作put(String
key, Object
value)和clear()方法,在更新HashMap时必须提前获取写锁,当写锁被获取后,其他线程对于读锁和写锁的获取均被阻塞,而只有写锁被释放
之后,其他读写操作才能继续。Cache使用读写锁提升读操作并发性,也保证每次写操作对所有的读写操作的可见性,同时简化了编程方式。

一般情况下,读写锁的性能都会比排它锁好,因为大多数场景读是多于写的。在读多于写的情况下,读写锁能够提供比排它锁更好的并发性和吞吐量。Java并发包提供读写锁的实现是ReentrantReadWriteLock,它提供的特性如表所示:

重入锁ReentrantLock是排他锁,排他锁在同一时刻仅有一个线程可以进行访问,但是在大多数场景下,大部分时间都是提供读服务,而写服务占有的时间较少。然而读服务不存在数据竞争问题,如果一个线程在读时禁止其他线程读势必会导致性能降低。所以就提供了读写锁。

3. 读写锁的实现分析

接下来将分析ReentrantReadWriteLock的实现,主要包括:读写状态的设计、写锁的获取与释放、读锁的获取与释放以及锁降级(以下没有特别说明读写锁均可认为是ReentrantReadWriteLock)。

澳门新葡萄京官网首页 1

读写锁维护着一对锁,一个读锁和一个写锁。通过分离读锁和写锁,使得并发性比一般的排他锁有了较大的提升:在同一时间可以允许多个读线程同时访问,但是在写线程访问时,所有读线程和写线程都会被阻塞。

3.1 读写状态的设计

读写锁同样依赖自定义同步器来实现同步功能,而读写状态就是其同步器的同步状态。回想ReentrantLock中自定义同步器的实现,同步状态
表示锁被一个线程重复获取的次数,而读写锁的自定义同步器需要在同步状态(一个整型变量)上维护多个读线程和一个写线程的状态,使得该状态的设计成为读写
锁实现的关键。

如果在一个整型变量上维护多种状态,就一定需要“按位切割使用”这个变量,读写锁是将变量切分成了两个部分,高16位表示读,低16位表示写,划分方式如图1所示。

图1. 读写锁状态的划分方式

澳门新葡萄京官网首页 2

如图1所示,当前同步状态表示一个线程已经获取了写锁,且重进入了两次,同时也连续获取了两次读锁。读写锁是如何迅速的确定读和写各自的状态呢?
答案是通过位运算。假设当前同步状态值为S,写状态等于 S &
0x0000FFFF(将高16位全部抹去),读状态等于 S >>>
16(无符号补0右移16位)。当写状态增加1时,等于S +
1,当读状态增加1时,等于S + (1 << 16),也就是S + 0×00010000。

根据状态的划分能得出一个推论:S不等于0时,当写状态(S &
0x0000FFFF)等于0时,则读状态(S >>> 16)大于0,即读锁已被获取。

ReentrantReadWriteLock的特性

读写锁的主要特性:

3.2 写锁的获取与释放

写锁是一个支持重进入的排它锁。如果当前线程已经获取了写锁,则增加写状态。如果当前线程在获取写锁时,读锁已经被获取(读状态不为0)或者该线程不是已经获取写锁的线程,则当前线程进入等待状态,获取写锁的代码如代码清单2所示。

代码清单2. ReentrantReadWriteLock的tryAcquire方法

protected final boolean tryAcquire(int acquires) {
  Thread current = Thread.currentThread();
  int c = getState();
  int w = exclusiveCount(c);
  if (c != 0) {
    // 存在读锁或者当前获取线程不是已经获取写锁的线程
    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;
}

该方法除了重入条件(当前线程为获取了写锁的线程)之外,增加了一个读锁是否存在的判断。如果存在读锁,则写锁不能被获取,原因在于:读写锁要确保
写锁的操作对读锁可见,如果允许读锁在已被获取的情况下对写锁的获取,那么正在运行的其他读线程就无法感知到当前写线程的操作。因此只有等待其他读线程都
释放了读锁,写锁才能被当前线程所获取,而写锁一旦被获取,则其他读写线程的后续访问均被阻塞。

写锁的释放与ReentrantLock的释放过程基本类似,每次释放均减少写状态,当写状态为0时表示写锁已被释放,从而等待的读写线程能够继续访问读写锁,同时前次写线程的修改对后续读写线程可见。

读写锁的实现分析

1 读写状态的设计

读写锁同样依赖自定义同步器来实现同步功能,而读写状态就是其同步器的同步状态。读写锁的自定义同步器需要在同步状态(一个整型变量)上维护多个读线程和一个写线程的状态,使得该状态的设计成为读写锁实现的关键。

如果在一个整型变量上维护多种状态,就一定需要“按位切割使用”这个变量,读写锁将变量切分成了两个部分,高16位表示读,低16位表示写,划分方式如图所示:

澳门新葡萄京官网首页 3

读写锁状态的划分方式

读写锁是如何迅速确定读和写各自的状态呢?答案是通过位运算。假设当前同步状态值为S,写状态等于S&0x0000FFFF(将高16位全部抹去),读状态等于S>>>16(无符号补0右移16位)。当写状态增加1时,等于S+1,当读状态增加1时,等于S+(1<<16),也就是S+0x00010000。

根据状态的划分能得出一个推论:S不等于0时,当写状态(S&0x0000FFFF)等于0时,则读状态(S>>>16)大于0,即读锁已被获取。

2 写锁的获取与释放

写锁是一个支持重进入的排它锁,获取情况有两种

  • 如果当前线程已经获取了写锁,则增加写状态。
  • 如果当前线程在获取写锁时,读锁已经被获取(读状态不为0)或者该线程不是已经获取写锁的线程,则当前线程进入等待状态,代码如下所示:

protected final boolean tryAcquire(int acquires) {
    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c);
    if (c != 0) {
        // 存在读锁或者当前获取线程不是已经获取写锁的线程
        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;
}

3 读锁的获取与释放

读锁是一个支持重进入的共享锁,它能够被多个线程同时获取。

  • 在没有其他写线程访问(或者写状态为0)时,读锁总会被成功地获取,而所做的也只是(线程安全的)增加读状态
  • 如果当前线程已经获取了读锁,则增加读状态。
  • 如果当前线程在获取读锁时,写锁已被其他线程获取,则进入等待状态。

protected final int tryAcquireShared(int unused) {
    for (;;) {
        int c = getState();
        int nextc = c + (1 << 16);
        if (nextc < c)
            throw new Error("Maximum lock count exceeded");
        // 如果当前线程不是获取写锁的线程,则获取读锁失败
        if (exclusiveCount(c) != 0 && owner != Thread.currentThread())
            return -1;
        // 如果当前线程是获取写锁的线程(锁降级) 
        // 或写锁未被线程获取,则获取读锁
        if (compareAndSetState(c, nextc))
            return 1;
    }
}

4 锁降级

如果当前线程拥有写锁,然后将其释放,最后再获取读锁,这种分段完成的过程不能称之为锁降级。

锁降级是指把持住(当前拥有的)写锁,再获取到读锁,随后释放(先前拥有的)写锁的过程。

那么锁降级的设计的目的是什么呢?为何要在拥有写锁的前提下去获取读锁?
通过查看一些文章,写一下自己的理解:

锁降级的目的其实是为了让线程对数据变化敏感,如果先释放写锁,再获取读锁,可能在获取之前,会有其他线程获取到写锁,阻塞读锁的获取,就无法感知数据变化了。所以需要先hold住写锁,保证数据无变化,获取读锁,然后再释放写锁。

例如有多个线程对同一块数据区域data进行读写操作,要求对每次数据的更改敏感。假设t1时刻data区域被写线程将状态s0更改为s1,更改完后若直接释放锁,那么可能会有其他线程获取写锁,将data区域的状态从s1更改为s2,这样一来整个过程就无法感知到data区域的s1状态。

如果采用了锁降级,那么获取写锁的线程t将data区域状态更改为s1后便持有读锁,那么其它想获取写锁的线程将会阻塞,直到线程t将读锁释放,那么这个过程中将会感知到data区域的s1状态。

澳门新葡萄京官网首页 4

锁降级对比

  1. 公平性:支持公平性和非公平性。
  2. 重入性:支持重入。读写锁最多支持65535个递归写入锁和65535个递归读取锁。
  3. 锁降级:遵循获取写锁、获取读锁在释放写锁的次序,写锁能够降级成为读锁

3.3 读锁的获取与释放

读锁是一个支持重进入的共享锁,它能够被多个线程同时获取,在没有其他写线程访问(或者写状态为0)时,读锁总会成功的被获取,而所做的也只是
(线程安全的)增加读状态。如果当前线程已经获取了读锁,则增加读状态。如果当前线程在获取读锁时,写锁已被其他线程获取,则进入等待状态。获取读锁的实
现从Java 5到Java
6变得复杂许多,主要原因是新增了一些功能,比如:getReadHoldCount()方法,返回当前线程获取读锁的次数。读状态是所有线程获取读锁次
数的总和,而每个线程各自获取读锁的次数只能选择保存在ThreadLocal中,由线程自身维护,这使获取读锁的实现变得复杂。因此,这里将获取读锁的
代码做了删减,保留必要的部分,代码如代码清单3所示。

代码清单3. ReentrantReadWriteLock的tryAcquireShared方法

protected final int tryAcquireShared(int unused) {
  for (;;) {
    int c = getState();
    int nextc = c + (1 << 16);
    if (nextc < c)
      throw new Error("Maximum lock count exceeded");
    if (exclusiveCount(c) != 0 && owner != Thread.currentThread())
      return -1;
    if (compareAndSetState(c, nextc))
      return 1;
  }
}

在tryAcquireShared(int
unused)方法中,如果其他线程已经获取了写锁,则当前线程获取读锁失败,进入等待状态。如果当前线程获取了写锁或者写锁未被获取,则当前线程(线程安全,依靠CAS保证)增加读状态,成功获取读锁。

读锁的每次释放均(线程安全的,可能有多个读线程同时释放读锁)减少读状态,减少的值是(1
<< 16)。

读写锁ReentrantReadWriteLock实现接口ReadWriteLock,该接口维护了一对相关的锁,一个用于只读操作,另一个用于写入操作。只要没有
writer,读取锁可以由多个 reader 线程同时保持。写入锁是独占的。

3.4 锁降级

锁降级指的是写锁降级成为读锁。如果当前线程拥有写锁,然后将其释放,最后再获取读锁,这种分段完成的过程不能称之为锁降级。锁降级是指把持住(当前拥有的)写锁,再获取到读锁,随后释放(先前拥有的)写锁的过程。

接下来看一个锁降级的示例:因为数据不常变化,所以多个线程可以并发的进行数据处理,当数据变更后,当前线程如果感知到数据变化,则进行数据的准备工作,同时其他处理线程被阻塞,直到当前线程完成数据的准备工作,示例代码如代码清单4所示。

代码清单4. processData方法

public void processData() {
  readLock.lock();
  if (!update) {
    // 必须先释放读锁
    readLock.unlock();
    // 锁降级从写锁获取到开始
    writeLock.lock();
    try {
      if (!update) {
        // 准备数据的流程(略)
        update = true;
      }
      readLock.lock();
    } finally {
      writeLock.unlock();
    }
    // 锁降级完成,写锁降级为读锁
  }
  try {
    // 使用数据的流程(略)
  } finally {
    readLock.unlock();
  }
}

上述示例中,当数据发生变更后,update变量(布尔类型且Volatile修饰)被设置为false,此时所有访问processData()
方法的线程都能够感知到变化,但只有一个线程能够获取到写锁,而其他线程会被阻塞在读锁和写锁的lock()方法上。当前程获取写锁完成数据准备之后,再
获取读锁,随后释放写锁,完成锁降级。

锁降级中读锁的获取是否必要呢?答案是必要的。主要原因是保证数据的可见性,如果当前线程不获取读锁而是直接释放写锁,假设此刻另一个线程(记作
线程T)获取了写锁并修改了数据,则当前线程无法感知线程T的数据更新。如果当前线程获取读锁,即遵循锁降级的步骤,则线程T将会被阻塞,直到当前线程使
用数据并释放读锁之后,线程T才能获取写锁进行数据更新。

RentrantReadWriteLock不支持锁升级(把持读锁、获取写锁,最后释放读锁的过程)。原因也是保证数据可见性,如果读锁已被多个线程获取,其中任意线程成功获取了写锁并更新了数据,则其更新对其他获取到读锁的线程不可见。

public interface ReadWriteLock {
    Lock readLock();
    Lock writeLock();
}

ReadWriteLock定义了两个方法。readLock()返回用于读操作的锁,writeLock()返回用于写操作的锁。ReentrantReadWriteLock定义如下:

    /** 内部类  读锁 */
    private final ReentrantReadWriteLock.ReadLock readerLock;
    /** 内部类  写锁 */
    private final ReentrantReadWriteLock.WriteLock writerLock;

    final Sync sync;

    /** 使用默认(非公平)的排序属性创建一个新的 ReentrantReadWriteLock */
    public ReentrantReadWriteLock() {
        this(false);
    }

    /** 使用给定的公平策略创建一个新的 ReentrantReadWriteLock */
    public ReentrantReadWriteLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
        readerLock = new ReadLock(this);
        writerLock = new WriteLock(this);
    }

    /** 返回用于写入操作的锁 */
    public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
    /** 返回用于读取操作的锁 */
    public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }

 abstract static class Sync extends AbstractQueuedSynchronizer {
        /**
         * 省略其余源代码
         */
    }

    public static class WriteLock implements Lock, java.io.Serializable{
        /**
         * 省略其余源代码
         */
    }

    public static class ReadLock implements Lock, java.io.Serializable {
        /**
         * 省略其余源代码
         */
    }

ReentrantReadWriteLock与ReentrantLock一样,其锁主体依然是Sync,它的读锁、写锁都是依靠Sync来实现的。所以ReentrantReadWriteLock实际上只有一个锁,只是在获取读取锁和写入锁的方式上不一样而已,它的读写锁其实就是两个类:ReadLock、writeLock,这两个类都是lock实现。

在ReentrantLock中使用一个int类型的state来表示同步状态,该值表示锁被一个线程重复获取的次数。但是读写锁ReentrantReadWriteLock内部维护着两个一对锁,需要用一个变量维护多种状态。所以读写锁采用“按位切割使用”的方式来维护这个变量,将其切分为两部分,高16为表示读,低16为表示写。分割之后,读写锁是如何迅速确定读锁和写锁的状态呢?通过为运算。假如当前同步状态为S,那么写状态等于
S & 0x0000FFFF(将高16位全部抹去),读状态等于S >>>
16(无符号补0右移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;

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

写锁

写锁就是一个支持可重入的排他锁。

写锁的获取

写锁的获取最终会调用tryAcquire(int arg),该方法在内部类Sync中实现:

    protected final boolean tryAcquire(int acquires) {
        Thread current = Thread.currentThread();
        //当前锁个数
        int c = getState();
        //写锁
        int w = exclusiveCount(c);
        if (c != 0) {
            //c != 0 && w == 0 表示存在读锁
            //当前线程不是已经获取写锁的线程
            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;
    }

该方法和ReentrantLock的tryAcquire(int
arg)大致一样,在判断重入时增加了一项条件:读锁是否存在。因为要确保写锁的操作对读锁是可见的,如果在存在读锁的情况下允许获取写锁,那么那些已经获取读锁的其他线程可能就无法感知当前写线程的操作。因此只有等读锁完全释放后,写锁才能够被当前线程所获取,一旦写锁获取了,所有其他读、写线程均会被阻塞。

写锁的释放

获取了写锁用完了则需要释放,WriteLock提供了unlock()方法释放写锁:

    public void unlock() {
        sync.release(1);
    }

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

写锁的释放最终还是会调用AQS的模板方法release(int
arg)方法,该方法首先调用tryRelease(int
arg)方法尝试释放锁,tryRelease(int
arg)方法为读写锁内部类Sync中定义了,如下:

    protected final boolean tryRelease(int releases) {
        //释放的线程不为锁的持有者
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        int nextc = getState() - releases;
        //若写锁的新线程数为0,则将锁的持有者设置为null
        boolean free = exclusiveCount(nextc) == 0;
        if (free)
            setExclusiveOwnerThread(null);
        setState(nextc);
        return free;
    }

写锁释放锁的整个过程和独占锁ReentrantLock相似,每次释放均是减少写状态,当写状态为0时表示
写锁已经完全释放了,从而等待的其他线程可以继续访问读写锁,获取同步状态,同时此次写线程的修改对后续的线程可见。

读锁

读锁为一个可重入的共享锁,它能够被多个线程同时持有,在没有其他写线程访问时,读锁总是或获取成功。

读锁的获取

读锁的获取可以通过ReadLock的lock()方法:

        public void lock() {
            sync.acquireShared(1);
        }

Sync的acquireShared(int arg)定义在AQS中:

    public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }

tryAcqurireShared(int
arg)尝试获取读同步状态,该方法主要用于获取共享式同步状态,获取成功返回
>= 0的返回结果,否则返回 < 0 的返回结果。

    protected final int tryAcquireShared(int unused) {
        //当前线程
        Thread current = Thread.currentThread();
        int c = getState();
        //exclusiveCount(c)计算写锁
        //如果存在写锁,且锁的持有者不是当前线程,直接返回-1
        //存在锁降级问题,后续阐述
        if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
            return -1;
        //读锁
        int r = sharedCount(c);

        /*
         * readerShouldBlock():读锁是否需要等待(公平锁原则)
         * r < MAX_COUNT:持有线程小于最大数(65535)
         * compareAndSetState(c, c + SHARED_UNIT):设置读取锁状态
         */
        if (!readerShouldBlock() &&
                r < MAX_COUNT &&
                compareAndSetState(c, c + SHARED_UNIT)) {
            /*
             * holdCount部分后面讲解
             */
            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);
    }

**```**
****
读锁获取的过程相对于独占锁而言会稍微复杂下,整个过程如下:

1. 因为存在锁降级情况,如果存在写锁且锁的持有者不是当前线程则直接返回失败,否则继续
2. 依据公平性原则,判断读锁是否需要阻塞,读锁持有线程数小于最大值(65535),且设置锁状态成功,执行以下代码(对于HoldCounter下面再阐述),并返回1。如果不满足改条件,执行fullTryAcquireShared()。

```java
    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 (firstReader == current) {
                }
                //HoldCounter后面讲解
                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)
                throw new Error("Maximum lock count exceeded");
            //CAS设置读锁成功
            if (compareAndSetState(c, c + SHARED_UNIT)) {
                //如果是第1次获取“读取锁”,则更新firstReader和firstReaderHoldCount
                if (sharedCount(c) == 0) {
                    firstReader = current;
                    firstReaderHoldCount = 1;
                }
                //如果想要获取锁的线程(current)是第1个获取锁(firstReader)的线程,则将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;
            }
        }
    }

fullTryAcquireShared(Thread
current)会根据“是否需要阻塞等待”,“读取锁的共享计数是否超过限制”等等进行处理。如果不需要阻塞等待,并且锁的共享计数没有超过限制,则通过CAS尝试获取锁,并返回1

读锁的释放

与写锁相同,读锁也提供了unlock()释放读锁:

        public void unlock() {
            sync.releaseShared(1);
        }

unlcok()方法内部使用Sync的releaseShared(int
arg)方法,该方法定义在AQS中:

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

调用tryReleaseShared(int
arg)尝试释放读锁,该方法定义在读写锁的Sync内部类中:

    protected final boolean tryReleaseShared(int unused) {
        Thread current = Thread.currentThread();
        //如果想要释放锁的线程为第一个获取锁的线程
        if (firstReader == current) {
            //仅获取了一次,则需要将firstReader 设置null,否则 firstReaderHoldCount - 1
            if (firstReaderHoldCount == 1)
                firstReader = null;
            else
                firstReaderHoldCount--;
        }
        //获取rh对象,并更新“当前线程获取锁的信息”
        else {
            HoldCounter rh = cachedHoldCounter;
            if (rh == null || rh.tid != getThreadId(current))
                rh = readHolds.get();
            int count = rh.count;
            if (count <= 1) {
                readHolds.remove();
                if (count <= 0)
                    throw unmatchedUnlockException();
            }
            --rh.count;
        }
        //CAS更新同步状态
        for (;;) {
            int c = getState();
            int nextc = c - SHARED_UNIT;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }

HoldCounter

在读锁获取锁和释放锁的过程中,我们一直都可以看到一个变量rh (HoldCounter
),该变量在读锁中扮演着非常重要的作用。

我们了解读锁的内在机制其实就是一个共享锁,为了更好理解HoldCounter
,我们暂且认为它不是一个锁的概率,而相当于一个计数器。一次共享锁的操作就相当于在该计数器的操作。获取共享锁,则该计数器

  • 1,释放共享锁,该计数器 –
    1。只有当线程获取共享锁后才能对共享锁进行释放、重入操作。所以HoldCounter的作用就是当前线程持有共享锁的数量,这个数量必须要与线程绑定在一起,否则操作其他线程锁就会抛出异常。我们先看HoldCounter的定义:

          static final class HoldCounter {
              int count = 0;
              final long tid = getThreadId(Thread.currentThread());
          }
    

HoldCounter 定义非常简单,就是一个计数器count 和线程 id tid
两个变量。按照这个意思我们看到HoldCounter
是需要和某给线程进行绑定了,我们知道如果要将一个对象和线程绑定仅仅有tid是不够的,而且从上面的代码我们可以看到HoldCounter
仅仅只是记录了tid,根本起不到绑定线程的作用。那么怎么实现呢?答案是ThreadLocal,定义如下:

        static final class ThreadLocalHoldCounter
            extends ThreadLocal<HoldCounter> {
            public HoldCounter initialValue() {
                return new HoldCounter();
            }
        }

通过上面代码HoldCounter就可以与线程进行绑定了。故而,HoldCounter应该就是绑定线程上的一个计数器,而ThradLocalHoldCounter则是线程绑定的ThreadLocal。从上面我们可以看到ThreadLocal将HoldCounter绑定到当前线程上,同时HoldCounter也持有线程Id,这样在释放锁的时候才能知道ReadWriteLock里面缓存的上一个读取线程(cachedHoldCounter)是否是当前线程。这样做的好处是可以减少ThreadLocal.get()的次数,因为这也是一个耗时操作。需要说明的是这样HoldCounter绑定线程id而不绑定线程对象的原因是避免HoldCounter和ThreadLocal互相绑定而GC难以释放它们(尽管GC能够智能的发现这种引用而回收它们,但是这需要一定的代价),所以其实这样做只是为了帮助GC快速回收对象而已。

看到这里我们明白了HoldCounter作用了,我们在看一个获取读锁的代码段:

                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
                }

这段代码涉及了几个变量:firstReader
、firstReaderHoldCount、cachedHoldCounter 。我们先理清楚这几个变量:

private transient Thread firstReader = null;
private transient int firstReaderHoldCount;
private transient HoldCounter cachedHoldCounter;

firstReader
看名字就明白了为第一个获取读锁的线程,firstReaderHoldCount为第一个获取读锁的重入数,cachedHoldCounter为HoldCounter的缓存。

理清楚上面所有的变量了,HoldCounter也明白了,我们就来给上面那段代码标明注释,如下:

    //如果获取读锁的线程为第一次获取读锁的线程,则firstReaderHoldCount重入数 + 1
    else if (firstReader == current) {
        firstReaderHoldCount++;
    } else {
        //非firstReader计数
        if (rh == null)
            rh = cachedHoldCounter;
        //rh == null 或者 rh.tid != current.getId(),需要获取rh
        if (rh == null || rh.tid != getThreadId(current))
            rh = readHolds.get();
            //加入到readHolds中
        else if (rh.count == 0)
            readHolds.set(rh);
        //计数+1
        rh.count++;
        cachedHoldCounter = rh; // cache for release
    }

这里解释下为何要引入firstRead、firstReaderHoldCount。这是为了一个效率问题,firstReader是不会放入到readHolds中的,如果读锁仅有一个的情况下就会避免查找readHolds。

锁降级

上开篇是LZ就阐述了读写锁有一个特性就是锁降级,锁降级就意味着写锁是可以降级为读锁的,但是需要遵循先获取写锁、获取读锁在释放写锁的次序。注意如果当前线程先获取写锁,然后释放写锁,再获取读锁这个过程不能称之为锁降级,锁降级一定要遵循那个次序。

在获取读锁的方法tryAcquireShared(int
unused)中,有一段代码就是来判读锁降级的:

        int c = getState();
        //exclusiveCount(c)计算写锁
        //如果存在写锁,且锁的持有者不是当前线程,直接返回-1
        //存在锁降级问题,后续阐述
        if (exclusiveCount(c) != 0 &&
                getExclusiveOwnerThread() != current)
            return -1;
        //读锁
        int r = sharedCount(c);

锁降级中读锁的获取释放为必要?肯定是必要的。试想,假如当前线程A不获取读锁而是直接释放了写锁,这个时候另外一个线程B获取了写锁,那么这个线程B对数据的修改是不会对当前线程A可见的。如果获取了读锁,则线程B在获取写锁过程中判断如果有读锁还没有释放则会被阻塞,只有当前线程A释放读锁后,线程B才会获取写锁成功。

推荐阅读

因为里面很多地方涉及到了AQS部分,推荐阅读如下部分:

  1. 【死磕Java并发】—–J.U.C之AQS:AQS简介
  2. 【死磕Java并发】—–J.U.C之AQS:CLH同步队列
  3. 【死磕Java并发】—–J.U.C之AQS:同步状态的获取与释放
  4. 【死磕Java并发】—–J.U.C之AQS:阻塞和唤醒线程

参考资料

  1. Doug Lea:《Java并发编程实战》
  2. 方腾飞:《Java并发编程的艺术》
  3. 【Java并发编程实战】—–“J.U.C”:ReentrantReadWriteLock](https://link.jianshu.com?t=%5Bhttp://cmsblogs.com/?p=1679)]()))
  4. Java多线程(十)之ReentrantReadWriteLock深入分析

发表评论

电子邮件地址不会被公开。 必填项已用*标注