[Java知识库]读写锁加锁流程以及源码分析

读写锁用的是同一个 Sycn 同步器，因此等待队列、state 等也是同一个。流程与 ReentrantLock 加锁相比没有特殊之处，不同是写锁状态占了 state 的低 16 位，而读锁使用的是 state 的高 16 位。

读写锁接口结构

public interface ReadWriteLock {
	//读锁上锁
    Lock readLock();

	//写锁上锁
    Lock writeLock();
}

以下源码摘自ReentrantReadWriteLock

一、写锁上锁

static final class NonfairSync extends Sync {
	// 外部类 WriteLock 方法, 方便阅读, 放在此处
	public void lock() {
		sync.acquire(1);
	}

	// AQS 继承过来的方法, 方便阅读, 放在此处
	public final void acquire(int arg) {
		if (
			// 尝试获得写锁失败
			!tryAcquire(arg) &&
			// 将当前线程关联到一个 Node 对象上, 模式为独占模式
			// 进入 AQS 队列阻塞
			acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
		) {
			//进入队列后自我打断
			selfInterrupt();
		}
	}
	
	// Sync 继承过来的方法, 方便阅读, 放在此处
	protected final boolean tryAcquire(int acquires) {
		// 获得低 16 位, 代表写锁的 state 计数
		Thread current = Thread.currentThread();
		int c = getState();
		int w = exclusiveCount(c);//获取state低16位的值
		
		//c != 0 说明当前有锁
		if (c != 0) {
			if (
				// c != 0 and w == 0 表示有读锁, 或者
				w == 0 ||
				// 如果 exclusiveOwnerThread 不是自己（独占）
				current != getExclusiveOwnerThread()
			) {
				// 获得锁失败
				return false;
			}
	
			// 写锁计数（重入次数）超过低 16 位, 报异常
			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;
	}
	
	// 非公平锁 writerShouldBlock 总是返回 false, 无需阻塞
	final boolean writerShouldBlock() {
		return false;
	}
}

二、写锁释放

	static final class NonfairSync extends Sync {
	// WriteLock 方法, 方便阅读, 放在此处
	public void unlock() {
		sync.release(1);
	}
	
	// AQS 继承过来的方法, 方便阅读, 放在此处
	public final boolean release(int arg) {
		// 尝试释放写锁成功
		if (tryRelease(arg)) {
			// unpark AQS 中等待的线程
			Node h = head;
			if (h != null && h.waitStatus != 0)
				unparkSuccessor(h);
			return true;
		}
		return false;
	}

	// Sync 继承过来的方法, 方便阅读, 放在此处
	protected final boolean tryRelease(int releases) {
		if (!isHeldExclusively())
			throw new IllegalMonitorStateException();
		int nextc = getState() - releases;
		// 因为可重入的原因, 写锁计数为 0, 才算释放成功
		boolean free = exclusiveCount(nextc) == 0;
		if (free) {
			setExclusiveOwnerThread(null);
		}
		setState(nextc);
		return free;
	}
}

三、读锁上锁

static final class NonfairSync extends Sync {
	// ReadLock 方法, 方便阅读, 放在此处
	public void lock() {
		sync.acquireShared(1);
	}
	
	// AQS 继承过来的方法, 方便阅读, 放在此处
	public final void acquireShared(int arg) {
		// tryAcquireShared 返回负数, 表示获取读锁失败
		if (tryAcquireShared(arg) < 0) {
			doAcquireShared(arg);
		}
	}
 
	// Sync 继承过来的方法, 方便阅读, 放在此处
	protected final int tryAcquireShared(int unused) {
		Thread current = Thread.currentThread();
		int c = getState();
		// 如果是其它线程持有写锁, 获取读锁失败
		if ( 
			exclusiveCount(c) != 0 &&
			getExclusiveOwnerThread() != current
		) {
			return -1;
		}
		int r = sharedCount(c);
		if (
			// 读锁不该阻塞(如果老二是写锁，读锁该阻塞), 并且
			!readerShouldBlock() &&
			// 小于读锁计数, 并且
			r < MAX_COUNT &&
			// 尝试增加计数成功
			compareAndSetState(c, c + SHARED_UNIT)
		) {
			return 1;
		}
		return fullTryAcquireShared(current);
	}

	// 非公平锁 readerShouldBlock 看 AQS 队列中第一个节点是否是写锁
	// true 则该阻塞, false 则不阻塞
	final boolean readerShouldBlock() {
		return apparentlyFirstQueuedIsExclusive();
	}

