性能文章>一文读懂并发包中的读写锁和Condition实现原理>

一文读懂并发包中的读写锁和Condition实现原理原创

1年前
311034

导读

上一篇文章我们谈到了Java并发的基石:AQS的实现原理 图解并发包中锁的通用实现。这两天就有朋友说ReentrantLock中的Condition条件等待以及ReentrantReadWriteLock读写锁的实现有点繁琐,问我能不能讲下其关键实现原理。这篇文章我们就来谈谈这个主题。

这篇文章我们来Java中的读写锁以及ReentrantLock中Condition条件等待的实现。阅读完本篇文章,你将了解到:

  1. 读写锁的使用场景和优缺点

  2. 读写锁的实现原理

  3. 如何使用读写锁

  4. ReentrantLock的Condition底层是如何实现的

如果你不知道ReentrantLock的Condition以及ReentrantReadWriteLock是干嘛的,可以先阅读下我的这篇文章:
ReentrantLock介绍与使用:
https://www.itzhai.com/cpj/introduction-and-use-of-reentrantlock.html

ReentrantReadWriteLock介绍与使用:

https://www.itzhai.com/cpj/introduction-and-use-of-reentrantreadwritelock.html

1、Java中的读写锁

有这样一种场景:

  • 如果对一个共享资源的写操作没有读操作那么频繁,这个时候可以允许多个线程同时读取共享资源;

  • 但是如果有一个线程想去写这些共享资源,那么其他线程此刻就不应该对这些资源进行读和写操作了。

Java中的ReentrantReadWriteLock正是为这种场景提供的锁。该类里面包括了读锁和写锁。

1.1、可获取读锁的情况

  • 没有其他线程正在持有写锁;

  • 尝试获取读锁的线程同时持有写锁。

1.2、可获取写锁的情况

  • 没有其他线程正在持有读锁;

  • 没有其他线程正在持有写锁。

1.3、读写锁特点

  • 允许并发读:只要没有线程正在更新数据,那么多个线程就可以同时读取数据;

  • 只能独占写:只要有一个线程正在写数据,那么就会导致其他线程的读或者写均被阻塞;但写的线程可以获取读锁,并通过释放写锁,让锁降级为读锁;(不能由读锁升级为写锁)

  • 只要有一个线程正在读数据,那么其他线程的写入就会阻塞,直到读锁被释放;

  • 公平性:支持非公平锁和公平锁,非公平锁吞吐量较高;

  • 可重入:无论是读锁还是写锁都是支持可重入的。

读写锁可以增加更新不频繁而读取频繁的共享数据结构的吞吐量。

2、实现原理

ReentrantReadWriteLock是可重入读写锁的实现。我们先来看看涉及到的类:

image-20200311163334398

我们可以看到,ReentrantReadWriteLock中也具有非公平锁NonfairSync公平锁FairSync的实现。同时ReentrantReadWriteLock组合了两把锁:写锁WriteLock读锁ReadLock

我们来看看具体的构造函数:

1public ReentrantReadWriteLock(boolean fair) {
2  sync = fair ? new FairSync() : new NonfairSync();
3  readerLock = new ReadLock(this);
4  writerLock = new WriteLock(this);
5}

可以发现,通过参数fair控制是创建非公平锁还是公平锁。同时ReentrantReadWriteLock持有了写锁和读锁。

而本质上,读锁和写锁都是通过持有ReentrantReadWriteLock.sync来进行加锁和释放锁的,用的是同一个AQS,Sync类提供类对ReentrantReadWriteLock的支持

1protected ReadLock(ReentrantReadWriteLock lock) {
2  sync = lock.sync; // 引用的是ReentrantReadWriteLock的sync实例
3}
1protected WriteLock(ReentrantReadWriteLock lock) {
2  sync = lock.sync; // 引用的是ReentrantReadWriteLock的sync实例
3}

基于对AQS原理的理解,我们知道sync是读写锁实现的关键,而aqs中核心是state字段和双端等待队列。下面我们来看看具体的实现。

2.1、看代码之前您必须了解的内容

在查看ReentrantReadWriteLock之前,您需要了解以下内容:

