并发锁之二：ReentrantReadWriteLock读写锁

Dear 丶 2023-06-02 15:58 114阅读 0赞

一、简介

　　读写锁是一种特殊的自旋锁，它把对共享资源对访问者划分成了读者和写者，读者只对共享资源进行访问，写者则是对共享资源进行写操作。读写锁在ReentrantLock上进行了拓展使得该锁更适合读操作远远大于写操作对场景。一个读写锁同时只能存在一个写锁但是可以存在多个读锁，但不能同时存在写锁和读锁。

　　如果读写锁当前没有读者，也没有写者，那么写者可以立刻获的读写锁，否则必须自旋，直到没有任何的写锁或者读锁存在。如果读写锁没有写锁，那么读锁可以立马获取，否则必须等待写锁释放。(但是有一个例外，就是读写锁中的锁降级操作，当同一个线程获取写锁后，在写锁没有释放的情况下可以获取读锁再释放读锁这就是锁降级的一个过程)

二、简单示例

复制代码

  1. 1 package cn.memedai;
  2.  2 
  3.  3 import java.util.Random;
  4.  4 import java.util.concurrent.ExecutorService;
  5.  5 import java.util.concurrent.Executors;
  6.  6 import java.util.concurrent.locks.ReadWriteLock;
  7.  7 
  8.  8 /**
  9.  9  * 读写锁Demo
  10. 10  */
  11. 11 public class ReentrantReadWriteLockDemo {
  12. 12 
  13. 13     class MyObject {
  14. 14         private Object object;
  15. 15 
  16. 16         private ReadWriteLock lock = new java.util.concurrent.locks.ReentrantReadWriteLock();
  17. 17 
  18. 18         public void get() throws InterruptedException {
  19. 19             lock.readLock().lock();//上读锁
  20. 20             try {
  21. 21                 System.out.println(Thread.currentThread().getName() + "准备读取数据");
  22. 22                 Thread.sleep(new Random().nextInt(1000));
  23. 23                 System.out.println(Thread.currentThread().getName() + "读数据为：" + this.object);
  24. 24             } finally {
  25. 25                 lock.readLock().unlock();
  26. 26             }
  27. 27         }
  28. 28 
  29. 29         public void put(Object object) throws InterruptedException {
  30. 30             lock.writeLock().lock();
  31. 31             try {
  32. 32                 System.out.println(Thread.currentThread().getName() + "准备写数据");
  33. 33                 Thread.sleep(new Random().nextInt(1000));
  34. 34                 this.object = object;
  35. 35                 System.out.println(Thread.currentThread().getName() + "写数据为" + this.object);
  36. 36             } finally {
  37. 37                 lock.writeLock().unlock();
  38. 38             }
  39. 39         }
  40. 40     }
  41. 41 
  42. 42     public static void main(String[] args) {
  43. 43         final MyObject myObject = new ReentrantReadWriteLockDemo().new MyObject();
  44. 44         ExecutorService executorService = Executors.newCachedThreadPool();
  45. 45         for (int i = 0; i < 3; i++) {
  46. 46             executorService.execute(new Runnable() {
  47. 47                 @Override
  48. 48                 public void run() {
  49. 49                     for (int j = 0; j < 3; j++) {
  50. 50 
  51. 51                         try {
  52. 52                             myObject.put(new Random().nextInt(1000));//写操作
  53. 53                         } catch (InterruptedException e) {
  54. 54                             e.printStackTrace();
  55. 55                         }
  56. 56                     }
  57. 57                 }
  58. 58             });
  59. 59         }
  60. 60 
  61. 61         for (int i = 0; i < 3; i++) {
  62. 62             executorService.execute(new Runnable() {
  63. 63                 @Override
  64. 64                 public void run() {
  65. 65                     for (int j = 0; j < 3; j++) {
  66. 66                         try {
  67. 67                             myObject.get();//多个线程读取操作
  68. 68                         } catch (InterruptedException e) {
  69. 69                             e.printStackTrace();
  70. 70                         }
  71. 71                     }
  72. 72                 }
  73. 73             });
  74. 74         }
  75. 75 
  76. 76         executorService.shutdown();
  77. 77     }
  78. 78 }

