1.类图结构

ReentrantReadWriteLock.png
image.png

前提

  1.     读写锁内容维护了一个ReadLock和 一个WriteLock，他们依赖Sync实现具体的功能，而Sync继承自AQS，并且也提供了非公平和公平的实现
  2.     在ReentrantReadWriteLock中使用高16位表示读锁，低16位表示写锁

核心参数

  1. static final int SHARED_SHIFT   = 16;
  2. //共享锁(读锁)的状态值 65536
  3. static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
  4. //共享线程的最大个数 65536
  5. static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
  6. //排他锁(写锁)掩码，二进制 15个1
  7. static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;
  9. //返回读锁的线程数
  10. static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
  11. //返回写锁的重入个数
  12. static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }
  14. //第一个获取到读锁的线程
  15. private transient Thread firstReader = null;
  16. //第一个获取到读锁的可重入次数
  17. private transient int firstReaderHoldCount;
  18. //记录最后一个获取锁的线程获取读锁的可重入次数
  19. private transient HoldCounter cachedHoldCounter;
  21. //ThreadLocal类型的变量，用来存储第一个获取读锁线程外其他线程获取读锁的可重入次数
  22. //继承自 ThreadLocal initialValue 方法返回一个 HoldCounter对象
  23. private transient ThreadLocalHoldCounter readHolds;

2.写锁的获取和释放

2.1 void lock()

  1.     写锁是一个独占锁，某个时间只有一个线程能够获取该锁，如果当前没有线程获取到读锁和写锁，则当前线程可以获取写锁然后返回
  2.     如果已经有线程获取到读锁和写锁，则当前线程会被阻塞
  3.   写锁是可重入锁，如果当前线程获取到写锁，再次获取时 可重入次数 +1 即可

