Java并发研究 自己写ReentrantLock和ReentrantReadWriteLock（4）
接上篇。在写完ReentrantLock之后，其实可以基于ReentrantLock写一个ReadWriteLock，《the art of multiprocessor programming》第八章有介绍。但是，本着不完全AQS（AbstractQueuedSynchronizer）介绍的系列主题，这里从零开始重新写一个ReentrantReadWriteLock。
按照ReadWriteLock的定义，任何时候都满足
- 没有线程持有锁
- 有1~n个线程持有共享锁（Read）
- 有1个线程持有独占锁（Write）
中的一个。
其次公平的ReadWriteLock要求新来的Read或者Write线程必须在队列中等待，非公平的ReadWriteLock允许新来的Read或者Write比队列中等待的线程先获取锁。关于非公平锁这里多说一句，理论上的非公平锁类似一群人哄抢的现象，但是实现多半是只允许新来和线程队列最前面的线程抢占锁。ReadWriteLock也是一样。如果你想要完全非公平的锁的话，可能AQS和这里的实现不满足你的需求。
为了实现ReadWriteLock的定义，你需要分别记录读写状态。考虑到独占（Write）状态只可能有一个线程，可能场景如下：
| No. | Reader Count | Writer Count |
|---|---|---|
| 1 | 0 | 0 |
| 2 | 1+ | 0 |
| 3 | 0 | 1 |
可以看到，从没有任何线程持有锁的状态1变成2或者3时，reader count和writer count分别变化，这个特性导致你必须同时CAS reader count和writer count，或者混合成一个变量，分别用低位和高位来区别reader count和writer count（或者用正数和负数也可以）。AQS采用的是后者。这里简化一下，并给几个简单的关系式：
count = 00000000 reader_count = count & 0x00FF writer_count = count & 0xFF00
还有一个要考虑的问题是重入（reentrant）如何实现。你可以增加reader count，或者和ReentrantLock一样每个读线程自己维护一个thread local变量。从效率上来说，后者比较好。
SimpleReadWriteLock
只有tryLock的基本实现
import javax.annotation.Nonnull;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
public class SimpleReadWriteLock implements ReadWriteLock {
private static final int WRITER_MASK = 0xFF00;
private static final int WRITER_UNIT = 0x0100;
// writer reentrant times, reader count
private final AtomicInteger count = new AtomicInteger(0);
private final ReadLock readLock = new ReadLock();
private final WriteLock writeLock = new WriteLock();
@Override
@Nonnull
public Lock readLock() {
return readLock;
}
@Override
@Nonnull
public Lock writeLock() {
return writeLock;
}
private static abstract class AbstractLock implements Lock {
@Override
public void lock() {
throw new UnsupportedOperationException();
}
@Override
public void lockInterruptibly() throws InterruptedException {
throw new UnsupportedOperationException();
}
@Override
public boolean tryLock(long time, @Nonnull TimeUnit unit) throws InterruptedException {
throw new UnsupportedOperationException();
}
@Override
@Nonnull
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}
@SuppressWarnings("Duplicates")
private class ReadLock extends AbstractLock {
// reentrant times of current reader
private final ThreadLocal<Integer> reentrantTimes = ThreadLocal.withInitial(() -> 0);
@Override
public boolean tryLock() {
int rt = reentrantTimes.get();
if (rt > 0) {
reentrantTimes.set(rt + 1);
return true;
}
int c = count.get();
if (((c & WRITER_MASK) == 0) && count.compareAndSet(c, c + 1)) {
reentrantTimes.set(1);
return true;
}
return false;
}
@Override
public void unlock() {
int rt = reentrantTimes.get();
if (rt <= 0) {
throw new IllegalStateException("attempt to unlock without holding lock");
}
if (rt > 1) {
reentrantTimes.set(rt - 1);
return;
}
reentrantTimes.set(0);
if (count.get() < 1) {
throw new IllegalStateException("no reentrantTimes");
}
count.decrementAndGet();
}
}
@SuppressWarnings("Duplicates")
private class WriteLock extends AbstractLock {
private Thread writer;
@Override
public boolean tryLock() {
if (writer == Thread.currentThread()) {
count.getAndAdd(WRITER_UNIT);
return true;
}
if (count.get() == 0 && count.compareAndSet(0, WRITER_UNIT)) {
writer = Thread.currentThread();
return true;
}
return false;
}
@Override
public void unlock() {
if (writer != Thread.currentThread()) {
throw new IllegalStateException("attempt to unlock without holding lock");
}
int c = count.get();
if (c < WRITER_UNIT) {
throw new IllegalStateException("no writer");
}
if (c == WRITER_UNIT) {
writer = null;
}
count.set(c - WRITER_UNIT);
}
}
}
重点关注其中的CAS操作是如何处理count的。代码这里不具体分析了，相信看过ReentrantLock分析的话，这里也不会太难。
接下来考虑如何加入队列。实现可能有两种选择
- 和Write一样，每个Read作为单独一个节点
- 连续的Read单独作为一个节点
后者看起来不错，因为头节点（这里指Write节点）在唤醒后续节点的时候只需要唤醒一个就行了，否则要唤醒多个连续的Read节点。但是实际上AQS选择的是第一种！原因是什么？个人认为第二种在唤醒上和第一种在本质上没有太大差别：唤醒后续的多个Read节点和唤醒后续的单个Read节点后再唤醒这个Read节点里其他的Read线程。于此同时带来了一系列细节问题，入队时有可能直接入队（最后的节点是Write节点时），也有可能跟随最后的一个Read节点，还要考虑唤醒丢失的问题等等。总之，得不偿失。所以个人的渐进式实现也是用第一种方法。
在唤醒后续节点上，考虑以下几种情况
第一种是只有Write的情况，必然只需要唤醒一个后继节点。第二种只有一个后继Read节点，所以也只要唤醒一个后继节点。但是第三种情况下，首节点可能要唤醒两个后续节点。第四种，Read线程作为首节点，只需要唤醒后续第一个Write节点。第五种情况比较特殊：这里表示连续唤醒的中途。
从图中可以看出，将一个Read线程作为一个节点的话，唤醒必须是多次的。更准确一点来说，如果后继节点是Read节点，那么连续唤醒，否则（Write节点），只唤醒一个。
再考虑一个问题，在哪个步骤唤醒？
- unlock
- 前一个Read节点获取了锁之后
以及谁唤醒谁？
- unlock的节点唤醒后续节点
- Read节点唤醒后续节点
这里其实没有2×2总共4种设计，实际可行的主要就以上两种。
第一种在unlock时唤醒后续的Read节点，注意只唤醒一个。然后Read节点在获取了锁之后唤醒后续的一个Read节点，依此类推。第二种在unlock时唤醒后续所有的Read节点。
AQS选择了第一种。原因是什么？个人认为原因是保证能够获取锁时才唤醒。假设使用第二种设计，第一个Read节点被唤醒，可以获取锁了，但是还没推进队列（acquire和release时队列的处理请参见之前ReentrantLock的介绍），此时W继续唤醒第二个Read节点时，由于第二个Read节点的前置节点，即第一个Read节点还没有成为head，所以只能继续park，等于无效唤醒。但是采用第一种设计的话，第一个Read节点只有在获取了锁之后，设置自己为head，接着唤醒后续Read节点，此时第二个Read节点肯定能够获取锁，所以不会是一次无效唤醒。
注意，第一种方法的图中，节点上方的唤醒和unlock时的唤醒是不同的，上方的唤醒只针对Read节点，也就是说，后继节点是Read节点才会唤醒，相对的，unlock无条件唤醒后续的单个节点。
你可能注意到，上述方法中并没有提到Node的PROPAGATE，即使看上去Read节点的连续唤醒很像传播。原因是ReadWriteLock不需要PROPAGATE，其次PROPAGATE可以说是针对基于AQS的Semaphore的解决方案。这里大致分析一下Semaphore没有PROPAGATE的话可能存在的问题。
对于Semaphore来说，主要有两种操作
- acquire
- release
其中release可能会唤醒多个等待acquire的线程，所以采用的是类似ReadLock，即没有WriteLock的ReadWriteLock模型。acquire方法对应AQS的acquireShared，release方法对应releaseShared。
- 假设一个Semaphore有permits为2，线程1持有1，线程2持有1，线程3等待中，线程4等待中。此时的队列类似上图。head这里不重要，可能是sentinel节点，也有可能是线程1或者2。两个waiter分别是线程3和4
- 线程1使用完permit，通过releaseShared返还。这时head唤醒第一个waiter，即线程3。线程3发现Semaphore中有一个permit，成功获取，同时因为只有1个permit，线程4不决定唤醒线程4（准确来说是tryAcquireShared返回的propagate为0）。此时线程3还未推进队列
- 线程2使用完permit，通过releaseShared返还。这时head没有变化，同时按照ReentrantLock一样的方式，如果唤醒过了后续节点的话，不会再尝试唤醒。即线程3不会被重复唤醒
- 线程3继续执行，没有唤醒后续节点4，同时permits保留1，但是线程4无法获取
这里个人认为一个直接的解决方法是，在唤醒上使用之前的unlock时连续唤醒，同时首节点无条件唤醒后续所有节点（不检查自己的status）。当然这样做的话无效唤醒的可能性会比较大。另一种方法是线程1返还permit时，确保在下一次release之前线程3成为首节点。老实说这比较难。
AQS的解决方法主要是
- 在推进队列时检查head的status（setHeadAndPropagate）
- releaseShared是从0改成PROPAGATE（doReleaseShared）
这样，上述场景的线程3发现head为PROPAGATE，并唤醒后续节点。
由于加入了PROPAGATE，一部分代码需要针对这个新状态做额外处理。但是在ReadWriteLock中，不存在这个问题：Read节点会无条件唤醒后续节点。
最终代码
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReadWriteLock;
@SuppressWarnings("Duplicates")
public class UnfairReadWriteTimedLock3 implements ReadWriteLock {
// private static final int READER_MASK = 0x00FF;
private static final int WRITER_MASK = 0xFF00;
private static final int WRITER_UNIT = 0x0100;
private final AtomicInteger count = new AtomicInteger(0);
private final Queue queue = new Queue();
private final ReadLock readLock = new ReadLock();
private final WriteLock writeLock = new WriteLock();
@Override
@Nonnull
public Lock readLock() {
return readLock;
}
@Override
@Nonnull
public Lock writeLock() {
return writeLock;
}
private static abstract class AbstractLock implements Lock {
@Override
@Nonnull
public Condition newCondition() {
throw new UnsupportedOperationException();
}
}
@SuppressWarnings("Duplicates")
private class ReadLock extends AbstractLock {
private final ThreadLocal<Integer> reentrantTimes = ThreadLocal.withInitial(() -> 0);
@Override
public void lock() {
if (tryLock()) {
return;
}
int c;
final Node node = new Node(Thread.currentThread(), true);
queue.enqueue(node);
while (true) {
if (node.predecessor.get() == queue.head.get()) {
c = count.get();
if ((c & WRITER_MASK) == 0 && count.compareAndSet(c, c + 1)) {
myTurn(node);
return;
}
}
if (isReadyToPark(node)) {
LockSupport.park(this);
}
}
}
@Override
public void lockInterruptibly() throws InterruptedException {
if (tryLock()) {
return;
}
int c;
final Node node = new Node(Thread.currentThread(), true);
Node predecessor = queue.enqueue(node);
while (true) {
if (predecessor == queue.head.get()) {
c = count.get();
if ((c & WRITER_MASK) == 0 && count.compareAndSet(c, c + 1)) {
myTurn(node);
return;
}
}
if (isReadyToPark(node)) {
LockSupport.park(this);
}
if (Thread.interrupted()) {
abort(node);
throw new InterruptedException();
}
}
}
@Override
public boolean tryLock() {
int rt = reentrantTimes.get();
if (rt > 0) {
reentrantTimes.set(rt + 1);
return true;
}
int c = count.get();
if ((c & WRITER_MASK) == 0 && count.compareAndSet(c, c + 1)) {
reentrantTimes.set(1);
return true;
}
return false;
}
@Override
public boolean tryLock(long time, @Nonnull TimeUnit unit) throws InterruptedException {
if (tryLock()) {
return true;
}
final long deadline = System.nanoTime() + unit.toNanos(time);
final Node node = new Node(Thread.currentThread(), true);
Node predecessor = queue.enqueue(node);
long nanos;
int c;
while (true) {
if (predecessor == queue.head.get()) {
c = count.get();
if ((c & WRITER_MASK) == 0 && count.compareAndSet(c, c + 1)) {
myTurn(node);
return true;
}
}
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
abort(node);
return false;
}
if (isReadyToPark(node)) {
LockSupport.parkNanos(this, nanos);
}
if (Thread.interrupted()) {
abort(node);
throw new InterruptedException();
}
}
}
private void myTurn(@Nonnull Node node) {
reentrantTimes.set(1);
node.clearThread();
queue.head.set(node);
/*
* propagate if successor is reader
*
* In ReadWriteLock, there's no need to check if propagate, it always propagates.
*/
if (node.resetSignalStatus()) {
Node successor = queue.findNormalSuccessor(node);
if (successor != null && successor.shared) {
LockSupport.unpark(successor.thread.get());
}
}
}
@Override
public void unlock() {
int rt = reentrantTimes.get();
if (rt < 1) {
throw new IllegalStateException("not the thread holding lock");
}
if (rt > 1) {
reentrantTimes.set(rt - 1);
return;
}
// rt == 1
reentrantTimes.set(0);
if (count.get() < 1) {
throw new IllegalStateException("count < 1");
}
if (count.decrementAndGet() > 0) {
return;
}
Node h = queue.head.get();
if (h != null && h.resetSignalStatus()) {
unparkNormalSuccessor(h);
}
}
}
@SuppressWarnings("Duplicates")
private class WriteLock extends AbstractLock {
private Thread owner;
@Override
public void lock() {
if (tryLock()) {
return;
}
Node node = new Node(Thread.currentThread());
Node predecessor = queue.enqueue(node);
while (true) {
if (predecessor == queue.head.get() &&
count.get() == 0 && count.compareAndSet(0, WRITER_UNIT)) {
myTurn(node);
return;
}
if (isReadyToPark(node)) {
LockSupport.park(this);
}
}
}
@Override
public void lockInterruptibly() throws InterruptedException {
if (tryLock()) {
return;
}
Node node = new Node(Thread.currentThread());
Node predecessor = queue.enqueue(node);
while (true) {
if (predecessor == queue.head.get() &&
count.get() == 0 && count.compareAndSet(0, WRITER_UNIT)) {
myTurn(node);
return;
}
if (isReadyToPark(node)) {
LockSupport.park(this);
}
if (Thread.interrupted()) {
abort(node);
throw new InterruptedException();
}
}
}
@Override
public boolean tryLock() {
if (owner == Thread.currentThread()) {
count.getAndAdd(WRITER_UNIT);
return true;
}
if (count.get() == 0 && count.compareAndSet(0, WRITER_UNIT)) {
owner = Thread.currentThread();
return true;
}
return false;
}
@Override
public boolean tryLock(long time, @Nonnull TimeUnit unit) throws InterruptedException {
if (tryLock()) {
return true;
}
long deadline = System.nanoTime() + unit.toNanos(time);
Node node = new Node(Thread.currentThread());
Node predecessor = queue.enqueue(node);
long nanos;
while (true) {
if (predecessor == queue.head.get() &&
count.get() == 0 && count.compareAndSet(0, WRITER_UNIT)) {
myTurn(node);
return true;
}
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
abort(node);
return false;
}
if (isReadyToPark(node)) {
LockSupport.parkNanos(this, nanos);
}
if (Thread.interrupted()) {
abort(node);
throw new InterruptedException();
}
}
}
private void myTurn(@Nonnull Node node) {
node.clearThread();
owner = Thread.currentThread();
queue.head.set(node);
}
@Override
public void unlock() {
if (owner != Thread.currentThread()) {
throw new IllegalStateException("not the thread holding write lock");
}
int c = count.get();
if (c < WRITER_UNIT) {
throw new IllegalStateException("no writer");
}
if (c > WRITER_UNIT) {
count.set(c - WRITER_UNIT);
return;
}
// c == WRITER_UNIT
owner = null;
// linearization point
count.set(0);
// signal successor
Node node = queue.head.get();
if (node != null && node.status.get() == Node.STATUS_SIGNAL) {
node.status.set(Node.STATUS_NORMAL);
unparkNormalSuccessor(node);
}
}
}
private boolean isReadyToPark(@Nonnull Node node) {
Node predecessor = node.predecessor.get();
int s = predecessor.status.get();
if (s == Node.STATUS_SIGNAL) {
return true;
}
if (s == Node.STATUS_ABORTED) {
predecessor = queue.skipAbortedPredecessor(node);
predecessor.successor.set(node);
} else {
predecessor.status.compareAndSet(Node.STATUS_NORMAL, Node.STATUS_SIGNAL);
}
return false;
}
private void abort(@Nonnull Node node) {
node.clearThread();
Node p = queue.skipAbortedPredecessor(node);
Node ps = p.successor.get();
node.status.set(Node.STATUS_ABORTED);
Node t = queue.tail.get();
if (t == node && queue.tail.compareAndSet(t, p)) {
p.successor.compareAndSet(ps, null);
return;
}
if (p != queue.head.get() && p.ensureSignalStatus() && p.thread.get() != null) {
Node s = node.successor.get();
if (s != null && s.status.get() != Node.STATUS_ABORTED) {
p.successor.compareAndSet(ps, s);
}
} else {
node.resetSignalStatus();
unparkNormalSuccessor(node);
}
}
private void unparkNormalSuccessor(@Nonnull Node node) {
Node successor = queue.findNormalSuccessor(node);
if (successor != null) {
LockSupport.unpark(successor.thread.get());
}
}
@SuppressWarnings("Duplicates")
private static class Queue {
final AtomicReference<Node> head = new AtomicReference<>();
final AtomicReference<Node> tail = new AtomicReference<>();
@Nonnull
Node enqueue(@Nonnull Node node) {
Node t;
while (true) {
t = tail.get();
if (t == null) {
Node sentinel = new Node();
if (head.compareAndSet(null, sentinel)) {
tail.set(sentinel);
}
} else {
node.predecessor.lazySet(t);
if (tail.compareAndSet(t, node)) {
t.successor.set(node);
return t;
}
}
}
}
@Nullable
Node findNormalSuccessor(@Nonnull Node node) {
Node s = node.successor.get();
if (s != null && s.status.get() != Node.STATUS_ABORTED) {
return s;
}
// find from tail
s = null;
Node c = tail.get();
while (c != null && c != node) {
if (c.status.get() != Node.STATUS_ABORTED) {
s = c;
}
c = c.predecessor.get();
}
return s;
}
@Nonnull
Node skipAbortedPredecessor(@Nonnull Node node) {
Node h = head.get();
Node p = node.predecessor.get();
while (p != h && p.status.get() == Node.STATUS_ABORTED) {
p = p.predecessor.get();
node.predecessor.set(p);
}
return p;
}
}
private static class Node {
static final int STATUS_NORMAL = 0;
static final int STATUS_SIGNAL = 1;
static final int STATUS_ABORTED = -1;
final AtomicReference<Thread> thread;
final boolean shared;
final AtomicInteger status = new AtomicInteger(STATUS_NORMAL);
final AtomicReference<Node> predecessor = new AtomicReference<>();
// optimization
final AtomicReference<Node> successor = new AtomicReference<>();
Node() {
this(null, false);
}
Node(@Nullable Thread thread) {
this(thread, false);
}
Node(@Nullable Thread thread, boolean shared) {
this.thread = new AtomicReference<>(thread);
this.shared = shared;
}
void clearThread() {
thread.set(null);
}
/**
* Ensure signal status.
* If current status is signal, just return.
* If current status is normal, then try to CAS status from normal to signal.
*
* @return true if changed to signal, otherwise false
*/
boolean ensureSignalStatus() {
int s = status.get();
return s == STATUS_SIGNAL || (s == STATUS_NORMAL && status.compareAndSet(STATUS_NORMAL, STATUS_SIGNAL));
}
/**
* Reset signal status.
* SIGNAL -> NORMAL
*
* @return true if successful, otherwise false
*/
boolean resetSignalStatus() {
return status.get() == STATUS_SIGNAL && status.compareAndSet(STATUS_SIGNAL, STATUS_NORMAL);
}
}
}
总结
“自己写ReentrantLock和ReentrantReadWriteLock“系列至此结束。总得来说，并发编程需要仔细思考很多情况，包括各种细节，有些时候细节会导致你的设计有所变化。希望这个系列对你学习AQS，学习Java语言下的并发编程有用，以及欢迎交流交换意见。
正文到此结束
热门推荐
相关文章
Loading...
![[HBLOG]公众号](http://www.liuhaihua.cn/img/qrcode_gzh.jpg)
