登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›23. Java JUC源码分析系列笔记-ReentrantReadWriteLock ...

23. Java JUC源码分析系列笔记-ReentrantReadWriteLock

觞刈 昨天 15:50
目录

  • 1. ReentrantReadWriteLock是什么
  • 2. 非公平ReentrantReadWriteLock

    • 2.1. 是什么
    • 2.2. 怎么使用
    • 2.3. 源码分析

      • 2.3.1. uml
      • 2.3.2. 构造方法
      • 2.3.3. 读锁加锁

        • 2.3.3.1. 使用AQS加共享锁

          • 2.3.3.1.1. 使用Sync尝试加共享锁

            • 2.3.3.1.1.1. 判断是否需要阻塞读【非公平锁】
            • 2.3.3.1.1.2. 快速尝试加锁失败，那么改用死循环加锁

          • 2.3.3.1.2. 尝试加锁失败，那么入AQS队列并阻塞，等待唤醒继续抢占锁

            • 2.3.3.1.2.1. 加入AQS队列并阻塞



      • 2.3.4. 读锁解锁

        • 2.3.4.1. 使用AQS解共享锁

          • 2.3.4.1.1. 使用Sync尝试解锁
          • 2.3.4.1.2. 所有共享锁都被释放，唤醒AQS队列中头节点的下一个节点

            • 2.3.4.1.2.1. 唤醒AQS队列中头节点的下一个节点



      • 2.3.5. 写锁加锁

        • 2.3.5.1. 调用AQS加互斥锁

          • 2.3.5.1.1. 使用Sync尝试加互斥锁

            • 2.3.5.1.1.1. 判断写是否需要阻塞【非公平锁】

          • 2.3.5.1.2. 尝试加锁失败，那么入AQS队列并阻塞，等待唤醒继续抢占锁


      • 2.3.6. 写锁解锁

        • 2.3.6.1. 调用AQS解互斥锁

          • 2.3.6.1.1. 调用Sync尝试解互斥锁
          • 2.3.6.1.2. 尝试解锁成功后，唤醒AQS队列中头节点的下一个节点




  • 3. 公平ReentrantReadWriteLock

    • 3.1. 是什么
    • 3.2. 怎么使用
    • 3.3. 源码分析

      • 3.3.1. uml
      • 3.3.2. 构造方法
      • 3.3.3. 读锁加锁

        • 3.3.3.1. 使用AQS加共享锁

          • 3.3.3.1.1. 使用Sync尝试加共享锁

            • 3.3.3.1.1.1. 判断是否需要阻塞读【公平】
            • 3.3.3.1.1.2. 快速尝试加锁失败，那么改用死循环加锁

          • 3.3.3.1.2. 尝试加锁失败，那么入AQS队列并阻塞，等待唤醒继续抢占锁

            • 3.3.3.1.2.1. 加入AQS队列并阻塞



      • 3.3.4. 读锁解锁

        • 3.3.4.1. 使用AQS解共享锁

          • 3.3.4.1.1. 使用Sync尝试解锁
          • 3.3.4.1.2. 所有共享锁都被释放，唤醒AQS队列中头节点的下一个节点

            • 3.3.4.1.2.1. 唤醒AQS队列中头节点的下一个节点



      • 3.3.5. 写锁加锁

        • 3.3.5.1. 调用AQS加互斥锁

          • 3.3.5.1.1. 使用Sync尝试加互斥锁

            • 3.3.5.1.1.1. 判断写是否需要阻塞【公平】

          • 3.3.5.1.2. 尝试加锁失败，那么入AQS队列并阻塞，等待唤醒继续抢占锁


      • 3.3.6. 写锁解锁

        • 3.3.6.1. 调用AQS解互斥锁

          • 3.3.6.1.1. 调用Sync尝试解互斥锁
          • 3.3.6.1.2. 尝试解锁成功后，唤醒AQS队列中头节点的下一个节点




  • 4. 参考

1. ReentrantReadWriteLock是什么

