网站建设的解决方案,南京网站搜索排名,厦门建设,cms网站建站流程概述
ReentrantReadWriteLock维护了一对相关的锁#xff0c;它们分别是共享readLock和独占writeLock。关于共享读锁和排他写锁的概念其实很好理解。所谓共享读锁就是一个线程读的时候#xff0c;其它线程也可以来读#xff08;共享#xff09;#xff0c;但是不能来写。排…概述
ReentrantReadWriteLock维护了一对相关的锁它们分别是共享readLock和独占writeLock。关于共享读锁和排他写锁的概念其实很好理解。所谓共享读锁就是一个线程读的时候其它线程也可以来读共享但是不能来写。排他写锁是指一个线程在写的时候其它线程不能来写或读排他。除了这个特点之外ReentrantReadWriteLock还有一个特点就是可重入的。它和ReentrantLock一样都是支持Condition的。而且ReentrantReadWerite还支持锁降级即允许将写锁降级为读锁。
简单使用
最最基础的用法如下 ReentrantReadWriteLock locknew ReentrantReadWriteLock();public void read(){lock.readLock().lock();//需要加读锁的操作lock.readLock().unlock();}public void write(){lock.writeLock().lock();//需要加写锁的操作lock.writeLock().unlock();}
ReentrantReadWriteLock无非就是这几种情况读读共享写写互斥读写互斥写读互斥。
下面我们就以这个最基础的用法来分析一下其内部的原理
源码分析
继承体系 共享读锁的实现原理分析#
lock方法#
首先进入调用具体的实现 public void lock() {sync.acquireShared(1);}然后调用了这个方法public final void acquireShared(int arg) { if (tryAcquireShared(arg) 0) doAcquireShared(arg); }
其中int tryAcquireShared(int unused)的具体实现如下 protected final int tryAcquireShared(int unused) {/** Walkthrough:* 1. If write lock held by another thread, fail.* 2. Otherwise, this thread is eligible for* lock wrt state, so ask if it should block* because of queue policy. If not, try* to grant by CASing state and updating count.* Note that step does not check for reentrant* acquires, which is postponed to full version* to avoid having to check hold count in* the more typical non-reentrant case.* 3. If step 2 fails either because thread* apparently not eligible or CAS fails or count* saturated, chain to version with full retry loop.*/Thread current Thread.currentThread();int c getState();//持有写锁的线程可以获取读锁如果获取锁的线程不是当前线程则返回-1if (exclusiveCount(c) ! 0 getExclusiveOwnerThread() ! current)return -1;int r sharedCount(c);//获取共享读锁的数量if (!readerShouldBlock() r MAX_COUNT compareAndSetState(c, c SHARED_UNIT)) {if (r 0) {//如果首次获取锁则初始化firstReader和firstReaderHoldCountfirstReader current;firstReaderHoldCount 1;} else if (firstReader current) {//如果当前线程是首次获取读锁的线程firstReaderHoldCount;} else {//更新HoldCounterHoldCounter 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体现写锁的排他性.如果在尝试获取锁是不需要阻塞等待由锁的公平性决定并且读锁的共享计数小于最大值那么就直接通过CAS更新读锁数量获取读锁。如果第二步执行失败了那么就会调用fullTryAcquireShared(current)
fullTryAcquireShared(current)的具体实现如下
final int fullTryAcquireShared(Thread current) {/** This code is in part redundant with that in* tryAcquireShared but is simpler overall by not* complicating tryAcquireShared with interactions between* retries and lazily reading hold counts.*/HoldCounter rh null;for (;;) { //自旋int c getState();if (exclusiveCount(c) ! 0) { //写锁已经被持有了if (getExclusiveOwnerThread() ! current) //持有写锁的不是单线程return -1; //其它线程持有读锁后就不能在获取写锁了// else we hold the exclusive lock; blocking here// would cause deadlock.} else if (readerShouldBlock()) {//由公平性决定需要阻塞// Make sure were not acquiring read lock reentrantlyif (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)//如果当前线程的持有读锁数为0那么就没必要使用计数器直接移除readHolds.remove();}}if (rh.count 0)return -1;}}if (sharedCount(c) MAX_COUNT) //如果读锁的数量超过最大值了throw new Error(Maximum lock count exceeded);if (compareAndSetState(c, c SHARED_UNIT)) { //CAS更新读锁数量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;}}} 可以看出其实int fullTryAcquireShared(Thread current)也每什么特别它的代码和int tryAcquireShared(int unused)差不多。只不过是增加了自旋重试和“持有读锁数的延迟读取”
我们回到void acquireShared(int arg)方法如果tryAcquireShared(arg)获取读锁失败后它调用的doAcquireShared(arg)又做了什么呢 它的具体实现如下 private void doAcquireShared(int arg) {final Node node addWaiter(Node.SHARED); //添加一个共享模式的Node到等待队列尾部boolean failed true;try {boolean interrupted false; //获取前驱节点for (;;) {final Node p node.predecessor();if (p head) {//如果前驱节点尝试获取资源int r tryAcquireShared(arg);if (r 0) {//获取成功更新等待队列并唤醒下一个等待的节点setHeadAndPropagate(node, r);p.next null; // help GCif (interrupted)selfInterrupt();failed false;return;}}if (shouldParkAfterFailedAcquire(p, node) //检查获取失败后是否可以阻塞parkAndCheckInterrupt())interrupted true;}} finally {if (failed)cancelAcquire(node);}} 其实整个获取共享读锁的源码看下来我们可以发现AQS框架下获取锁一般的流程就是首先尝试去直接获取如果获取不到了那么尝试自旋获取如果还是获取不到那么就去等待队列排队排队的时候如果发现自己是第二个那么就再次尝试获取锁如果还是没获取到那么就老老实实的在等待队列中park阻塞等待了。 我们通过源码也可发现AQS框架下的锁其实如果线程之间对锁的争用很低的时候大多数时候直接就能拿到锁几乎不需要排队阻塞之类的性能非常之高。 unlock方法 第一步还是调用具体的实现、 public void unlock() {sync.releaseShared(1);} 具体的实现如下 public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;} 首先来看tryReleaseShared(arg) protected final boolean tryReleaseShared(int unused) {Thread current Thread.currentThread();if (firstReader current) { //如过当前线程是第一获取到读锁的线程// assert firstReaderHoldCount 0;//直接更新线程持有数if (firstReaderHoldCount 1)firstReader null;elsefirstReaderHoldCount--;} 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;}for (;;) { //自旋int c getState();int nextc c - SHARED_UNIT;if (compareAndSetState(c, nextc)) //更新state// Releasing the read lock has no effect on readers,// but it may allow waiting writers to proceed if// both read and write locks are now free.return nextc 0;}} 我们从tryReleaseShared(arg)的实现中可以看出它的主要是去更新锁计数器和state。如果state为0的话就返回true否则就返回false。 我们回过头看如果tryReleaseShared(arg)返回true即锁释放后state为0了那么它会执行doReleaseShared();方法它的具体实现如下 private void doReleaseShared() {/** Ensure that a release propagates, even if there are other* in-progress acquires/releases. This proceeds in the usual* way of trying to unparkSuccessor of head if it needs* signal. But if it does not, status is set to PROPAGATE to* ensure that upon release, propagation continues.* Additionally, we must loop in case a new node is added* while we are doing this. Also, unlike other uses of* unparkSuccessor, we need to know if CAS to reset status* fails, if so rechecking.*/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 casesunparkSuccessor(h);}else if (ws 0 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue; // loop on failed CAS}if (h head) // loop if head changedbreak;}} 这个方法的作用就是唤醒等待队列中线程现在资源已经空闲了等待的线程可以唤醒来获取锁了。 排他写锁的实现原理分析# 排他写锁的实现原理其实和ReentrantLock一致。我们只看几处和共享读锁不同的地方。 //公平锁实现
protected final boolean tryAcquire(int acquires) {final Thread current Thread.currentThread();int c getState();if (c 0) {if (!hasQueuedPredecessors() //判断当前线程是否还有前节点compareAndSetState(0, acquires)) {//CAS修改state//获取锁成功设置锁的持有线程为当前线程setExclusiveOwnerThread(current);return true;}}else if (current getExclusiveOwnerThread()) {//该线程之前已经拿到锁int nextc c acquires; //重入的体现if (nextc 0)throw new Error(Maximum lock count exceeded);setState(nextc); //更新Statereturn true;}return false;} 其实非公平锁的实现也差不多只不过少了!hasQueuedPredecessors()它不会去判断当前线程是否还有前驱节点直接就开始获取锁了。 unlock方法也差不多我就不赘述了。