2.1.1、Sync.HoldCounter类

读锁计数器类,为每个获取读锁的线程进行计数。Sync类中有一个cachedHoldCounter字段,该字段主要是缓存上一个线程的读锁计数器,节省ThreadLocal查找次数。

1static final class HoldCounter {
2  // 某个读线程的重入次数
3  int count = 0;
4  // 某个线程的tid字段
5  // Use id, not reference, to avoid garbage retention
6  final long tid = getThreadId(Thread.currentThread());
7}

2.1.2、Sync.ThreadLocalHoldCounter类

当前线程持有的可重入读锁的数量,当数量下降到0的时候进行删除。

1static final class ThreadLocalHoldCounter
2  extends ThreadLocal<HoldCounter
{
3  public HoldCounter initialValue() {
4    return new HoldCounter();
5  }
6}

2.1.3、读写锁中AQS的state状态设计

AQS中的state为了能够同时记录读锁和写锁的状态,把32位变量分为了两部分:

image-20200311222446270

如上图,高16位存储读状态,读锁是共享锁,这里记录持有读锁的线程数;低16位是写状态,写锁是排他锁,这里0表示没有线程持有,大于0表示持有线程对锁的重入次数。

2.1.4、关于读写锁的数据结构

虽然读写锁看起来有两把锁,但是底层用的都是同一个state,同一个等待队列。只不过是通过ReadLock和WriteLock分别提供了读锁和写锁的API,底层还是用同一个AQS。如下图:

image-20200312233429041

由于读写锁是互斥的,所以线程1获取写锁,线程2获取读锁,并发执行的时候,一定有一个会失败;

如果是已经获取了读锁的线程尝试获取写锁,则会获取成功;

公平模式下,先进入等待队列的线程先被处理;非公平模式下,如果尝试获取写锁的线程节点在头节点后面,尝试获取读锁的线程要让步,进入等待队列;

线程节点获取到读锁之后,会判断下一个节点是否处于共享模式,如果是则会一直传播并唤醒后续共享模式节点;

如果有其他线程获取了写锁,那么获取写锁就会被阻塞。

公平和非公平是针对等待队列中的线程节点的处理来说的:

  • 公平模式一般都是从队列头开始处理,并且如果等待队列还有待处理节点,新的线程全部都入等待队列

  • 非公平模式一般不管等待队列里面有没有待处理节点,都会先尝试竞争获取锁;特殊情况:如果等待队列中有写锁线程,那么新来的读锁线程必须排队让写锁线程先进行处理。

其实关于读写锁的原理就差不多是这么多了。

以下是详细的代码分析,可能会比较枯燥,为了避免让大家一头陷入源码中,于是在上面先把源码做的事情都给讲出来了。建议感兴趣的同学打开电脑跟踪源码一起来阅读。

2.2、ReadLock实现原理

2.2.1、lock

查看ReadLock的lock相关方法,调用的是AQS的acquireShared方法,该方法会以共享模式获取锁:

1public final void acquireShared(int arg) {
2  // 尝试获取锁
3  if (tryAcquireShared(arg) < 0)
4    // 如果获取锁失败了,那么会进入ASQ的等待队列,等待被唤醒后重新尝试获取锁
5    doAcquireShared(arg);
6}

下面看看关键获取锁的tryAcquireShared方法,该方法主要处理逻辑:

  • 因为读写是互斥的,如果另一个线程持有写锁,则失败;

  • 否则,此线程具备锁定write状态的条件,因此判断是否应该进入阻塞。如果不是,请尝试CAS获取读锁许可并更新读锁计数。请注意,该步骤不检查重入,这将推迟到最后fullTryAcquireShared方法;

  • 如果第2步失败,或者由于线程不符合锁定条件或者CAS失败或读锁计数饱和,将会使用fullTryAcquireShared进一步重试。