ReentrantLock保证了同一时间只有一个线程可以在临界区读或者写数据，这意味着如果有两个读线程同时读取数据，ReentrantLock也只允许其中一个通过，但我们想要的是读可以并发执行，一旦有写则其他线程等待。如下表：
是否可以同时进行读写读√×写××因此，ReentrantReadWriteLock就诞生了
2. 非公平ReentrantReadWriteLock

2.1. 是什么

无论队列前面是否有人排队等待锁，我直接去抢
2.2. 怎么使用
  1. public class ReadWriteLockTest
  2. {
  3.     private static ReadWriteLock lock = new ReentrantReadWriteLock();//默认非公平
  4.     private static Lock readLock = lock.readLock();
  5.     private static Lock writeLock = lock.writeLock();
  7.     private static List<Integer> data = new ArrayList<>();
  9.     public static void main(String[] args) throws InterruptedException
  10.     {
  11.         Thread readThread = new Thread(() -> {
  12.             while (true)
  13.             {
  14.                 try
  15.                 {
  16.                     TimeUnit.MILLISECONDS.sleep(500);
  17.                     readLock.lock();
  18.                     System.out.println(Thread.currentThread().getName() + " read: " + data);
  19.                 }
  20.                 catch (InterruptedException e)
  21.                 {
  22.                     e.printStackTrace();
  23.                 }
  24.                 finally
  25.                 {
  26.                     readLock.unlock();
  27.                 }
  28.             }
  29.         });
  30.         Thread readThread2 = new Thread(() -> {
  31.             while (true)
  32.             {
  33.                 try
  34.                 {
  35.                     TimeUnit.MILLISECONDS.sleep(300);
  37.                     readLock.lock();
  38.                     System.out.println(Thread.currentThread().getName() + " read: " + data);
  39.                 }
  40.                 catch (InterruptedException e)
  41.                 {
  42.                     e.printStackTrace();
  43.                 }
  44.                 finally
  45.                 {
  46.                     readLock.unlock();
  47.                 }
  48.             }
  49.         });
  50.         Thread writeThread = new Thread(() -> {
  52.             int i = 0;
  53.             while (true)
  54.             {
  55.                 try
  56.                 {
  57.                     TimeUnit.MILLISECONDS.sleep(200);
  59.                     writeLock.lock();
  60.                     if (i % 2 == 0)
  61.                     {
  62.                         data.add(i);
  63.                     }else
  64.                     {
  65.                         data.remove(0);
  66.                     }
  67.                     i++;
  68.                 }
  69.                 catch (InterruptedException e)
  70.                 {
  71.                     e.printStackTrace();
  72.                 }
  73.                 finally
  74.                 {
  75.                     writeLock.unlock();
  76.                 }
  77.             }
  78.         });
  80.         readThread.start();
  81.         readThread2.start();
  82.         writeThread.start();
  84.         readThread.join();
  85.         readThread2.join();
  86.         writeThread.join();
  88.     }
  90. }
复制代码
2.3. 源码分析

2.3.1. uml
  1. @startuml
  2. skinparam classAttributeIconSize 0
  4. interface AQS{
  5. }
  7. class Sync{
  8. }
  10. interface Lock{
  11. }
  13. interface ReadWriteLock{
  14. }
  16. class ReentrantReadWriteLock{
  17. }
  19. class WriteLock{
  20. }
  22. class ReadLock{
  23. }
  25. class WriteLock{
  27. }
  29. Lock <|-- ReadLock
  31. Lock <|-- WriteLock
  33. ReadWriteLock <|-- ReentrantReadWriteLock
  34. AQS <|-- Sync
  36. ReentrantReadWriteLock --> ReadLock
  37. ReentrantReadWriteLock --> WriteLock
  38. ReadLock --> Sync
  39. WriteLock --> Sync
  40. @enduml
复制代码
2.3.2. 构造方法


  • ReentrantReadWriteLock
  1. public ReentrantReadWriteLock() {
  2.     //默认是false
  3.     this(false);
  4. }
  6. public ReentrantReadWriteLock(boolean fair) {
  7.     //初始化了Sync
  8.     //false的话使用的是NonfairSync
  9.     sync = fair ? new FairSync() : new NonfairSync();
  10.     //初始化读写锁
  11.     readerLock = new ReadLock(this);
  12.     writerLock = new WriteLock(this);
  13. }