复制代码

下面是代码运行结果的一种：

复制代码

  1. pool-1-thread-1准备写数据
  2. pool-1-thread-1写数据为513
  3. pool-1-thread-1准备写数据
  4. pool-1-thread-1写数据为173
  5. pool-1-thread-1准备写数据
  6. pool-1-thread-1写数据为487
  7. pool-1-thread-2准备写数据
  8. pool-1-thread-2写数据为89
  9. pool-1-thread-2准备写数据
  10. pool-1-thread-2写数据为814
  11. pool-1-thread-2准备写数据
  12. pool-1-thread-2写数据为1
  13. pool-1-thread-3准备写数据
  14. pool-1-thread-3写数据为701
  15. pool-1-thread-3准备写数据
  16. pool-1-thread-3写数据为503
  17. pool-1-thread-3准备写数据
  18. pool-1-thread-3写数据为694
  19. pool-1-thread-4准备读取数据
  20. pool-1-thread-5准备读取数据
  21. pool-1-thread-6准备读取数据
  22. pool-1-thread-4读数据为：694
  23. pool-1-thread-4准备读取数据
  24. pool-1-thread-4读数据为：694
  25. pool-1-thread-4准备读取数据
  26. pool-1-thread-6读数据为：694
  27. pool-1-thread-6准备读取数据
  28. pool-1-thread-5读数据为：694
  29. pool-1-thread-5准备读取数据
  30. pool-1-thread-6读数据为：694
  31. pool-1-thread-6准备读取数据
  32. pool-1-thread-4读数据为：694
  33. pool-1-thread-5读数据为：694
  34. pool-1-thread-5准备读取数据
  35. pool-1-thread-6读数据为：694
  36. pool-1-thread-5读数据为：694

复制代码

从数据中也可以发现一开始读取的数据可能不一样，但是你会发现下面的时候线程4和线程5、线程6之间的读取的数据都是一样的，这就是共享读的特性。

三、实现原理

ReentrantReadWriteLock的基本原理和ReentrantLock没有很大的区别，只不过在ReentantLock的基础上拓展了两个不同类型的锁，读锁和写锁。首先可以看一下ReentrantReadWriteLock的内部结构：

format_png

内部维护了一个ReadLock和一个WriteLock，整个类的附加功能也就是通过这两个内部类实现的。

那么内部又是怎么实现这个读锁和写锁的呢。由于一个类既要维护读锁又要维护写锁，那么这两个锁的状态又是如何区分的。在ReentrantReadWriteLock对象内部维护了一个读写状态：

format_png 1

读写锁依赖自定义同步器实现同步功能，读写状态也就是同步器的同步状态。读写锁将整形变量切分成两部分，高16位表示读，低16位表示写：

format_png 2

读写锁的状态低16位为写锁，高16位为读锁

读写锁通过位运算计算各自的同步状态。假设当前同步状态的值为c，写状态就为c&0x0000FFFF，读状态为c >>> 16（无符号位补0右移16位）。当写状态增加1状态变为c+1，当读状态增加1时，状态编码就是c+（1 <<< 16）。

怎么维护读写状态的已经了解了，那么就可以开始了解具体怎么样实现的多个线程可以读，一个线程写的情况。

首先介绍的是ReadLock获取锁的过程

lock()：获取读锁方法

  1. 1 　　　　 public void lock() {
  2. 2             sync.acquireShared(1);//自定义实现的获取锁方式
  3. 3       }

acquireShared(int arg)：这是一个获取共享锁的方法

