1. Redis实现分布式锁

分布式锁演进-阶段一

image-20200724214015598

  1. 多个客户端同时获取锁（setnx）
  2. 获取成功，执行业务逻辑，执行完成释放锁（del）
  3. 其他客户端等待重试

代码：

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
public void testLock() {
    // 1. 从redis中获取锁,setnx
    Boolean lock = this.redisTemplate.opsForValue().setIfAbsent("lock", "111");
    if (lock) {
        // 查询redis中的num值
        String value = this.redisTemplate.opsForValue().get("num");
        // 没有该值return
        if (StringUtils.isBlank(value)){
            return ;
        }
        // 有值就转成成int
        int num = Integer.parseInt(value);
        // 把redis中的num值+1
        this.redisTemplate.opsForValue().set("num", String.valueOf(++num));

        // 2. 释放锁 del
        this.redisTemplate.delete("lock");
    } else {
        // 3. 每隔1秒钟回调一次，再次尝试获取锁
        try {
            Thread.sleep(1000);
            testLock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

问题：setnx刚好获取到锁，业务逻辑出现异常，导致锁无法释放

解决：设置过期时间，自动释放锁。

分布式锁演进-阶段二

image-20200724214039789

分布式锁演进-阶段三

image-20200724214141947

分布式锁演进-阶段四

image-20200724214228424

分布式锁演进-阶段五-最终形态

image-20200724214247087

删除LUA脚本：

1
if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end

实现：

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
public void testLock() {
    // 1. 从redis中获取锁,setnx
    String uuid = UUID.randomUUID().toString();
    Boolean lock = this.redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
    if (lock) {
        // 查询redis中的num值
        String value = this.redisTemplate.opsForValue().get("num");
        // 没有该值return
        if (StringUtils.isBlank(value)){
            return ;
        }
        // 有值就转成成int
        int num = Integer.parseInt(value);
        // 把redis中的num值+1
        this.redisTemplate.opsForValue().set("num", String.valueOf(++num));

        // 2. 释放锁 del
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        this.redisTemplate.execute(new DefaultRedisScript<>(script), Arrays.asList("lock"), uuid);
        //            if (StringUtils.equals(uuid, this.redisTemplate.opsForValue().get("num"))) {
        //                this.redisTemplate.delete("lock");
        //            }
    } else {
        // 3. 每隔1秒钟回调一次，再次尝试获取锁
        try {
            Thread.sleep(1000);
            testLock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

2. 分布式锁之Redisson

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

1568176834908

官方文档地址：https://github.com/redisson/redisson/wiki

2.1. 快速入门

  1. 引入依赖
1
2
3
4
5
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.11.2</version>
</dependency>
  1. 添加配置
1
2
3
4
5
6
7
8
9
10
11
@Configuration
public class RedissonConfig {

    @Bean
    public RedissonClient redissonClient(){
        Config config = new Config();
        // 可以用"rediss://"来启用SSL连接
        config.useSingleServer().setAddress("redis://172.16.116.100:6379");
        return Redisson.create(config);
    }
}
  1. 代码实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
@Autowired
private RedissonClient redissonClient;

@Override
public void testLock() {

    RLock lock = this.redissonClient.getLock("lock"); // 只要锁的名称相同就是同一把锁
    lock.lock(); // 加锁

    // 查询redis中的num值
    String value = this.redisTemplate.opsForValue().get("num");
    // 没有该值return
    if (StringUtils.isBlank(value)) {
        return;
    }
    // 有值就转成成int
    int num = Integer.parseInt(value);
    // 把redis中的num值+1
    this.redisTemplate.opsForValue().set("num", String.valueOf(++num));

    lock.unlock(); // 解锁
}

使用ab压力测试，查看redis内容：

1568180869665

2.2. 可重入锁（Reentrant Lock）

基于Redis的Redisson分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口。

大家都知道，如果负责储存这个分布式锁的Redisson节点宕机以后，而且这个锁正好处于锁住的状态时，这个锁会出现锁死的状态。为了避免这种情况的发生，Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期。默认情况下，看门狗的检查锁的超时时间是30秒钟，也可以通过修改Config.lockWatchdogTimeout来另行指定。

另外Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间。超过这个时间后锁便自动解开了。

快速入门使用的就是可重入锁。也是最常使用的锁。

最常见的使用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
RLock lock = redisson.getLock("anyLock");
// 最常使用
lock.lock();


// 加锁以后10秒钟自动解锁
// 无需调用unlock方法手动解锁
lock.lock(10, TimeUnit.SECONDS);


// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
   try {
     ...
   } finally {
       lock.unlock();
   }
}

改造程序：

1586069265237

重启后在浏览器测试：

1586069552062

在这10s期间，可以在redis客户端看到lock锁的内容：

1568181219597

2.3. 读写锁（ReadWriteLock）

基于Redis的Redisson分布式可重入读写锁RReadWriteLock Java对象实现了java.util.concurrent.locks.ReadWriteLock接口。其中读锁和写锁都继承了RLock接口。

分布式可重入读写锁允许同时有多个读锁和一个写锁处于加锁状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
RReadWriteLock rwlock = redisson.getReadWriteLock("anyRWLock");
// 最常见的使用方法
rwlock.readLock().lock();
// 或
rwlock.writeLock().lock();


// 10秒钟以后自动解锁
// 无需调用unlock方法手动解锁
rwlock.readLock().lock(10, TimeUnit.SECONDS);
// 或
rwlock.writeLock().lock(10, TimeUnit.SECONDS);

// 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
boolean res = rwlock.readLock().tryLock(100, 10, TimeUnit.SECONDS);
// 或
boolean res = rwlock.writeLock().tryLock(100, 10, TimeUnit.SECONDS);
...
lock.unlock();

IndexController中的两个方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
@GetMapping("read")
public ResponseVo<String> read(){
    String msg = indexService.readLock();

    return ResponseVo.ok(msg);
}

@GetMapping("write")
public ResponseVo<String> write(){
    String msg = indexService.writeLock();

    return ResponseVo.ok(msg);
}

IndexService接口方法：注意保证锁的名称一致，才能使用同一把锁

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
public String readLock() {
    // 初始化读写锁
    RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("readwriteLock");
    RLock rLock = readWriteLock.readLock(); // 获取读锁

    rLock.lock(10, TimeUnit.SECONDS); // 加10s锁

    String msg = this.redisTemplate.opsForValue().get("msg");

    //rLock.unlock(); // 解锁
    return msg;
}

public String writeLock() {
    // 初始化读写锁
    RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("readwriteLock");
    RLock rLock = readWriteLock.writeLock(); // 获取写锁

    rLock.lock(10, TimeUnit.SECONDS); // 加10s锁

    this.redisTemplate.opsForValue().set("msg", UUID.randomUUID().toString());

    //rLock.unlock(); // 解锁
    return "成功写入了内容。。。。。。";
}

打开开两个浏览器窗口测试：

  • 同时访问写：一个写完之后，等待一会儿（约10s），另一个写开始

  • 同时访问读：不用等待

  • 先写后读：读要等待（约10s）写完成

  • 先读后写：写要等待（约10s）读完成

2.4. 信号量（Semaphore）和闭锁（CountDownLatch）

基于Redis的Redisson的分布式信号量（Semaphore）Java对象RSemaphore采用了与java.util.concurrent.Semaphore相似的接口和用法。

1
2
3
4
5
6
7
8
9
RSemaphore semaphore = redisson.getSemaphore("semaphore");
semaphore.acquire();
//或
semaphore.acquire(23);
semaphore.tryAcquire();
semaphore.tryAcquire(23, TimeUnit.SECONDS);

// 释放资源
semaphore.release();

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

1
2
3
4
5
6
7
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.trySetCount(1);
latch.await();

// 在其他线程或其他JVM里
RCountDownLatch latch = redisson.getCountDownLatch("anyCountDownLatch");
latch.countDown();

需要两个线程，一个等待。一个计数countDown

演示代码

IndexController：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
/**
     * 等待
     * @return
     */
@GetMapping("latch")
public ResponseVo<Object> countDownLatch(){

    String msg = indexService.latch();

    return ResponseVo.ok(msg);
}

/**
     * 计数
     * @return
     */
@GetMapping("out")
public ResponseVo<Object> out(){

    String msg = indexService.countDown();

    return ResponseVo.ok(msg);
}

IndexService：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public String latch() {
    RCountDownLatch countDownLatch = this.redissonClient.getCountDownLatch("countdown");
    try {
        countDownLatch.trySetCount(6);
        countDownLatch.await();

        return "关门了。。。。。";
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return null;
}

public String countDown() {
    RCountDownLatch countDownLatch = this.redissonClient.getCountDownLatch("countdown");

    countDownLatch.countDown();
    return "出来了一个人。。。";
}

重启测试，打开两个页面：当第二个请求执行6次之后，第一个请求才会执行。

1586070347231