复制代码

  • ReentrantReadWriteLock.ReadLock
  1. protected ReadLock(ReentrantReadWriteLock lock) {
  2.     //其实就是保存了ReentrantReadWriteLock的Sync
  3.     sync = lock.sync;
  4. }
复制代码

  • ReentrantReadWriteLock.WriteLock
  1. protected WriteLock(ReentrantReadWriteLock lock) {
  2.     //其实就是保存了ReentrantReadWriteLock的Sync
  3.     sync = lock.sync;
  4. }
复制代码
2.3.3. 读锁加锁


  • ReentrantReadWriteLock.readLock
  1. //返回就是读锁
  2. public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }
复制代码

  • ReadLock.lock
  1. public void lock() {
  2.     //使用AQS加共享锁
  3.     sync.acquireShared(1);
  4. }
复制代码
2.3.3.1. 使用AQS加共享锁


  • AQS acquireShared
  1. public final void acquireShared(int arg) {
  2.     //ReentrantReadWriteLock的Sync重写了tryAcquireShared
  3.     //所以这里调用的是Sync的tryAcquireShared
  4.    
  5.     //如果返回<0说明尝试加锁失败，执行doAcquireShared入AQS队列并阻塞，等待唤醒
  6.     //返回>=0说明加锁成功，执行后续的业务逻辑
  7.     if (tryAcquireShared(arg) < 0)
  8.         doAcquireShared(arg);
  9. }
复制代码
2.3.3.1.1. 使用Sync尝试加共享锁


  • ReentrantReadWriteLock.Sync.tryAcquireShared
  1. protected final int tryAcquireShared(int unused) {
  2.     //当前线程
  3.     Thread current = Thread.currentThread();
  4.     //当前的state数目（或者说加锁【既包括读锁也包括写锁】的数量）
  5.     int c = getState();
  6.     //从state中获取互斥锁的数目（写锁的数量）
  7.     //如果数量不为0（即>0）说明已经加了写锁
  8.     if (exclusiveCount(c) != 0 &&
  9.         //判断加锁的线程是否当前线程
  10.         getExclusiveOwnerThread() != current)
  11.         //不是的话返回-1表示已经有其他线程加了写锁（后面就需要入AQS队列阻塞等待唤醒）
  12.         return -1;
  13.    
  14.     //走到这里说明没有线程加写锁或者加写锁的就是本线程
  15.    
  16.     //从state中获取共享锁的数目（读锁的数目）
  17.     int r = sharedCount(c);
  18.     //非公平锁：这里调用NonfairSync的readerShouldBlock判断是否需要阻塞读
  19.     //返回false的话不需要阻塞，接着执行&&后面的逻辑
  20.     if (!readerShouldBlock() &&
  21.         //判断读锁数量是否小于最大数目
  22.         r < MAX_COUNT &&
  23.         //使用CAS加读锁
  24.         compareAndSetState(c, c + SHARED_UNIT)) {
  26.         //上面的if中的逻辑执行完毕后，锁state已经加好了读锁
  27.         //下面的逻辑负责处理其他需要修改的属性
  29.         //第一次加读锁
  30.         if (r == 0) {
  31.             //设置第一次加锁的线程以及初始化读锁数量
  32.             firstReader = current;
  33.             firstReaderHoldCount = 1;
  34.         //第n次加锁的仍旧是第一次加锁的线程
  35.         } else if (firstReader == current) {
  36.             //修改读锁数量即可
  37.             firstReaderHoldCount++;
  38.         //第n次加锁的不是第一次加锁的线程
  39.         } else {
  40.             //cachedHoldCounter是缓存
  41.             HoldCounter rh = cachedHoldCounter;
  42.             if (rh == null || rh.tid != getThreadId(current))
  43.                 cachedHoldCounter = rh = readHolds.get();
  44.             else if (rh.count == 0)
  45.                 readHolds.set(rh);
  46.             rh.count++;
  47.         }
  48.         //返回1（正数）表示加读锁成功
  49.         return 1;
  50.     }
  52.     //走到这里说明发生了以下几种情况：
  53.     //1.需要阻塞读
  54.     //2.读锁数目已经超过最大值
  55.     //3.CAS加读锁失败
  56.     return fullTryAcquireShared(current);
  57. }
