从读写锁来看抽象队列同步器（加读锁）

Love The Way You Lie 2022-12-19 08:51 142阅读 0赞

## 越来越有意思了 ##

**读写锁源码分析**

public class ReentrantReadWriteLock
            implements ReadWriteLock, java.io.Serializable { 
     
        //读锁
        private final ReentrantReadWriteLock.ReadLock readerLock;
        //写锁
        private final ReentrantReadWriteLock.WriteLock writerLock;
        //同步机制 抽象类
        final Sync sync;
    
       
        public ReentrantReadWriteLock() { 
            this(false);
        }
        public ReentrantReadWriteLock(boolean fair) { 
            //初始化同步机制 也就是公平锁还是非公平锁
            sync = fair ? new FairSync() : new NonfairSync();
            //初始化读锁
            readerLock = new ReadLock(this);
            //初始化写锁
            writerLock = new WriteLock(this);
        }
        
     	/*-------------------------------同步机制-----------------------------*/
    	//非公平
    	static final class NonfairSync extends Sync { 
            private static final long serialVersionUID = -8159625535654395037L;
            //读数据是否阻塞
            final boolean writerShouldBlock() { 
                return false; // writers can always barge
            }
            //写数据是否阻塞
            final boolean readerShouldBlock() { 
                //返回 同步器中是否有其他等待获取锁的线程或者node结点
                return AbstractQueuedSynchronizer.apparentlyFirstQueuedIsExclusive(){ 
    				Node h, s;
    		        return (h=head)!=nul && 
    		         (s = h.next)!= null &&
    		         !s.isShared() && 
    		         s.thread != null;
    			};
            }
        }
        //公平
    	static final class FairSync extends Sync { 
            private static final long serialVersionUID = -2274990926593161451L;
            final boolean writerShouldBlock() { 
                return AbstractQueuedSynchronizer.hasQueuedPredecessors(){ 
    				Node t = tail; // Read fields in reverse initialization order
            		Node h = head;
            		Node s;
            		//如果队列中存在等待的结点 并且当前线程不是第二个结点 那么返回true 也即是需要阻塞
            		return h != t &&
                		((s = h.next) == null || s.thread != Thread.currentThread());
    			};
            }
            //同上writerShouldBlock
            final boolean readerShouldBlock() { 
                return hasQueuedPredecessors();
            }
        }
      	/*-------------------------------读锁-----------------------------*/
      	public static class ReadLock implements Lock, java.io.Serializable { 
            private static final long serialVersionUID = -5992448646407690164L;
            private final Sync sync;
    		//初始化同步机制
            protected ReadLock(ReentrantReadWriteLock lock) { 
                sync = lock.sync;
            }
    	}
    		
    
    		//释放锁
            public void unlock() { 
                sync.releaseShared(1);
            }

**首先从我们的readLock.lock()方法进去分析**

ReadWriteLock lock = reentranLockTest.getLock();
    Lock readLock = lock.readLock();
    readLock.lock();

**也就是下面的这个方法**

public void lock() { 
     	 sync.acquireShared(1);
    }
    //也即是抽象同步队列里面的这个 获取共享锁 重点！！！
    public final void acquireShared(int arg) { 
        //获取锁，如果返回值<0说明失败了
    	if (tryAcquireShared(arg) < 0)
    	    //加入队列 自旋去获取锁
        	doAcquireShared(arg);
    }

**1、tryAcquireShared() 获取锁的具体实现**

//首先来看获取锁的具体逻辑
    protected final int tryAcquireShared(int unused) { 
         //获取当前线程
         Thread current = Thread.currentThread();
         //获取读写锁的状态
         int c = getState();
         //如果存在排它锁，并且本线程不是排它锁的线程 直接返回-1 也就是失败
         if (exclusiveCount(c) != 0 && getExclusiveOwnerThread() != current)
             return -1;
         //获取线程持有共享锁的锁的数量 因为有重入锁 一个线程可能会获取两三次锁， 
         //所以是锁的数量，而不是线程的数量
         int r = sharedCount(c);
         
         //查看读数据是否需要阻塞 也就是下一个结点是不是独占模式
         //在默认的非公平模式下 他的判断逻辑是这个
         /** 判断头结点不为null并且下一个结点是排它结点，则返回true final boolean apparentlyFirstQueuedIsExclusive() { Node h, s; return (h = head) != null && (s = h.next) != null && !s.isShared() && s.thread != null; } */
         if (!readerShouldBlock() && r < MAX_COUNT &&  compareAndSetState(c, c + SHARED_UNIT){ 
             //不是独占模式，那么尝试修改state，也就是加锁 SHARED_UNIT=1<<16=65536
             //为什么要这么做呢？ 为什么是1<<16呢？ 因为它使用高16位记录读锁 低16位记录写锁
             if (r == 0) { 
                 //共享锁的数量为0 将本线程赋值给firstReader
                 firstReader = current;
                 //将线程数量设置为1
                 firstReaderHoldCount = 1;
             //线程重入了 也就是本线程多次获取锁
             } else if (firstReader == current) { 
                 //将线程持有的锁数量+1
                 firstReaderHoldCount++;
             //当前线程不是第一个读线程 记录每一个线程读的次数
             } else {  
                 //线程持有的锁的计数器
                 HoldCounter rh = cachedHoldCounter;
                 //缓存的线程计数器不是本线程的
                 if (rh == null || rh.tid != getThreadId(current))
                     //从threadlocal中获取线程计数器 并赋值给cachedHoldCounter
                     cachedHoldCounter = rh = readHolds.get();
                 //如果线程持有的锁的数量为0，也就是新的线程，需要将其加入到threadlocal中
                 //在我debug的过程中，发现根本不会走到这里 可能是后面去释放了锁，然后再获取锁才会走到这里
                 else if (rh.count == 0)
                     readHolds.set(rh);
                 //将线程持有的锁的数量+1
                 rh.count++;
             }
             //返回成功
             return 1;
         }
         //再次尝试获取锁
         return fullTryAcquireShared(current);
     }

**如果上面获取锁失败，也就是存在独占锁，或者是cas失败了（其他线程cas成功，导致本线程失败），再一次尝试获取锁**

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，有可能是重入方式获取锁
                   if (firstReader == current) { 
                   } 
                   else { 
                   //这里面的逻辑没怎么看懂 ，每一步代码做什么都知道 但是为什么这么做？？？
                       if (rh == null) { 
                           //拿到缓存的线程计数器
                           rh = cachedHoldCounter;
                           //不是本线程
                           if (rh == null || rh.tid != getThreadId(current)) { 
                               //从threadLocal中获取
                               rh = readHolds.get();
                               if (rh.count == 0)
                                   //如果线程持有的锁的数量为0 将其移出threadlocal
                                   readHolds.remove();
                           }
                       }
                       //缓存的线程持有的锁的数量为0 也就是其他线程也没获取锁，那么直接返回失败
                       if (rh.count == 0)
                           return -1;
                   }
               }
               //获取共享锁的数量
               if (sharedCount(c) == MAX_COUNT)
                   throw new Error("Maximum lock count exceeded");
               //获取锁 这一块同上面 tryAcquireShared（）方法
               if (compareAndSetState(c, c + SHARED_UNIT)) { 
                   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;
               }
           }
       }