复制代码

  1. 1        protected final int tryAcquireShared(int unused) {
  2. 17             Thread current = Thread.currentThread();//获取当前线程
  3. 18             int c = getState();//获取锁状态
  4. 19             if (exclusiveCount(c) != 0 &&
  5. 20                 getExclusiveOwnerThread() != current)//如果获取锁的不是当前线程，并且由独占式锁的存在就不去获取，这里会发现必须同时满足两个条件才能判断其不能获取读锁这也会后面的锁降级做了准备
  6. 21                 return -1;
  7. 22             int r = sharedCount(c);//获取当前共享资源的数量
  8. 23             if (!readerShouldBlock() &&
  9. 24                 r < MAX_COUNT &&
  10. 25                 compareAndSetState(c, c + SHARED_UNIT)) {//代表可以获取读锁
  11. 26                 if (r == 0) {//如果当前没有线程获取读锁
  12. 27                     firstReader = current;//当前线程是第一个读锁获取者
  13. 28                     firstReaderHoldCount = 1;//在计数器上加1
  14. 29                 } else if (firstReader == current) {
  15. 30                     firstReaderHoldCount++;//代表重入锁计数器累加
  16. 31                 } else {
  17. 　　　　　　　　　　　　　　//内部定义的线程记录缓存
  18. 32                     HoldCounter rh = cachedHoldCounter;//HoldCounter主要是一个类用来记录线程已经线程获取锁的数量
  19. 33                     if (rh == null || rh.tid != current.getId())//如果不是当前线程
  20. 34                         cachedHoldCounter = rh = readHolds.get();//从每个线程的本地变量ThreadLocal中获取
  21. 35                     else if (rh.count == 0)//如果记录为0初始值设置
  22. 36                         readHolds.set(rh);//设置记录
  23. 37                     rh.count++;//自增
  24. 38                 }
  25. 39                 return 1;//返回1代表获取到了同步状态
  26. 40             }
  27. 41             return fullTryAcquireShared(current);//用来处理CAS设置状态失败的和tryAcquireShared非阻塞获取读锁失败的
  28. 42         }

复制代码

内部运用到了ThreadLocal线程本地对象，将每个线程获取锁的次数保存到每个线程内部，这样释放锁的时候就不会影响到其它的线程。

fullTryAcquireShared(Thread current)：此方法用于处理在获取读锁过程中CAS设置状态失败的和非阻塞获取读锁失败的线程

复制代码

  1. 1 　　　　　　final int fullTryAcquireShared(Thread current) {
  2.  2             //内部线程记录器
  3.  8             HoldCounter rh = null;
  4.  9             for (;;) {
  5. 10                 int c = getState();//同步状态
  6. 11                 if (exclusiveCount(c) != 0) {//代表存在独占锁
  7. 12                     if (getExclusiveOwnerThread() != current)//获取独占锁的线程不是当前线程返回失败
  8. 13                         return -1;
  9. 16                 } else if (readerShouldBlock()) {//判断读锁是否应该被阻塞
  10. 18                     if (firstReader == current) {
  11. 20                     } else {
  12. 21                         if (rh == null) {//为null
  13. 22                             rh = cachedHoldCounter;//从缓存中进行获取
  14. 23                             if (rh == null || rh.tid != current.getId()) {
  15. 24                                 rh = readHolds.get();//获取线程内部计数状态
  16. 25                                 if (rh.count == 0)
  17. 26                                     readHolds.remove();//移除
  18. 27                             }
  19. 28                         }
  20. 29                         if (rh.count == 0)//如果内部计数为0代表获取失败
  21. 30                             return -1;
  22. 31                     }
  23. 32                 }
  24. 33                 if (sharedCount(c) == MAX_COUNT)
  25. 34                     throw new Error("Maximum lock count exceeded");
  26. 35                 if (compareAndSetState(c, c + SHARED_UNIT)) {//CAS设置成功
  27. 36                     if (sharedCount(c) == 0) {
  28. 37                         firstReader = current;//代表为第一个获取读锁
  29. 38                         firstReaderHoldCount = 1;
  30. 39                     } else if (firstReader == current) {
  31. 40                         firstReaderHoldCount++;//重入锁
  32. 41                     } else {
  33. 42                         if (rh == null)
  34. 43                             rh = cachedHoldCounter;
  35. 44                         if (rh == null || rh.tid != current.getId())
  36. 45                             rh = readHolds.get();
  37. 46                         else if (rh.count == 0)
  38. 47                             readHolds.set(rh);
  39. 48                         rh.count++;
  40. 49                         cachedHoldCounter = rh; //将当前多少读锁记录下来
  41. 50                     }
  42. 51                     return 1;//返回获取同步状态成功
  43. 52                 }
  44. 53             }
  45. 54         }