复制代码
2.3.3.1.1.1. 判断是否需要阻塞读【非公平锁】


  • ReentrantReadWriteLock.NonfairSync#readerShouldBlock
  1. final boolean readerShouldBlock() {
  3.     //队头是互斥节点（加写锁的节点）情况下才需要阻塞读
  4.     //这就是非公平锁的特点
  5.     //即使队列前面已经有其他加读锁的线程等待我也不管
  6.     return apparentlyFirstQueuedIsExclusive();
  7. }
复制代码

  • apparentlyFirstQueuedIsExclusive
  1. final boolean apparentlyFirstQueuedIsExclusive() {
  2.     Node h, s;
  4.     //队列不为空
  5.     return (h = head) != null &&
  6.         //并且 队列的实际头节点【之所以说实际队列的头节点是个占位符】不为空
  7.         (s = h.next)  != null &&
  8.         //并且 队列的实际头节点不是共享的头节点（即加的不是读锁）
  9.         !s.isShared()         &&
  10.         //并且 队列的实际头节点的线程不为空
  11.         s.thread != null;
  12.     //满足以上所有情况才需要阻塞当前加读锁的线程
  13. }
复制代码
2.3.3.1.1.2. 快速尝试加锁失败，那么改用死循环加锁


  • ReentrantReadWriteLock.Sync#fullTryAcquireShared
  1. final int fullTryAcquireShared(Thread current) {
  3.     HoldCounter rh = null;
  4.     //死循环
  5.     for (;;) {
  6.         int c = getState();
  7.         //写锁的数量不为0
  8.         if (exclusiveCount(c) != 0) {
  9.             //且加写锁的不是本线程
  10.             if (getExclusiveOwnerThread() != current)
  11.                 //返回-1
  12.                 return -1;
  13.             // else we hold the exclusive lock; blocking here
  14.             // would cause deadlock.
  15.         //没有人加写锁，如果读需要阻塞
  16.         } else if (readerShouldBlock()) {
  17.             // Make sure we're not acquiring read lock reentrantly
  18.             if (firstReader == current) {
  19.                 // assert firstReaderHoldCount > 0;
  20.             } else {
  21.                 if (rh == null) {
  22.                     rh = cachedHoldCounter;
  23.                     if (rh == null || rh.tid != getThreadId(current)) {
  24.                         rh = readHolds.get();
  25.                         if (rh.count == 0)
  26.                             readHolds.remove();
  27.                     }
  28.                 }
  29.                 if (rh.count == 0)
  30.                     return -1;
  31.             }
  32.         }
  33.         //走到这里说明没有加写锁，读也不需要阻塞
  34.         if (sharedCount(c) == MAX_COUNT)
  35.             //加锁已超过最大值
  36.             throw new Error("Maximum lock count exceeded");
  37.         //加读锁，同上面的tryAcquireShared
  38.         if (compareAndSetState(c, c + SHARED_UNIT)) {
  39.             if (sharedCount(c) == 0) {
  40.                 firstReader = current;
  41.                 firstReaderHoldCount = 1;
  42.             } else if (firstReader == current) {
  43.                 firstReaderHoldCount++;
  44.             } else {
  45.                 if (rh == null)
  46.                     rh = cachedHoldCounter;
  47.                 if (rh == null || rh.tid != getThreadId(current))
  48.                     rh = readHolds.get();
  49.                 else if (rh.count == 0)
  50.                     readHolds.set(rh);
  51.                 rh.count++;
  52.                 cachedHoldCounter = rh; // cache for release
  53.             }
  54.             return 1;
  55.         }
  56.     }
  57. }