下面是详细的说明:

 1protected final int tryAcquireShared(int unused) {
2  Thread current = Thread.currentThread();
3  int c = getState();
4  // 如果存在写锁,并且写锁不是当前线程,则直接失败让线程进入等待队列
5  if (exclusiveCount(c) != 0 &&
6      getExclusiveOwnerThread() != current)
7    return -1;
8  int r = sharedCount(c);
9  // 判断读锁是否应该被阻塞,公平模式下,先进入等待队列则先被处理;非公平模式下写锁优先级比较高,如果头节点的下一个节点不是共享模式,即是尝试获取写锁的线程,读锁需要让步
10  if (!readerShouldBlock() &&
11      // 读锁是否已到达获取上线
12      r < MAX_COUNT &&
13      // CAS修改读锁状态,+1
14      compareAndSetState(c, c + SHARED_UNIT)) {
15      // 获取读锁成功
16      if (r == 0) {
17        // 如果是第一个获取读锁的线程,也就是把读锁状态从0变到1的那个线程,那么存入firstReader中
18        firstReader = current;
19        // firstReader持有锁=1
20        firstReaderHoldCount = 1;
21      } else if (firstReader == current) {
22        // firstReader已经是当前线程,则firstReaderHoldCount++
23        firstReaderHoldCount++;
24      } else { // 读锁数量不为0,并且第一个读线程不为当前线程
25        // 获取缓存读锁计数器
26        HoldCounter rh = cachedHoldCounter;
27        if (rh == null || rh.tid != getThreadId(current))
28          // 缓存读锁计数器为空或者计数器不是当前线程的,则尝试通过ThreadLocal获取当前线程对应的计数器
29          cachedHoldCounter = rh = readHolds.get();
30        else if (rh.count == 0)
31          readHolds.set(rh);
32        rh.count++;
33      }
34      return 1;
35    }
36    // 以上执行失败,则进入该逻辑
37    return fullTryAcquireShared(current);
38}

让我们接着看fullTryAcquireShared方法,这个方法可知,只有其他线程持有写锁,或者使用的是公平锁并且头节点后面还有其他等待的线程,或者头节点后面的节点不是共享模式,或者读锁计数器达到了上限,则阻塞,否则一直会循环尝试获取锁:

 1final int fullTryAcquireShared(Thread current) {
2  HoldCounter rh = null;
3  for (;;) {
4    int c = getState();
5    // 如果存在写锁,并且写锁不是当前线程,则返回false
6    if (exclusiveCount(c) != 0) {
7      if (getExclusiveOwnerThread() != current)
8        return -1;
9      // else we hold the exclusive lock; blocking here
10      // would cause deadlock.
11    // 不存在写锁,继续判断是否应该阻塞:如果是公平锁并且头节点后有其他等待的线程,则阻塞,如果是非公平锁,判断头节点后面的节点是否共享模式,如果不是则阻塞
12    } else if (readerShouldBlock()) {
13      // Make sure we're not acquiring read lock reentrantly
14      // 如果当前线程是firstReader,说明是重入
15      if (firstReader == current) {
16        // assert firstReaderHoldCount > 0;
17      } else {
18        // 进入该分支,说明没有读写锁冲突,并且不是重入,当前线程也不是firstReader
19        if (rh == null) {
20          rh = cachedHoldCounter;
21          // 判断上一个获取到锁的线程是否当前线程,不是则进入AQS等待队列
22          if (rh == null || rh.tid != getThreadId(current)) {
23            rh = readHolds.get();
24            if (rh.count == 0)
25              readHolds.remove();
26          }
27        }
28        // rh.count == 0 表示rh是刚新获取到的,直接返回,进入等待队列
29        if (rh.count == 0)
30          return -1;
31      }
32    }
33    // 共享锁达到上限了
34    if (sharedCount(c) == MAX_COUNT)
35      throw new Error("Maximum lock count exceeded");
36    // 读锁自增,以下代码与上一个方法中的类似
37    if (compareAndSetState(c, c + SHARED_UNIT)) {
38      if (sharedCount(c) == 0) {
39        firstReader = current;
40        firstReaderHoldCount = 1;
41      } else if (firstReader == current) {
42        firstReaderHoldCount++;
43      } else {
44        if (rh == null)
45          rh = cachedHoldCounter;
46        if (rh == null || rh.tid != getThreadId(current))
47          rh = readHolds.get();
48        else if (rh.count == 0)
49          readHolds.set(rh);
50        rh.count++;
51        cachedHoldCounter = rh; // cache for release
52      }
53      return 1;
54    }
55  }
56}