复制代码

　　　分析完上面的方法可以总结一下获取读锁的过程：首先读写锁中读状态为所有线程获取读锁的次数，由于是可重入锁，又因为每个锁获取的读锁的次数由每个锁的本地变量ThreadLocal对象去保存因此增加了读取获取的流程难度，在每次获取读锁之前都会进行一次判断是否存在独占式写锁，如果存在独占式写锁就直接返回获取失败，进入同步队列中。如果当前没有写锁被获取，则线程可以获取读锁，由于共享锁的存在，每次获取都会判断线程的类型，以便每个线程获取同步状态的时候都在其对应的本地变量上进行自增操作。

lock(int arg)：写锁的获取








1

2

3

public void lock() {

       sync.acquire(1);//AQS独占式获取锁

   }

　　

tryAcquire(int arg)：独占式的获取写锁

复制代码

  1. 1 　　　　protected final boolean tryAcquire(int acquires) {
  2. 13             Thread current = Thread.currentThread();//获取当前线程
  3. 14             int c = getState();//获取同步状态值
  4. 15             int w = exclusiveCount(c);//获取独占式资源值
  5. 16             if (c != 0) {//已经有线程获取了
  6. 　　　　　　　　　　  //代表已经存在读锁，或者当前线程不是获取到写锁的线程
  7. 18                 if (w == 0 || current != getExclusiveOwnerThread())
  8. 19                     return false;//获取失败
  9. 20                 if (w + exclusiveCount(acquires) > MAX_COUNT)
  10. 21                     throw new Error("Maximum lock count exceeded");
  11. 22                 //设置同步状态
  12. 23                 setState(c + acquires);
  13. 24                 return true;
  14. 25             }
  15. 26             if (writerShouldBlock() ||
  16. 27                 !compareAndSetState(c, c + acquires))//判断当前写锁线程是否应该阻塞，这里会有公平锁和非公平锁之间的区分
  17. 28                 return false;
  18. 29             setExclusiveOwnerThread(current);//设置为当前线程
  19. 30             return true;
  20. 31         }

复制代码

获取写锁相比获取读锁就简单了很多，在获取读锁之前只需要判断当前是否存在读锁，如果存在读锁那么获取失败，进而再判断获取写锁的线程是否为当前线程如果不是也就是失败否则就是重入锁在已有的状态值上进行自增

unlock()：读锁释放

  1. 　　　　　public  void unlock() {
  2.             sync.releaseShared(1);//AQS释放共享锁操作
  3.         }

tryReleaseShared(int arg)：释放共享锁　　

复制代码

  1. 1 　　　　protected final boolean tryReleaseShared(int unused) {
  2.  2             Thread current = Thread.currentThread();//获取当前线程
  3.  3             if (firstReader == current) {//如果当前线程就是获取读锁的线程
  4.  5                 if (firstReaderHoldCount == 1)//如果此时获取资源为1
  5.  6                     firstReader = null;//直接赋值null
  6.  7                 else
  7.  8                     firstReaderHoldCount--;//否则计数器自减
  8.  9             } else {
  9. 　　　　　　　　　　　//其他线程
  10. 10                 HoldCounter rh = cachedHoldCounter;//获取本地计数器
  11. 11                 if (rh == null || rh.tid != current.getId())
  12. 12                     rh = readHolds.get();
  13. 13                 int count = rh.count;
  14. 14                 if (count <= 1) {//代表只获取了一次
  15. 15                     readHolds.remove();
  16. 16                     if (count <= 0)
  17. 17                         throw unmatchedUnlockException();
  18. 18                 }
  19. 19                 --rh.count;
  20. 20             }
  21. 21             for (;;) {
  22. 22                 int c = getState();
  23. 23                 int nextc = c - SHARED_UNIT;
  24. 24                 if (compareAndSetState(c, nextc))
  25. 28                     return nextc == 0;//代表已经全部释放
  26. 29             }
  27. 30         }