复制代码
2.3.3.1.2. 尝试加锁失败，那么入AQS队列并阻塞，等待唤醒继续抢占锁
  1. private void doAcquireShared(int arg) {
  2.     //构造SHARE节点加入AQS队列阻塞等待唤醒
  3.     final Node node = addWaiter(Node.SHARED);
  4.     boolean failed = true;
  5.     try {
  6.         boolean interrupted = false;
  7.         //死循环抢占锁
  8.         for (;;) {
  9.             //当前节点的前一个节点是头节点
  10.             final Node p = node.predecessor();
  11.             if (p == head) {
  12.                 //继续尝试加共享锁
  13.                 int r = tryAcquireShared(arg);
  14.                 if (r >= 0) {
  15.                     setHeadAndPropagate(node, r);
  16.                     p.next = null; // help GC
  17.                     if (interrupted)
  18.                         selfInterrupt();
  19.                     failed = false;
  20.                     return;
  21.                 }
  22.             }
  23.             //判断是否需要阻塞
  24.             if (shouldParkAfterFailedAcquire(p, node) &&
  25.                 //需要的话进行阻塞
  26.                 parkAndCheckInterrupt())
  27.                 interrupted = true;
  28.         }
  29.     } finally {
  30.         if (failed)
  31.             cancelAcquire(node);
  32.     }
  33. }
复制代码
2.3.3.1.2.1. 加入AQS队列并阻塞


  • addWaiter
  1. private Node addWaiter(Node mode) {
  2.     Node node = new Node(Thread.currentThread(), mode);
  3.     // Try the fast path of enq; backup to full enq on failure
  4.     //快速尝试加入队尾
  5.     Node pred = tail;
  6.     if (pred != null) {
  7.         node.prev = pred;
  8.         if (compareAndSetTail(pred, node)) {
  9.             pred.next = node;
  10.             return node;
  11.         }
  12.     }
  13.     //快速尝试加入队尾失败，那么改用enq加入队尾
  14.     enq(node);
  15.     return node;
  16. }
复制代码
参考：5.AQS.md
2.3.4. 读锁解锁


  • ReentrantReadWriteLock.ReadLock#unlock
  1. public void unlock() {
  2.     //使用AQS解共享锁
  3.     sync.releaseShared(1);
  4. }
复制代码
2.3.4.1. 使用AQS解共享锁


  • AbstractQueuedSynchronizer#releaseShared
  1. public final boolean releaseShared(int arg) {
  2.     //调用Sync尝试解共享锁
  3.     //如果共享锁【读锁】已经全部释放完，那么执行doReleaseShared
  4.     if (tryReleaseShared(arg)) {
  5.         doReleaseShared();
  6.         return true;
  7.     }
  8.     return false;
  9. }
复制代码
2.3.4.1.1. 使用Sync尝试解锁


  • ReentrantReadWriteLock.Sync#tryReleaseShared
  1. protected final boolean tryReleaseShared(int unused) {
  2.     Thread current = Thread.currentThread();
  3.     //当前线程就是第一次加读锁的线程
  4.     if (firstReader == current) {
  5.         //所有读锁都释放完毕
  6.         if (firstReaderHoldCount == 1)
  7.             //那么置加锁线程为空
  8.             firstReader = null;
  9.         //没有释放完那么减读锁数量
  10.         else
  11.             firstReaderHoldCount--;
  12.     //当前线程不是第一次加读锁的线程
  13.     } else {
  14.         HoldCounter rh = cachedHoldCounter;
  15.         if (rh == null || rh.tid != getThreadId(current))
  16.             rh = readHolds.get();
  17.         int count = rh.count;
  18.         if (count <= 1) {
  19.             readHolds.remove();
  20.             if (count <= 0)
  21.                 throw unmatchedUnlockException();
  22.         }
  23.         --rh.count;
  24.     }
  25.     //死循环修改state数量
  26.     for (;;) {
  27.         int c = getState();
  28.         int nextc = c - SHARED_UNIT;
  29.         //CAS修改state数量
  30.         if (compareAndSetState(c, nextc))
  31.             return nextc == 0;//减为0了那么返回true
  32.     }
  33. }
复制代码
3.3. 源码分析