最后我们来看看doAcquireShared方法:

 1private void doAcquireShared(int arg) {
2  // 添加一个共享等待节点
3  final Node node = addWaiter(Node.SHARED);
4  boolean failed = true;
5  try {
6    boolean interrupted = false;
7    for (;;) {
8      // 判断新增的节点的前一个节点是否头节点
9      final Node p = node.predecessor();
10      if (p == head) { // 是头节点,那么在此尝试获取共享锁
11        int r = tryAcquireShared(arg);
12        if (r >= 0) { 
13          // 获取成功,把当前节点变为新的head节点,并且检查后续节点是否可以在共享模式下等待,并且允许继续传播,则调用doReleaseShared继续唤醒下一个节点尝试获取锁
14          setHeadAndPropagate(node, r);
15          p.next = null// help GC
16          if (interrupted)
17            selfInterrupt();
18          failed = false;
19          return;
20        }
21      }
22      // 阻塞节点
23      if (shouldParkAfterFailedAcquire(p, node) &&
24          parkAndCheckInterrupt())
25        interrupted = true;
26    }
27  } finally {
28    if (failed)
29      // 取消获取锁
30      cancelAcquire(node);
31  }
32}

2.2.2、unlock

接下来我们看看释放锁的代码。

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

AbstractQueuedSynchronizer.releaseShared()

1public final boolean releaseShared(int arg) {
2  if (tryReleaseShared(arg)) {
3    doReleaseShared();
4    return true;
5  }
6  return false;
7}

主要处理方法是tryReleaseShared,该方法主要是清理ThreadLocal中的锁计数器,然后CAS修改读锁个数减1:

 1protected final boolean tryReleaseShared(int unused) {
2  Thread current = Thread.currentThread();
3  if (firstReader == current) {
4    // assert firstReaderHoldCount > 0;
5    if (firstReaderHoldCount == 1)
6      firstReader = null;
7    else
8      firstReaderHoldCount--;
9  } else {
10    HoldCounter rh = cachedHoldCounter;
11    if (rh == null || rh.tid != getThreadId(current))
12      rh = readHolds.get();
13    int count = rh.count;
14    if (count <= 1) {
15      readHolds.remove();
16      if (count <= 0)
17        throw unmatchedUnlockException();
18    }
19    --rh.count;
20  }
21  for (;;) {
22    int c = getState();
23    int nextc = c - SHARED_UNIT;
24    if (compareAndSetState(c, nextc))
25      // Releasing the read lock has no effect on readers,
26      // but it may allow waiting writers to proceed if
27      // both read and write locks are now free.
28      return nextc == 0;
29  }
30}

2.3、WriteLock实现原理

2.3.1、lock

查看WriteLock的lock锁相关方法,调用的是sync.acquire方法,该方法直接继承了ASQ的acquire()方法的实现:

1public void lock() {
2    sync.acquire(1);
3}

与ReentrantLock的实现区别在具体的tryAcquire()方法的实现,我们来看看ReentrantReadWriteLock.Sync中该方法的实现,主要做了以下事情:

  • 如果读锁数量>0,或者写锁数量>0,并且不是重入的,那么直接失败了;

  • 如果锁数量为0,那么该线程有资格获取到写锁,进而尝试获取。

 1protected final boolean tryAcquire(int acquires) {
2  Thread current = Thread.currentThread();
3  int c = getState();
4  int w = exclusiveCount(c);
5  if (c != 0) { // 存在读锁或者写锁
6    // 不存在写锁,或者当前线程不是写锁持有的线程,那么直接失败
7    if (w == 0 || current != getExclusiveOwnerThread())
8      return false;
9    // 写锁超多最大数量限制,也直接失败
10    if (w + exclusiveCount(acquires) > MAX_COUNT)
11      throw new Error("Maximum lock count exceeded");
12    // Reentrant acquire
13    // 写锁持有的线程重入,直接修改state即可
14    setState(c + acquires);
15    return true;
16  }
17  // 判断是否应该阻塞:非公平模式,无需阻塞,公平模式如果前面有其他节点则需要排队阻塞
18  if (writerShouldBlock() ||
19      // 尝试获取写锁
20      !compareAndSetState(c, c + acquires))
21    return false;
22  setExclusiveOwnerThread(current);
23  return true;
24}