代码实现

  1. public void lock() {
  2.     sync.acquire(1);
  3. }
  4. //和ReentrantLock一样，调用AQS中的acquire()方法
  5. public final void acquire(int arg) {
  6.     if (!tryAcquire(arg) &&
  7.         acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  8.         selfInterrupt();
  9. }

查看Sync中重写的 tryAquire()

  1. protected final boolean tryAcquire(int acquires) {
  3.     Thread current = Thread.currentThread();
  4.     int c = getState();
  5.     int w = exclusiveCount(c);
  6.     //说明写锁或读锁已经被其他线程获取了
  7.     if (c != 0) {
  8.         // w == 0说明写锁没有被获取，但是 c != 0 可以得出 读锁被其他线程获取了
  9.         // w != 0说明写锁被其他线程获取了，且当前线程不是当前线程
  10.         //上面两种情况都返回 false
  11.         if (w == 0 || current != getExclusiveOwnerThread())
  12.             return false;
  13.         //能执行到这里说明是当前线程获取到了写锁，则计算可重入的次数，不能超过写锁的最大可重入次数
  14.         if (w + exclusiveCount(acquires) > MAX_COUNT)
  15.             throw new Error("Maximum lock count exceeded");
  16.         // 设置可重入的次数
  17.         setState(c + acquires);
  18.         return true;
  19.     }
  20.     //锁没有被人获取，则直接获取锁
  21.     if (writerShouldBlock() ||
  22.         !compareAndSetState(c, c + acquires))
  23.         return false;
  24.     setExclusiveOwnerThread(current);
  25.     return true;
  26. }

2.2 void unlock()

  1. public void unlock() {
  2.     sync.release(1);
  3. }
  5. public final boolean release(int arg) {
  6.     if (tryRelease(arg)) {
  7.         //激活阻塞队列中的一个线程
  8.         Node h = head;
  9.         if (h != null && h.waitStatus != 0)
  10.             unparkSuccessor(h);
  11.         return true;
  12.     }
  13.     return false;
  14. }

Sync自定义的 tryRelease() 方法

  1. protected final boolean tryRelease(int releases) {
  2.     //确认当前线程是否是写锁的持有者
  3.     if (!isHeldExclusively())
  4.         throw new IllegalMonitorStateException();
  5.     //计算可重入次数
  6.     int nextc = getState() - releases;
  7.     //如果可重入的次数为0则返回true
  8.     boolean free = exclusiveCount(nextc) == 0;
  9.     if (free)
  10.         //置空
  11.         setExclusiveOwnerThread(null);
  12.     //更新状态值
  13.     setState(nextc);
  14.     return free;
  15. }

3.读锁的获取和释放

3.1 void lock()

获取读锁，如果当前没有其他线程持有写锁，则当前线程可以获取读锁，AQS的状态值state的高16位会 + 1，然后返回 如果一个线程持有写锁，则当前线程会被阻塞

  1. public void lock() {
  2.     sync.acquireShared(1);
  3. }
  4. //调用AQS中的 acquireShared
  5. public final void acquireShared(int arg) {
  6.     if (tryAcquireShared(arg) < 0)
  7.         //调用AQS中的 doAcquireShared 方法
  8.         doAcquireShared(arg);
  9. }

然后Sync中自定义了 tryAcquireShared()方法

  1. protected final int tryAcquireShared(int unused) {
  3.     Thread current = Thread.currentThread();
  4.     int c = getState();
  5.     //判断写锁是否被占用，如果写锁被占用且不是当前线程则返回 -1(false)
  6.     //这里也就说明了一个线程既可以获取写锁，也可以获取读锁，但是释放的时候都要释放
  7.     if (exclusiveCount(c) != 0 &&
  8.         getExclusiveOwnerThread() != current)
  9.         return -1;
  10.     //获取读锁计数
  11.     int r = sharedCount(c);
  12.     //尝试获取读锁，多个读线程只有一个会成功，不成功的会进入到 fullTryAcquireShared() 进行重试
  13.     //readerShouldBlock() 在非公平的实现中，会去查看AQS队列中是否存在一个元素，判断第一个元素是否正在尝试获取写锁
  14.     //如果不是，则判断获取读锁的线程是否达到了最大值
  15.     //最后执行CAS操作将高16位的值 + 1
  16.     if (!readerShouldBlock() &&
  17.         r < MAX_COUNT &&
  18.         compareAndSetState(c, c + SHARED_UNIT)) {
  19.         //如果还没有线程获取过读锁
  20.         //则记录当前线程位读锁的第一个获取线程
  21.         if (r == 0) {
  22.             firstReader = current;
  23.             firstReaderHoldCount = 1;
  24.             //如果当前线程是第一个获取读锁的线程，则可重入次数 +1
  25.         } else if (firstReader == current) {
  26.             firstReaderHoldCount++;
  27.         } else {
  28.             //记录最后一个获取读锁的线程或记录其他线程读锁的可重入次数
  29.             HoldCounter rh = cachedHoldCounter;
  30.             if (rh == null || rh.tid != getThreadId(current))
  31.                 cachedHoldCounter = rh = readHolds.get();
  32.             else if (rh.count == 0)
  33.                 readHolds.set(rh);
  34.             rh.count++;
  35.         }
  36.         return 1;
  37.     }
  38.     //自旋获取，类似于 tryAquireShared()
  39.     return fullTryAcquireShared(current);
  40. }

3.2 void unlock()

  1. public void unlock() {
  2.     sync.releaseShared(1);
  3. }
  5. public final boolean releaseShared(int arg) {
  6.     if (tryReleaseShared(arg)) {
  7.         doReleaseShared();
  8.         return true;
  9.     }
  10.     return false;
  11. }

同样是Sync自定义的 tryReleaseShared()

  1. protected final boolean tryReleaseShared(int unused) {
  2.     Thread current = Thread.currentThread();
  3.     //如果当前线程是第一个获取读锁的线程
  4.     if (firstReader == current) {
  5.         //且重入的次数为1，则 firstReader = null
  6.         if (firstReaderHoldCount == 1)
  7.             firstReader = null;
  8.         else
  9.             //否则 重入次数-1
  10.             firstReaderHoldCount--;
  11.     } else {
  13.         HoldCounter rh = cachedHoldCounter;
  14.         //如果当前线程不是最后一个获取读锁的线程
  15.         if (rh == null || rh.tid != getThreadId(current))
  16.             rh = readHolds.get();
  17.         //则从中间 readHolds 统计的重入次数 -1
  18.         int count = rh.count;
  19.         if (count <= 1) {
  20.             //如果 count = 1 ，则移除 HoldCounter 计数
  21.             readHolds.remove();
  22.             //如果 == 0则说明，无法进行 -1操作，抛出异常
  23.             if (count <= 0)
  24.                 throw unmatchedUnlockException();
  25.         }
  26.         //计数-1
  27.         --rh.count;
  28.     }
  29.     //自旋更新读锁的状态值
  30.     for (;;) {
  31.         int c = getState();
  32.         //相当于高16位的值进行了 -1操作
  33.         int nextc = c - SHARED_UNIT;
  34.         if (compareAndSetState(c, nextc))
  35.             //当前线程释放读锁成功，但是可能还有其他线程持有读锁，所以要判断 nextc == 0
  36.             //不为0则说明还有其他线程持有读锁，则读锁释放失败
  37.             //为0则说明没有其他线程持有读锁了，读锁释放成功
  38.             return nextc == 0;
  39.     }
  40. }

4.案例

使用ReentrantReadWriteLock完成一个线程安全的List

  1. import java.util.ArrayList;
  2. import java.util.List;
  3. import java.util.concurrent.locks.Lock;
  4. import java.util.concurrent.locks.ReentrantReadWriteLock;
  6. public class ReentrantList {
  8.     private List<String> seed = new ArrayList<>();
  10.     private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  12.     private final Lock readLock = lock.readLock();
  14.     private final Lock writeLock = lock.writeLock();
  17.     public void add(String e){
  18.         writeLock.lock();
  19.         try {
  20.             seed.add(e);
  21.         }finally {
  22.             writeLock.unlock();
  23.         }
  24.     }
  26.     public void remove(String e){
  27.         writeLock.lock();
  28.         try {
  29.             seed.remove(e);
  30.         }finally {
  31.             writeLock.unlock();
  32.         }
  33.     }
  35.     public String get(int index){
  36.         readLock.lock();
  37.         try {
  38.             return seed.get(index);
  39.         }finally {
  40.             readLock.unlock();
  41.         }
  42.     }
  43. }

5.总结

  1.     ReentrantReadWriteLock的底层原理就是使用AQS实现的，ReentrantReadWriteLock使用高16位和低16位分别表示读锁和写锁的个数
  2.   通过CAS对其进行操作实现了读写分离，主要在读多写少的场景下使用