3.3.1. uml
  1. @startuml
  2. skinparam classAttributeIconSize 0
  4. interface AQS{
  5. }
  7. class Sync{
  8. }
  10. interface Lock{
  11. }
  13. interface ReadWriteLock{
  14. }
  16. class ReentrantReadWriteLock{
  17. }
  19. class WriteLock{
  20. }
  22. class ReadLock{
  23. }
  25. class WriteLock{
  27. }
  29. Lock <|-- ReadLock
  31. Lock <|-- WriteLock
  33. ReadWriteLock <|-- ReentrantReadWriteLock
  34. AQS <|-- Sync
  36. ReentrantReadWriteLock --> ReadLock
  37. ReentrantReadWriteLock --> WriteLock
  38. ReadLock --> Sync
  39. WriteLock --> Sync
  40. @enduml
复制代码
3.3.2. 构造方法


  • ReentrantReadWriteLock
  1. private void unparkSuccessor(Node node) {
  2.    
  3.     int ws = node.waitStatus;
  4.     //如果当前节点的状态是正常的？
  5.     if (ws < 0)
  6.         //当前节点的状态<0,则把状态改为0
  7.         //0是空的状态，因为node这个节点的线程释放了锁后续不需要做任何
  8.         compareAndSetWaitStatus(node, ws, 0);
  10.    
  11.      //获取当前节点的下一个节点
  12.     Node s = node.next;
  13.     //如果下一个节点是空（即当前节点是尾节点）或者下一个节点的状态>0（取消）
  14.     if (s == null || s.waitStatus > 0) {
  15.         s = null;
  16.         //从尾节点往前遍历至当前节点
  17.         for (Node t = tail; t != null && t != node; t = t.prev)
  18.             //找到最靠近当前节点的状态<=0（非取消）的节点
  19.             if (t.waitStatus <= 0)
  20.                 s = t;
  21.     }
  22.     //唤醒当前节点的下一个节点
  23.     if (s != null)
  24.         LockSupport.unpark(s.thread);
  25. }
复制代码

  • ReentrantReadWriteLock.ReadLock
  1. protected ReadLock(ReentrantReadWriteLock lock) {
  2.     //其实就是保存了ReentrantReadWriteLock的Sync
  3.     sync = lock.sync;
  4. }
复制代码

  • ReentrantReadWriteLock.WriteLock
  1. protected WriteLock(ReentrantReadWriteLock lock) {
  2.     //其实就是保存了ReentrantReadWriteLock的Sync
  3.     sync = lock.sync;
  4. }
复制代码
3.3.3. 读锁加锁


  • ReentrantReadWriteLock.readLock
  1. //返回就是读锁
  2. public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }
复制代码

  • ReadLock.lock
  1. public void lock() {
  2.     //使用AQS加共享锁
  3.     sync.acquireShared(1);
  4. }
复制代码
3.3.3.1. 使用AQS加共享锁


  • AQS acquireShared
  1. public final void acquireShared(int arg) {
  2.     //ReentrantReadWriteLock的Sync重写了tryAcquireShared
  3.     //所以这里调用的是Sync的tryAcquireShared
  4.    
  5.     //如果返回<0说明尝试加锁失败，执行doAcquireShared入AQS队列并阻塞，等待唤醒
  6.     //返回>=0说明加锁成功，执行后续的业务逻辑
  7.     if (tryAcquireShared(arg) < 0)
  8.         doAcquireShared(arg);
  9. }