2.3.2、unlock

查看WriteLock的unlock相关方法,调用的是sync.release方法,该方法直接继承了AQS的release实现:

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

以下是release方法:

 1public final boolean release(int arg) {
2  // 尝试释放锁
3  if (tryRelease(arg)) {
4    // 释放锁成功,则唤醒队列中头节点后的一个线程
5    Node h = head;
6    if (h != null && h.waitStatus != 0)
7      unparkSuccessor(h);
8    return true;
9  }
10  return false;
11}

释放锁的逻辑主要在tryRelease方法,下面是详细代码:

 1protected final boolean tryRelease(int releases) {
2  // 如果当前线程没有获取写锁,则释放直接抛异常
3  if (!isHeldExclusively())
4    throw new IllegalMonitorStateException();
5  int nextc = getState() - releases;
6  boolean free = exclusiveCount(nextc) == 0;
7  // 如果当前线程完全释放了写锁,则去除独占标识
8  if (free)
9    setExclusiveOwnerThread(null);
10  // 修改state
11  setState(nextc);
12  return free;
13}

3、读写锁使用例子

下面是读写锁的使用例子,该例子实现了一个支持并发访问的ArrayList。

因为读写锁是互斥的,保证了不会因为写导致读取出现的不一致。

代码如下:

 1public class ReentrantReadWriteLockTest {
2
3    static final int READER_SIZE = 10;
4    static final int WRITER_SIZE = 2;
5
6    public static void main(String[] args) {
7        Integer[] initialElements = {33288699};
8
9        ReadWriteList<Integer> sharedList = new ReadWriteList<>(initialElements);
10
11        for (int i = 0; i < WRITER_SIZE; i++) {
12            new Writer(sharedList).start();
13        }
14
15        for (int i = 0; i < READER_SIZE; i++) {
16            new Reader(sharedList).start();
17        }
18
19    }
20
21}
22
23class Reader extends Thread {
24    private ReadWriteList<Integer> sharedList;
25
26    public Reader(ReadWriteList<Integer> sharedList) {
27        this.sharedList = sharedList;
28    }
29
30    public void run() {
31        Random random = new Random();
32        int index = random.nextInt(sharedList.size());
33        Integer number = sharedList.get(index);
34
35        System.out.println(getName() + " -> get: " + number);
36
37        try {
38            Thread.sleep(100);
39        } catch (InterruptedException ie ) { ie.printStackTrace(); }
40
41    }
42}
43
44class Writer extends Thread {
45    private ReadWriteList<Integer> sharedList;
46
47    public Writer(ReadWriteList<Integer> sharedList) {
48        this.sharedList = sharedList;
49    }
50
51    public void run() {
52        Random random = new Random();
53        int number = random.nextInt(100);
54        sharedList.add(number);
55
56        try {
57            Thread.sleep(100);
58            System.out.println(getName() + " -> put: " + number);
59        } catch (InterruptedException ie ) { ie.printStackTrace(); }
60    }
61}
62
63/**
64 * 支持并发读写的ArrayList
65 */

66class ReadWriteList<E{
67    private List<E> list = new ArrayList<>();
68    private ReadWriteLock rwLock = new ReentrantReadWriteLock();
69
70    public ReadWriteList(E... initialElements) {
71        list.addAll(Arrays.asList(initialElements));
72    }
73
74    public void add(E element) {
75        Lock writeLock = rwLock.writeLock();
76        writeLock.lock();
77
78        try {
79            list.add(element);
80        } finally {
81            writeLock.unlock();
82        }
83    }
84
85    public E get(int index) {
86        Lock readLock = rwLock.readLock();
87        readLock.lock();
88
89        try {
90            return list.get(index);
91        } finally {
92            readLock.unlock();
93        }
94    }
95
96    public int size() {
97        Lock readLock = rwLock.readLock();
98        readLock.lock();
99
100        try {
101            return list.size();
102        } finally {
103            readLock.unlock();
104        }
105    }
106
107}