**2、如果获取锁时返回-1，那么就是获取锁失败了，则会加入队列中 源码如下**

//如果获取锁失败 就走到这里 将其添加到队列中 这一块移步我的抽象队列同步器的分析
    private void doAcquireShared(int arg) { 
        //将节点挂在到队列 并设置其为尾结点
           final Node node = addWaiter(Node.SHARED);
           boolean failed = true;
           try { 
               //线程是否被中断过
               boolean interrupted = false;
               for (;;) { 
                   //2.1获取上一个结点
                   final Node p = node.predecessor();
                   if (p == head) { 
                       //再一次尝试获取锁 前一个结点已经是头节点了，那么下一次唤醒的线程就是本线程，
                       //所以再一次尝试获取锁， 实现方法也就是 步骤1 的分析
                       int r = tryAcquireShared(arg);
                       if (r >= 0) { 
                           //2.2将自己设置成头节点 如果还可以获取锁，则唤醒下一个线程
                           setHeadAndPropagate(node, r);
                           p.next = null; // help GC
                           if (interrupted)
                               selfInterrupt();
                           failed = false;
                           return;
                       }
                   }
                   //2.3再一次判断前面的结点是不是都是在被唤醒的状态
                   //2.4调用 parkAndCheckInterrupt，线程进入等待状态
                   if (shouldParkAfterFailedAcquire(p, node) &&
                       parkAndCheckInterrupt())
                       interrupted = true;
               }
           } finally { 
               //2.5如果获取锁的过程中出现异常，会将本线程踢出队列
               if (failed)
                   cancelAcquire(node);
           }
       }

