【SimpleFunction系列二.3】Redisson分布式锁8种锁模式剖析

2. Redisson分布式锁8种锁模式剖析

【Redisson分布式锁—官方文档】

本文前置文：

【SimpleFunction系列二.1】渐进式理解Redis分布式锁

【SimpleFunction系列二.2】SpringBoot注解整合Redisson分布式锁

2.1 创建测试类

@SpringBootTest
public class LockModelTests {
    @Autowired
    RedissonClient redissonClient;
}
1
2
3
4
5

2.2 可重入锁（Reentrant Lock）

基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

可重入锁就是我们前面讲解的Redis分布式锁的Redisson实现，对于延时、过期等功能，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。

看门狗默认设置锁的超时时间是30S，在消耗时间超过1/3时对锁进行续租，我们也可以通过修改Config.lockWatchdogTimeout来另行指定。

demo：不设置等待时间会一直等待，不设置租赁时间会启动看门狗机制

@Test
void reentrantLock(){
  // 工作时间15s,注意看Redis中的ddl会在过去1/3时间后续约。
  new Thread(()->{
    RLock lock = redissonClient.getLock("testLock");
    lock.lock();
    doSomething(true, lock, 15L, Thread.currentThread().getName());
  }).start();
  // 租赁时间10s,设置了租赁时间，看门狗失效，工作时间15s,释放锁会报IllegalMonitorStateException
  new Thread(()->{
    RLock lock = redissonClient.getLock("testLock");
    lock.lock(5, TimeUnit.SECONDS);
    doSomething(true, lock, 20L, Thread.currentThread().getName());
  }).start();
  // 等待时间45S,未设置租赁时间，看门狗生效
  new Thread(()->{
    RLock lock = redissonClient.getLock("testLock");
    boolean lockFlag = false;
    try {
      lockFlag = lock.tryLock(45L, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    doSomething(lockFlag, lock, 15L, Thread.currentThread().getName());
  }).start();
  // 等待时间50S,租赁时间5S,工作时间10S,释放锁会报IllegalMonitorStateException
  new Thread(()->{
    RLock lock = redissonClient.getLock("testLock");
    boolean lockFlag = false;
    try {
      lockFlag = lock.tryLock(50L,5L, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    doSomething(lockFlag, lock, 10L, Thread.currentThread().getName());
  }).start();
  Scanner scanner = new Scanner(System.in);
  scanner.nextLine();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

2.3 公平锁（Fair Lock）

基于Redis的Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

公平锁保证了当多个Redisson客户端线程同事请求加锁时，优先分配给先发出请求的线程。所有请求线程会在一个队列中排队，当某个线程出现宕机时，Redisson 会等待5秒后继续下一个线程，也就是说如果前面有5个线程都处于等待状态，那么后面的线程会等待至少25秒。

在锁争夺较多时，程序利用大量cas去获取锁非常消耗CPU，可以考虑设置为公平锁。在锁的抢夺较少的时候就没必要设置成公平锁，毕竟公平锁也是需要成本的。

demo：主要理解公平二字，demo几乎同上

@Test
void fairLock() {
  // 租赁时间5s,设置了租赁时间，看门狗失效，工作时间50s,释放锁会报IllegalMonitorStateException
  new Thread(()->{
    RLock lock = redissonClient.getFairLock("testLock");
    lock.lock(5, TimeUnit.SECONDS);
    doSomething(true, lock, 50L, Thread.currentThread().getName());
  }).start();
  // 工作时间15s,注意看Redis中的ddl会在过去1/3时间后续约。
  new Thread(()->{
    RLock lock = redissonClient.getFairLock("testLock");
    sleepThread(1);
    lock.lock();
    doSomething(true, lock, 15L, Thread.currentThread().getName());
  }).start();
  // 等待时间45S,未设置租赁时间，看门狗生效
  new Thread(()->{
    RLock lock = redissonClient.getFairLock("testLock");
    boolean lockFlag = false;
    try {
      sleepThread(2);
      lockFlag = lock.tryLock(100L, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    doSomething(lockFlag, lock, 15L, Thread.currentThread().getName());
  }).start();
  // 等待时间50S,租赁时间5S,工作时间10S,释放锁会报IllegalMonitorStateException
  new Thread(()->{
    RLock lock = redissonClient.getFairLock("testLock");
    boolean lockFlag = false;
    try {
      sleepThread(3);
      lockFlag = lock.tryLock(50L,5L, TimeUnit.SECONDS);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
    doSomething(lockFlag, lock, 10L, Thread.currentThread().getName());
  }).start();
  Scanner scanner = new Scanner(System.in);
  scanner.nextLine();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

2.4 联锁（MultiLock）

基于Redis的Redisson分布式联锁RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例。

联锁指的是：同时对多个资源进行加锁操作，只有所有资源都加锁成功的时候，联锁才会成功。

@Test
void multiLockTest(){
  RLock lock = redissonClient.getFairLock("testLock");
  RLock lock2 = redissonClient.getFairLock("testLock2");
  RLock lock3 = redissonClient.getFairLock("testLock3");
  RedissonMultiLock multiLock = new RedissonMultiLock(lock, lock2, lock3);
  // 同时加锁：testLock testLock2 testLock3
  // 所有的锁都上锁成功才算成功。
  try {
    boolean tryLock = multiLock.tryLock(1, TimeUnit.SECONDS);
    doSomething(tryLock, multiLock, 10, Thread.currentThread().getName());
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2.5 红锁（RedLock）

基于Redis的Redisson红锁RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例。

与联锁比较相似，都是对多个资源进行加锁，但是红锁与连锁不同的是，红锁只需要在大部分资源加锁成功即可，大部分是指n/2+1。

@Test
void testRedLock(){
  new Thread(()->{
    RLock lock = redissonClient.getLock("testLock");
    lock.lock();
    doSomething(true, lock, 100, Thread.currentThread().getName());
  }).start();
  sleepThread(1);
  RLock lock = redissonClient.getLock("testLock");
  RLock lock2 = redissonClient.getLock("testLock2");
  RLock lock3 = redissonClient.getLock("testLock3");
  RedissonRedLock redLock = new RedissonRedLock(lock, lock2, lock3);
  boolean flag = redLock.tryLock();
  doSomething(flag, redLock, 10, Thread.currentThread().getName());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

2.6 读写锁（ReadWriteLock）

基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。

读写锁：允许多个线程同时读，但是只允许一个线程写，在线程获取到写锁的时候，其他写操作和读操作都会处于阻塞状态，读锁和写锁也是互斥的，所以在读的时候是不允许写的。

@Test
void readWriteLockTest(){
  new Thread(()->{
    RLock writeLock = redissonClient.getReadWriteLock("testLock").writeLock();
    writeLock.lock();
    doSomething(true,writeLock,10,Thread.currentThread().getName());
  }).start();
  sleepThread(1);
  new Thread(()->{
    RLock readLock = redissonClient.getReadWriteLock("testLock").readLock();
    readLock.lock();
    doSomething(true,readLock,10,Thread.currentThread().getName());
  }).start();
  new Thread(()->{
    RLock readLock = redissonClient.getReadWriteLock("testLock").readLock();
    readLock.lock();
    doSomething(true,readLock,10,Thread.currentThread().getName());
  }).start();
  Scanner scanner = new Scanner(System.in);
  scanner.nextLine();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

2.7 信号量（Semaphore）

基于Redis的Redisson的分布式信号量（Semaphore）Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。同时还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

信号量（英语：semaphore）又称为信号标，是一个同步对象，用于保持在0至指定最大值之间的一个计数值。当线程完成一次对该semaphore对象的等待（wait）时，该计数值减一；当线程完成一次对semaphore对象的释放（release）时，计数值加一。当计数值为0，则线程等待该semaphore对象不再能成功直至该semaphore对象变成signaled状态。semaphore对象的计数值大于0，为signaled状态；计数值等于0，为nonsignaled状态。——维基百科

@Test
void semaphoreLockTest(){
  RSemaphore testLock = redissonClient.getSemaphore("testLock");
  int permits = testLock.availablePermits();
  testLock.addPermits(2-permits);
  new Thread(()->{
    RSemaphore semaphore = redissonClient.getSemaphore("testLock");
    try {
      semaphore.acquire();
      System.out.println("获取到一个许可"+Thread.currentThread().getName());
      sleepThread(10);
      semaphore.release();
      System.out.println("释放一个许可"+Thread.currentThread().getName());
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }).start();
  new Thread(()->{
    RSemaphore semaphore = redissonClient.getSemaphore("testLock");
    try {
      semaphore.acquire();
      System.out.println("获取到一个许可"+Thread.currentThread().getName());
      sleepThread(10);
      semaphore.release();
      System.out.println("释放一个许可"+Thread.currentThread().getName());
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }).start();
  new Thread(()->{
    RSemaphore semaphore = redissonClient.getSemaphore("testLock");
    try {
      semaphore.acquire();
      System.out.println("获取到一个许可"+Thread.currentThread().getName());
      sleepThread(10);
      semaphore.release();
      System.out.println("释放一个许可"+Thread.currentThread().getName());
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }).start();
  Scanner scanner = new Scanner(System.in);
  scanner.nextLine();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

2.8 可过期信号量（PermitExpirableSemaphore）

基于Redis的Redisson可过期性信号量（PermitExpirableSemaphore）是在RSemaphore对象的基础上，为每个信号增加了一个过期时间。每个信号可以通过独立的ID来辨识，释放时只能通过提交这个ID才能释放。它提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

@Test
void permitExpirableSemaphoreLockTest() {
  RPermitExpirableSemaphore expirableSemaphore = redissonClient.getPermitExpirableSemaphore("PermitExpirableSemaphore");
  int permits = expirableSemaphore.availablePermits();
  expirableSemaphore.addPermits(2-permits);
  // 获取一个信号，有效期只有2秒钟。
  try {
    String permitId = expirableSemaphore.acquire(2, TimeUnit.SECONDS);
    System.out.println(permitId);
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
  try {
    sleepThread(5);
    String permitId = expirableSemaphore.acquire();
    System.out.println(permitId);
    expirableSemaphore.release(permitId);
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
  Scanner scanner = new Scanner(System.in);
  scanner.nextLine();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

2.9 闭锁（CountDownLatch）

基于Redisson的Redisson分布式闭锁（CountDownLatch）Java对象RCountDownLatch采用了与java.util.concurrent.CountDownLatch相似的接口和用法。

闭锁（Latch）：一种同步方法，可以延迟线程的进度直到线程到达某个终点状态。通俗的讲就是，一个闭锁相当于一扇大门，在大门打开之前所有线程都被阻断，一旦大门打开所有线程都将通过，但是一旦大门打开，所有线程都通过了，那么这个闭锁的状态就失效了，门的状态也就不能变了，只能是打开状态。也就是说闭锁的状态是一次性的，它确保在闭锁打开之前所有特定的活动都需要在闭锁打开之后才能完成。

@Test
void countDownLatchTest(){
  RCountDownLatch countDownLatch = redissonClient.getCountDownLatch("CountDownLatch");
  countDownLatch.trySetCount(1);
  new Thread(()->{
    RCountDownLatch downLatch = redissonClient.getCountDownLatch("CountDownLatch");
    sleepThread(10);
    downLatch.countDown();
  }).start();
  try {
    countDownLatch.await();
    System.out.println("主线程休眠结束");
  } catch (InterruptedException e) {
    throw new RuntimeException(e);
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

2.9 LockModelTests.java