	// AQS 继承过来的方法, 方便阅读, 放在此处
	// 与 tryAcquireShared 功能类似, 但会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞
	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 (sharedCount(c) == MAX_COUNT)
				throw new Error("Maximum lock count exceeded");
			if (compareAndSetState(c, c + SHARED_UNIT)) {
				// ... 省略不重要的代码
				return 1;
			}
		}
	}
 
	// AQS 继承过来的方法, 方便阅读, 放在此处
	private void doAcquireShared(int arg) {
		// 将当前线程关联到一个 Node 对象上, 模式为共享模式
		final Node node = addWaiter(Node.SHARED);
		boolean failed = true;
		try {
			boolean interrupted = false;
			for (;;) {
				final Node p = node.predecessor();
				if (p == head) {
					// 再一次尝试获取读锁
					int r = tryAcquireShared(arg);
					// 成功
					if (r >= 0) {
						// r 表示可用资源数, 在这里总是 1 允许传播
						//（唤醒 AQS 中下一个 Share 节点）
						setHeadAndPropagate(node, r);
						p.next = null; // help GC
						if (interrupted)
							selfInterrupt();
						failed = false;
						return;
					}
				}
				if (
				// 是否在获取读锁失败时阻塞（前一个阶段 waitStatus == Node.SIGNAL）
				shouldParkAfterFailedAcquire(p, node) &&
					// park 当前线程
					parkAndCheckInterrupt()
					) {
						interrupted = true;
					}
			}
		} finally {
			if (failed)
				cancelAcquire(node);
		}
	}
 
	//AQS 继承过来的方法, 方便阅读, 放在此处
	private void setHeadAndPropagate(Node node, int propagate) {
		Node h = head; // Record old head for check below
		// 设置自己为 head
		setHead(node);
	
		// propagate 表示有共享资源（例如共享读锁或信号量）
		// 原 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
		// 现在 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
		if (propagate > 0 || h == null || h.waitStatus < 0 ||
		(h = head) == null || h.waitStatus < 0) {
			Node s = node.next;
			// 如果是最后一个节点或者是等待共享读锁的节点
			if (s == null || s.isShared()) {
				doReleaseShared();
			}
		}
	}
 
	//AQS 继承过来的方法, 方便阅读, 放在此处
	private void doReleaseShared() {
	// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
	// 如果 head.waitStatus == 0 ==> Node.PROPAGATE, 为了解决 bug, 见后面分析
	for (;;) {
		Node h = head;
		// 队列还有节点
		if (h != null && h != tail) {
			int ws = h.waitStatus;
			if (ws == Node.SIGNAL) {
				if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
					continue; // loop to recheck cases
				// 下一个节点 unpark 如果成功获取读锁
				// 并且下下个节点还是 shared, 继续 doReleaseShared
				unparkSuccessor(h);
		}
		else if (ws == 0 &&
		!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
			continue; // loop on failed CAS
		}
		if (h == head) // loop if head changed
			break;
		}
	}
}

四、读锁释放

static final class NonfairSync extends Sync {
	// ReadLock 方法, 方便阅读, 放在此处
	public void unlock() {
		sync.releaseShared(1);
	}
	
	// AQS 继承过来的方法, 方便阅读, 放在此处
	public final boolean releaseShared(int arg) {
		if (tryReleaseShared(arg)) {
			doReleaseShared();
			return true;
		}
		return false;
	}
	
	// Sync 继承过来的方法, 方便阅读, 放在此处
	protected final boolean tryReleaseShared(int unused) {
		// ... 省略不重要的代码
		for (;;) {
			int c = getState();
			int nextc = c - SHARED_UNIT;
			if (compareAndSetState(c, nextc)) {
				// 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程
				// 计数为 0 才是真正释放
				return nextc == 0;
			}
		}
	}
	
	// AQS 继承过来的方法, 方便阅读, 放在此处
	private void doReleaseShared() {
	// 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
	// 如果 head.waitStatus == 0 ==> Node.PROPAGATE 
	for (;;) {
		Node h = head;
		if (h != null && h != tail) {
			int ws = h.waitStatus;
			// 如果有其它线程也在释放读锁，那么需要将 waitStatus 先改为 0
			// 防止 unparkSuccessor 被多次执行
			if (ws == Node.SIGNAL) {
				if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
				continue; // loop to recheck cases
				unparkSuccessor(h);
			}
			// 如果已经是 0 了，改为 -3，用来解决传播性，见后文信号量 bug 分析
			else if (ws == 0 &&
			!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
				continue; // loop on failed CAS
			}
			if (h == head) // loop if head changed
				break;
		}
	} 
}