**2.1 获取上一个结点 node:队列的尾结点。解释一波：如果队列为空，那么在第一个线程进入到这里等待的时候，会创建一个头结点，头节点不保存数据，然后将本线程包装成node，挂在头节点的后面。所以这里肯定是有前一个节点的**

final Node predecessor() throws NullPointerException { 
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

**2.2 将自己设置成头节点 如果还可以获取锁，则唤醒下一个线程**

private void setHeadAndPropagate(Node node, int propagate) { 
        Node h = head; // Record old head for check below
        setHead(node);
        //如果还有剩余量，继续唤醒下一个结点的线程
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) { 
            //获取下一个结点 如果下一个结点需要共享锁
            Node s = node.next;
            if (s == null || s.isShared())
                //todo 后续在锁的释放分析
                doReleaseShared();
        }
      }

**2.3 再一次判断前面的结点是不是都是在等待被唤醒（也就是自己需不需要阻塞等待）**

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
        int ws = pred.waitStatus;
        //上一个结点是等待被唤醒的状态，直接返回true
        if (ws == Node.SIGNAL)
            return true;
        //ws>0说明改结点是cancel状态，需要从队列中剔除 循环剔除这些结点线程
        if (ws > 0) { 
            do { 
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else { 
            //将上一个结点设置成signal（-1） 也即是等待被唤醒的状态
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

**2.4 调用 parkAndCheckInterrupt，线程进入阻塞等待状态**

private final boolean parkAndCheckInterrupt() { 
        //调用park()使线程进入waiting状态
        LockSupport.park(this);
        //如果被唤醒，查看自己是不是被中断的。
        return Thread.interrupted();
    }

**2.5 如果获取锁的过程中出现异常，会将本线程踢出队列**

private void cancelAcquire(Node node) { 
        if (node == null)
            return;
        node.thread = null;
        //获取前置结点
        Node pred = node.prev;
        //前置结点也是cancel状态，则剔除
        while (pred.waitStatus > 0)
            node.prev = pred = pred.prev;
        //找到正常状态的pred（waitStatus小于0的node结点线程）
        Node predNext = pred.next;
        //将本结点设置为取消状态
        node.waitStatus = Node.CANCELLED;
        //本线程结点是尾结点 设置尾结点为pred结点
        if (node == tail && compareAndSetTail(node, pred)) { 
            //成功之后，设置pred的后置结点设置为null
            compareAndSetNext(pred, predNext, null);
        } else { 
            //如果本结点不是尾结点 或者上面的cas设置本结点的前置结点失败了
            int ws;
            //pred不是头节点 并且 （pred状态是正常阻塞等待被唤醒的结点并且成功将其状态设置为了-1） 
            // 并且pred的线程不为null 进入if
            if (pred != head &&
                ((ws = pred.waitStatus) == Node.SIGNAL ||
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                pred.thread != null) { 
                //获取node的后置结点
                Node next = node.next;
                if (next != null && next.waitStatus <= 0)
                	//next是等待唤醒状态 将next挂在到pred的后面
                    compareAndSetNext(pred, predNext, next);
            } else { 
                //走到这里，说明pred是头节点 或者 pred结点状态设置失败
                //2.6 唤醒下一个结点
                unparkSuccessor(node);
            }
            //将结点往后移 这个时候，pred的next指向 node的next node的next的pre是node node的pre是pred
            //在下一个线程来获取锁的时候，或者需要添加到队列中的线程会将status为1的node都剔除掉
            //参照2.3 线程在加入队列的时候，会判断前面的结点都是<0的， 如果不是，会将这些>0的都剔除调
            node.next = node; // help GC
        }
    }

**2.6 唤醒下一个结点**

private void unparkSuccessor(Node node) { 
        
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
        //获取下一个结点
        Node s = node.next;
        if (s == null || s.waitStatus > 0) { 
            s = null;
            //从后往前找，找到最前面的状态是<=0的结点
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        //如果找到了，将其唤醒
        if (s != null)
            LockSupport.unpark(s.thread);
    }

**到了这里，非公平模式下的读锁的整个获取流程已经结束**

**遗留点：2.2 最后有一行代码：doReleaseShared(); 在下一篇的从 读写锁来看抽象队列同步器（释放读锁）来分析**