复制代码

释放锁的过程不难，但是有一个注意点，并不是释放一次就已经代表可以获取独占式写锁了，只有当同步状态的值为0的时候也就是代表既没有读锁存在也没有写锁存在才代表完全释放了读锁。

unlock()：释放写锁

  1. 1 　　　　　public void unlock() {
  2. 2             sync.release(1);//释放独占式同步状态
  3. 3         }

tryRelease(int arg)：释放独占式写锁

复制代码

  1. 1 　　　　 protected final boolean tryRelease(int releases) {
  2.  2             if (!isHeldExclusively())//判断是否
  3.  3                 throw new IllegalMonitorStateException();
  4.  4             int nextc = getState() - releases;//同步状态值自减
  5.  5             boolean free = exclusiveCount(nextc) == 0;//如果状态值为0代表全部释放
  6.  6             if (free)
  7.  7                 setExclusiveOwnerThread(null);
  8.  8             setState(nextc);
  9.  9             return free;
  10. 10         }

复制代码

写锁的释放相比读锁的释放简单很多，只需要判断当前的写锁是否全部释放完毕即可

四、读写锁之锁降级操作

　　 什么是锁降级，锁降级就是从写锁降级成为读锁。在当前线程拥有写锁的情况下，再次获取到读锁，随后释放写锁的过程就是锁降级。这里可以举个例子：

复制代码

  1. 1 public class CacheDemo {  
  2.  3      private Map<String, Object> cache = new HashMap<String, Object>();
  3.  4    
  4.  5      private ReadWriteLock rwl = new ReentrantReadWriteLock();
  5.  6 　　　public ReadLock rdl = rwl.readLock();
  6.  7 　　　public WriteLock wl = rwl.writeLock();
  7.  8         
  8.  9      public volatile boolean update = false;
  9. 10      public void processData(){
  10. 11          rdl.lock();//获取读锁
  11. 12          if(!update){
  12. 13              rdl.unlock();//释放读锁
  13. 14              wl.lock();//获取写锁
  14. 15              try{
  15. 16                 if(!update){
  16. 17                    update =true;
  17. 18                 }
  18. 19                 rdl.lock();//获取读锁
  19. 20              finally{
  20. 21                 wl.unlock();//释放写锁
  21. 22              }
  22. 23          }
  23. 24          try{
  24. 25           }finally{
  25. 26               rdl.unlock();//释放读锁
  26. 27           } 
  27. 29 }

复制代码

五、总结

　　　读写锁是在重入锁ReentrantLock基础上的一大改造，其通过在重入锁上维护一个读锁一个写锁实现的。对于ReentrantLock和ReentrantreadWriteLock的使用需要在开发者自己根据实际项目的情况而定。对于读写锁当读的操作远远大于写操作的时候会增加程序很高的并发量和吞吐量。虽说在高并发的情况下，读写锁的效率很高，但是同时又会存在一些问题，比如当读并发很高时读操作长时间占有锁，导致写锁长时间无法被获取而导致的线程饥饿问题，因此在JDK1.8中又在ReentrantReadWriteLock的基础上新增了一个读写并发锁StampLock。

发表评论

表情：
评论列表 （有 0 条评论，114人围观）

还没有评论，来说两句吧...

相关阅读

    相关 读写锁ReentrantReadWriteLock

    1. 写锁比读锁的优先级要高，拥有写锁之后还可以再获取读锁，但是拥有读锁的线程在释放前无法再获取写锁。 2. 允许锁降级，即从写锁降级为读锁，实现的步骤是：先获取写锁，再获

    红太狼红太狼/ 0 赞/ 350 阅读