4、ReentrantLock的Condition实现原理

接下来我们来ReentrantLock中的Condition实现原理。

有如下的ReentrantLock和Condition:

1// 锁和条件变量
2private final Lock lock = new ReentrantLock();
3// 条件
4private final Condition condition1 = lock.newCondition();

下面来看看执行await和signal的流程。

4.1、await等待

一般地,只有线程获取到lock之后,才可以使用condition的await方法。假设此时线程1获取到了ReentrantLock锁,在执行代码逻辑的时候,发现某些条件不符合,于是调用了以下代码:

1while(xxx条件不满足) {
2  condition1.await();
3}

此时AQS主要执行以下动作:

  1. 线程1把自己包装成节点,waitStatus设为CONDITION(-2),追加到ConditionObject中的条件队列(每个ConditionObject有一个自己的条件队列);

  2. 线程1释放锁,把state设置为0;

  3. 然后唤醒等待队列中head节点的下一个节点;

如下:

image-20200314121818770

接下来进入一个循环,如果判断到当前线程的节点不在等待队列,那么会一直让当前线程阻塞,代码如下:

1while (!isOnSyncQueue(node)) {
2  LockSupport.park(this);
3  if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
4    break;
5}

这个时候已经唤醒其他线程继续处理了,只有其他线程执行了condition1.signal或者condition1.signalAll之后,才会唤醒线程1进行处理后续的流程。

4.2、signal唤醒

当另一个线程执行了 condition1.signal之后,主要是做了以下事情:

  1. 把条件队列中的第一个节点追加到等待队列中;

  2. 把等待队列原来尾节点的waitStatus设置为SIGNAL。

image-20200314144725065

然后继续处理自己的事情,自己的事情处理完成之后,会释放锁,唤醒等待队列中head节点的下一个节点线程进行工作。

4.3、await恢复后继续执行

被唤醒的如果是之前执行了await方法的线程,那么该线程会接着就像往await方法里面阻塞处的下面继续执行,下面是源码:

 1// 如果当前节点不在等待队列,会一直进行阻塞
2while (!isOnSyncQueue(node)) {
3  LockSupport.park(this);
4  if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
5    break;
6}
7// 该方法主要做以下事情:
8// 1.尝试获取ReentrantLock锁
9// 2.获取成功,把现在线程节点变为新的head节点
10// 3. 否则根据继续休眠等待
11if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
12  interruptMode = REINTERRUPT;
13if (node.nextWaiter != null// 如果等待节点被取消了,那么从条件队列中移除
14  unlinkCancelledWaiters();
15if (interruptMode != 0)
16  reportInterruptAfterWait(interruptMode);

可以发现,这里主要是判断到当前线程节点已经放入等待队列了,那么会尝试获取锁,获取成功则继续往下执行代码。

第一节我们知道只有线程获取到ReentrantLock的锁之后才可以继续往下执行,中间可能会因为执行await而进入条件队列并释放锁,最后又会被唤醒重新获取锁,继续往下执行。最后按照书写规范,我们一定会在代码中执行ReentrantLock.unlock()释放锁,然后继续唤醒等待队列后续线程继续执行。

这篇文章的内容就差不多介绍到这里了,能够阅读到这里的朋友真的是很有耐心,为你点个赞。

本文为arthinking基于相关技术资料和官方文档撰写而成,确保内容的准确性,如果你发现了有何错漏之处,烦请高抬贵手帮忙指正,万分感激。

大家可以关注我的博客:itzhai.com 获取更多文章,我将持续更新后端相关技术,涉及JVM、Java基础、架构设计、网络编程、数据结构、数据库、算法、并发编程、分布式系统等相关内容。

点赞收藏
arthinking
请先登录,查看3条精彩评论吧
快去登录吧,你将获得
  • 浏览更多精彩评论
  • 和开发者讨论交流,共同进步
4
3