复制代码
3.3.3.1.1. 使用Sync尝试加共享锁


  • ReentrantReadWriteLock.Sync.tryAcquireShared
  1. protected final int tryAcquireShared(int unused) {
  2.     //当前线程
  3.     Thread current = Thread.currentThread();
  4.     //当前的state数目（或者说加锁【既包括读锁也包括写锁】的数量）
  5.     int c = getState();
  6.     //从state中获取互斥锁的数目（写锁的数量）
  7.     //如果数量不为0（即>0）说明已经加了写锁
  8.     if (exclusiveCount(c) != 0 &&
  9.         //判断加锁的线程是否当前线程
  10.         getExclusiveOwnerThread() != current)
  11.         //不是的话返回-1表示已经有其他线程加了写锁（后面就需要入AQS队列阻塞等待唤醒）
  12.         return -1;
  13.    
  14.     //走到这里说明没有线程加写锁或者加写锁的就是本线程
  15.    
  16.     //从state中获取共享锁的数目（读锁的数目）
  17.     int r = sharedCount(c);
  18.     //非公平锁：这里调用NonfairSync的readerShouldBlock判断是否需要阻塞读
  19.     //返回false的话不需要阻塞，接着执行&&后面的逻辑
  20.     if (!readerShouldBlock() &&
  21.         //判断读锁数量是否小于最大数目
  22.         r < MAX_COUNT &&
  23.         //使用CAS加读锁
  24.         compareAndSetState(c, c + SHARED_UNIT)) {
  26.         //上面的if中的逻辑执行完毕后，锁state已经加好了读锁
  27.         //下面的逻辑负责处理其他需要修改的属性
  29.         //第一次加读锁
  30.         if (r == 0) {
  31.             //设置第一次加锁的线程以及初始化读锁数量
  32.             firstReader = current;
  33.             firstReaderHoldCount = 1;
  34.         //第n次加锁的仍旧是第一次加锁的线程
  35.         } else if (firstReader == current) {
  36.             //修改读锁数量即可
  37.             firstReaderHoldCount++;
  38.         //第n次加锁的不是第一次加锁的线程
  39.         } else {
  40.             //cachedHoldCounter是缓存
  41.             HoldCounter rh = cachedHoldCounter;
  42.             if (rh == null || rh.tid != getThreadId(current))
  43.                 cachedHoldCounter = rh = readHolds.get();
  44.             else if (rh.count == 0)
  45.                 readHolds.set(rh);
  46.             rh.count++;
  47.         }
  48.         //返回1（正数）表示加读锁成功
  49.         return 1;
  50.     }
  52.     //走到这里说明发生了以下几种情况：
  53.     //1.需要阻塞读
  54.     //2.读锁数目已经超过最大值
  55.     //3.CAS加读锁失败
  56.     return fullTryAcquireShared(current);
  57. }
复制代码
3.3.3.1.1.1. 判断是否需要阻塞读【公平】


  • ReentrantReadWriteLock.FairSync#readerShouldBlock
  1. //...
复制代码

  • AQS hasQueuedPredecessors
  1. public void unlock() {
  2.     //AQS解锁
  3.     sync.release(1);
  4. }
复制代码
3.3.3.1.1.2. 快速尝试加锁失败，那么改用死循环加锁


  • ReentrantReadWriteLock.Sync#fullTryAcquireShared
  1. final int fullTryAcquireShared(Thread current) {
  3.     HoldCounter rh = null;
  4.     //死循环
  5.     for (;;) {
  6.         int c = getState();
  7.         //写锁的数量不为0
  8.         if (exclusiveCount(c) != 0) {
  9.             //且加写锁的不是本线程
  10.             if (getExclusiveOwnerThread() != current)
  11.                 //返回-1
  12.                 return -1;
  13.             // else we hold the exclusive lock; blocking here
  14.             // would cause deadlock.
  15.         //没有人加写锁，如果读需要阻塞
  16.         } else if (readerShouldBlock()) {
  17.             // Make sure we're not acquiring read lock reentrantly
  18.             if (firstReader == current) {
  19.                 // assert firstReaderHoldCount > 0;
  20.             } else {
  21.                 if (rh == null) {
  22.                     rh = cachedHoldCounter;
  23.                     if (rh == null || rh.tid != getThreadId(current)) {
  24.                         rh = readHolds.get();
  25.                         if (rh.count == 0)
  26.                             readHolds.remove();
  27.                     }
  28.                 }
  29.                 if (rh.count == 0)
  30.                     return -1;
  31.             }
  32.         }
  33.         //走到这里说明没有加写锁，读也不需要阻塞
  34.         if (sharedCount(c) == MAX_COUNT)
  35.             //加锁已超过最大值
  36.             throw new Error("Maximum lock count exceeded");
  37.         //加读锁，同上面的tryAcquireShared
  38.         if (compareAndSetState(c, c + SHARED_UNIT)) {
  39.             if (sharedCount(c) == 0) {
  40.                 firstReader = current;
  41.                 firstReaderHoldCount = 1;
  42.             } else if (firstReader == current) {
  43.                 firstReaderHoldCount++;
  44.             } else {
  45.                 if (rh == null)
  46.                     rh = cachedHoldCounter;
  47.                 if (rh == null || rh.tid != getThreadId(current))
  48.                     rh = readHolds.get();
  49.                 else if (rh.count == 0)
  50.                     readHolds.set(rh);
  51.                 rh.count++;
  52.                 cachedHoldCounter = rh; // cache for release
  53.             }
  54.             return 1;
  55.         }
  56.     }
  57. }
复制代码
3.3.3.1.2. 尝试加锁失败，那么入AQS队列并阻塞，等待唤醒继续抢占锁
  1. private void doAcquireShared(int arg) {
  2.     //构造SHARE节点加入AQS队列阻塞等待唤醒
  3.     final Node node = addWaiter(Node.SHARED);
  4.     boolean failed = true;
  5.     try {
  6.         boolean interrupted = false;
  7.         //死循环抢占锁
  8.         for (;;) {
  9.             //当前节点的前一个节点是头节点
  10.             final Node p = node.predecessor();
  11.             if (p == head) {
  12.                 //继续尝试加共享锁
  13.                 int r = tryAcquireShared(arg);
  14.                 if (r >= 0) {
  15.                     setHeadAndPropagate(node, r);
  16.                     p.next = null; // help GC
  17.                     if (interrupted)
  18.                         selfInterrupt();
  19.                     failed = false;
  20.                     return;
  21.                 }
  22.             }
  23.             //判断是否需要阻塞
  24.             if (shouldParkAfterFailedAcquire(p, node) &&
  25.                 //需要的话进行阻塞
  26.                 parkAndCheckInterrupt())
  27.                 interrupted = true;
  28.         }
  29.     } finally {
  30.         if (failed)
  31.             cancelAcquire(node);
  32.     }
  33. }
复制代码
3.3.3.1.2.1. 加入AQS队列并阻塞


  • addWaiter
  1. private Node addWaiter(Node mode) {
  2.     Node node = new Node(Thread.currentThread(), mode);
  3.     // Try the fast path of enq; backup to full enq on failure
  4.     //快速尝试加入队尾
  5.     Node pred = tail;
  6.     if (pred != null) {
  7.         node.prev = pred;
  8.         if (compareAndSetTail(pred, node)) {
  9.             pred.next = node;
  10.             return node;
  11.         }
  12.     }
  13.     //快速尝试加入队尾失败，那么改用enq加入队尾
  14.     enq(node);
  15.     return node;
  16. }
复制代码
参考：5.AQS.md
3.3.4. 读锁解锁


  • ReentrantReadWriteLock.ReadLock#unlock
  1. public void unlock() {
  2.     //使用AQS解共享锁
  3.     sync.releaseShared(1);
  4. }
复制代码
3.3.4.1. 使用AQS解共享锁


  • AbstractQueuedSynchronizer#releaseShared
  1. public final boolean releaseShared(int arg) {
  2.     //调用Sync尝试解共享锁
  3.     //如果共享锁【读锁】已经全部释放完，那么执行doReleaseShared
  4.     if (tryReleaseShared(arg)) {
  5.         doReleaseShared();
  6.         return true;
  7.     }
  8.     return false;
  9. }
复制代码
3.3.4.1.1. 使用Sync尝试解锁


  • ReentrantReadWriteLock.Sync#tryReleaseShared
[code]protected final boolean tryReleaseShared(int unused) {    Thread current = Thread.currentThread();    //当前线程就是第一次加读锁的线程    if (firstReader == current) {        //所有读锁都释放完毕        if (firstReaderHoldCount == 1)            //那么置加锁线程为空            firstReader = null;        //没有释放完那么减读锁数量        else            firstReaderHoldCount--;    //当前线程不是第一次加读锁的线程    } else {        HoldCounter rh = cachedHoldCounter;        if (rh == null || rh.tid != getThreadId(current))            rh = readHolds.get();        int count = rh.count;        if (count

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
觞刈
昨天 15:50

0

粉丝关注

10

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