分布式锁：有关于分布式锁合集(二)

本片文章为 分布式锁 相关的系列文章

​ 本系列文章 将从 缓存的使用 本地锁 ** 一直写到 **分布式锁 以及相关的一些 知识点 问题


上一篇文章说明了本地锁的在集群环境下的局限性 以及引出 分布式锁

上一篇文章：“分布式锁：有关于分布式锁集合（一）”


为什么？

​ 随着业务发展的需要 原来单体单机部署的系统现在演化成了分布式集群系统后 有用分布式系统 多线程 多进程 并且分布在不同机器上 使得原来的单机部署情况下的并发控制锁策略失效 单纯的JAVA API不能提供分布式锁的能力

​ 简而言之：本地锁 在集群环境下 锁不住共享资源 为了解决这个问题 就需要一种跨JVM的互斥机制来控制共享资源的访问 这就是分布式锁要解决的问题

有哪些？（解决方案）

  • 基于数据库实现分布式锁 性能低 要去操作IO
  • 基于缓存（Redis等） 性能最高
  • 基于Zookeeper 可靠性最高
    • Zookeeper 属性目录服务 是一个平衡二叉树 想要完成分布式事务 看的是每一个节点 每一个节点都是一个锁 如果当前有这个节点 那么久看做有这把锁

使用redis实现分布式锁

redis 的 set 命令

​ SET key value [EX seconds] [PX milliseconds] [NX|XX]

从 Redis 2.6.12 版本开始， SET 命令的行为可以通过一系列参数来修改：

  • EX second ：设置键的过期时间为 second秒。 SET key value EX second效果等同于 SETEX key second value 。
  • PX millisecond：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond效果等同于 PSETEX key millisecond value。
  • NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
  • XX ：只在键已经存在时，才对键进行设置操作。

如果 存在此key 那么 setnx 没有意义 所以 可以将setnx当做一把锁

如果 setnx 生效 那么 意味着 里面原来是不存在锁的 也就可以执行业务代码

否则 意味着 原来存在锁

例子：set k1 v1 ex 10 nx

​ 当k1不存在的时候 创建一个k1所对应的v1 过期时间为 10秒


分布式锁的 基本思想

  • 多个客户端同时获得锁（setnx）
  • 一个获得成功 执行业务逻辑(db中获取数据 放入缓存) 其他的线程失败 等待 自旋
  • 业务逻辑执行结束 释放锁资源
  • 其他客户端 等待重试

使用setnx实现分布式锁

​ 使用 setnx（k，v）命令

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
@Override
public void testLock() {
    // 1. 从redis中获取锁,setnx(锁，锁的名称)
    Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock", "111");
    // 判断锁 是否成功
    if (flag) {
        // 锁 获取成功
        // 执行业务逻辑
        // 直接获取缓存中的数据
        String value = (String)redisTemplate.opsForValue().get("num");
        // 没有该值return
        if (StringUtils.isBlank(value)){
            return ;
        }
        // 有值就转成成int
        int num = Integer.parseInt(value);
        // 把redis中的num值+1
        redisTemplate.opsForValue().set("num", String.valueOf(++num));
        // 释放锁资源
        redisTemplate.delete("lock");
    } else {
        // 获取锁失败 意味着被上锁了
        // 等待自旋 从新请求此方法
        try {
            Thread.sleep(500);
            testLock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

使用ab压力测试工具

启动网关微服务 使用网关 启用负载均衡 访问数据

ab -n 5000 -c
100 http://192.168.200.1/admin/product/test/testLock

使用redis桌面管理工具连接上此redis 最后显示的num数 为5000

也就是说 分布式锁 有效


但是 此方法 有没有什么缺陷呢？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 如果在执行业务逻辑的时候 出现异常
if (flag) {
            // 锁 获取成功
            // 执行业务逻辑
            // 直接获取缓存中的数据
            String value = (String)redisTemplate.opsForValue().get("num");
            // 没有该值return
            if (StringUtils.isBlank(value)){
                return ;
            }
    		// 出现异常 
    		int i = 1/0;
    		// 此后面的代码都不再执行 也就是说 将不会释放锁资源（死锁）
            // 有值就转成成int
            int num = Integer.parseInt(value);
            // 把redis中的num值+1
            redisTemplate.opsForValue().set("num", String.valueOf(++num));
            // 释放锁资源
            redisTemplate.delete("lock");
        }

上述代码 如果在业务执行流程的时候 出现异常 将会产生死锁

为了避免发生此情形 我们可以 加上一个过期时间 自动释放锁资源

给锁增加一个过期时间

redis中给锁添加过期时间有两种

  • setex()
  • expire()

这两种 该选谁？

​ 首先 想到通过expire设置过期时间 但是expire缺乏原子性 如果在setnx 和 expire之间出现异常 锁也无法释放

​ 所以 选择set时 指定过期时间

1
2
3
4
5
6
7
8
9
10
11
12
// 方案一： expire
// 伪代码
Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock", "111");
if (flag) {
		// 业务逻辑
    
    	// 如果此处发生异常
    	int i = 1/0;
    	// 下面的代码依然执行不到（缺乏原子性）白给
    
    	expire("lock",2); // 设置两秒过期
        } 
1
2
3
// 方案二： 
// set k1 v1 px 过期时间 nx
     Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock", "111",3, TimeUnit.SECONDS);

添加了过期时间 这样就没有其他的问题了吗？

如果 业务执行需要7秒钟 而我3秒钟时间到了 就释放了锁资源

后面该如何？是否会释放锁资源？7秒后释放的锁资源是谁的锁？

如此一来 相当于 第三个进来的线程 只执行了一秒 就被释放了锁资源

设置锁的时候 可能会产生 误删 他人锁的情况

给锁加个UUID（唯一标识）以防止 误删锁 操作

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
@Override
public void testLock() {
    // 1. 从redis中获取锁,setnx(锁，锁的名称)

    // 增加UUID 防止误删锁操作
    String uuid = UUID.randomUUID().toString();
    Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);
    // 判断锁 是否成功
    if (flag) {
        // 锁 获取成功
        // 执行业务逻辑
        // 直接获取缓存中的数据
        String value = (String)redisTemplate.opsForValue().get("num");
        // 没有该值return
        if (StringUtils.isBlank(value)){
            return ;
        }
        // 有值就转成成int
        int num = Integer.parseInt(value);
        // 把redis中的num值+1
        redisTemplate.opsForValue().set("num", String.valueOf(++num));

        // 判断 删除的锁 是不是我这把锁
        if (uuid.equals(redisTemplate.opsForValue().get("lock"))) {
            // 如果 key 对应的 value相同 则 是我的锁
            // 释放锁资源
            redisTemplate.delete("lock");
        }

    } else {
        // 获取锁失败 意味着被上锁了
        // 等待自旋 从新请求此方法
        try {
            Thread.sleep(500);
            testLock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

使用ab压力测试工具

启动网关微服务 使用网关 启用负载均衡 访问数据

ab -n 5000 -c
100 http://192.168.200.1/admin/product/test/testLock

使用redis桌面管理工具连接上此redis 最后显示的num数 为5000

也就是说 分布式锁 有效

那么 问题又来了 防止了误删锁 还有没有其他的问题呢？

小葵花妈妈课堂开课了 孩子总问为什么 怎么办？ 打一顿就好了

​ 会不会 存在这么一种情况 就是设置了过期时间 然后时间到了 锁释放了 正好有个线程来了 CPU还正好分配了 执行此方法

不知道我在说什么？ 行吧 看代码注释

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
45
46
47
48
49
@Override
   public void testLock() {
       // 1. 从redis中获取锁,setnx(锁，锁的名称)

       // 增加UUID 防止误删锁操作
       String uuid = UUID.randomUUID().toString();
       Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);
       // 判断锁 是否成功
       if (flag) {
           // 锁 获取成功
           // 执行业务逻辑
           // 直接获取缓存中的数据
           String value = (String)redisTemplate.opsForValue().get("num");
           // 没有该值return
           if (StringUtils.isBlank(value)){
               return ;
           }
           // 有值就转成成int
           int num = Integer.parseInt(value);
           // 把redis中的num值+1
           redisTemplate.opsForValue().set("num", String.valueOf(++num));

           // 判断 删除的锁 是不是我这把锁
           if (uuid.equals(redisTemplate.opsForValue().get("lock"))) {
               // 如果 key 对应的 value相同 则 是我的锁
               // 释放锁资源
               
               // 代码执行到此处 index1 if判断完成 执行delete()
               // 此时 锁的时间正好过期
               // 正好 此时有一个新的线程(index2)过来了
               // CPU 还将此线程分配给了现在这个
               // 锁已经过期 redis中已经没有锁了
               // index2获得了CPU资源 开始执行方法
               // index1执行删除 就会把index2的lock删除掉
               redisTemplate.delete("lock");
           }

       } else {
           // 获取锁失败 意味着被上锁了
           // 等待自旋 从新请求此方法
           try {
               Thread.sleep(500);
               testLock();
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
   }

这个东西是不是有点眼熟？

​ 本来是要删除index1的锁 结果删成了index2的锁

多套 生产者 消费者 产生的 虚假唤醒 场景！！！（找个时间 我将会整理一下有关于线程的知识点）

问题来了 如何保证原子性呢？


使用LUA脚本 保证删除的原子性

1
2
3
4
5
// LUA脚本
     		/*使用lua脚本来锁*/
            // 定义lua 脚本
            String script="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
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@Override
public void testLock() {
    // 1. 从redis中获取锁,setnx(锁，锁的名称)

    // 增加UUID 防止误删锁操作
    String uuid = UUID.randomUUID().toString();
    Boolean flag = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);
    // 判断锁 是否成功
    if (flag) {
        // 锁 获取成功
        // 执行业务逻辑
        // 直接获取缓存中的数据
        String value = (String)redisTemplate.opsForValue().get("num");
        // 没有该值return
        if (StringUtils.isBlank(value)){
            return ;
        }
        // 有值就转成成int
        int num = Integer.parseInt(value);
        // 把redis中的num值+1
        redisTemplate.opsForValue().set("num", String.valueOf(++num));
        // 上面一套方案存在缺陷 推荐使用LUA 脚本删除 保证原子性
        /*
            if redis.call("get",KEYS[1]) == ARGV[1]
            then
                return redis.call("del",KEYS[1])
            else
                return 0
            end
         */
        /*使用lua脚本来锁*/
        // 定义lua 脚本
        String script="if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

        // java中 如何调用并执行 LUA脚本呢？
        // 第一种传值
        // DefaultRedisScript<Object> redisScript = new DefaultRedisScript<>(script);
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        // 第二种传值
        redisScript.setScriptText(script);
        // 设置一下返回值类型 为Long
        // 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，那么返回字符串与0 会有发生错误。
        redisScript.setResultType(Long.class);
        // 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
        redisTemplate.execute(redisScript, Arrays.asList("lock"),uuid);

    } else {
        // 获取锁失败 意味着被上锁了
        // 等待自旋 从新请求此方法
        try {
            Thread.sleep(500);
            testLock();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

使用ab压力测试工具

ab -n 5000 -c
100 http://192.168.200.1:8216/admin/product/test/testLock

使用redis桌面管理工具连接上此redis 最后显示的num数 为5000

也就是说 分布式锁 有效

注意 LUA脚本测试的时候 并没有使用网关 那么使用网关会怎么样呢？(或者说 使用LUA脚本 在分布式集群状态下 就不会有问题了吗？)

凉凉

答案肯定是否定的

redis集群状态下的问题：

  • 客户端A从master获取到锁
  • 在master主机将锁同步到slave从机之前 主机宕了
  • slave从机晋升为master主机了
  • 客户端B取得了同一个资源被客户端A已经获取到的另外一个锁

安全失效

凉凉

结论 ：LUA脚本防止误删除锁 只 支持Redis单机版 不支持 集群

为了保证分布式锁可用 必须至少保证以下四点

  • 互斥性 在同一任何时刻 只要一个客户端能持有锁
  • 不会发生死锁 即 有一个客户端在持有锁的期间崩溃 而没有主动解锁的时候 也能保证其他客户端后续能加锁
  • 加锁 和 解锁 必须是同一个客户端 客户端自己不能把别人的锁给解了
  • 加锁 和 解锁 必须具有原子性

那么 LUA脚本也有缺陷 那我们应该怎么办呢？

Redisson redlock (红锁) 解决此问题

  • 此方法不靠谱
    • why？
    • redlock 认为 超过百分之五十的 加锁 成功 那么 就 加锁成功了
      • 也就意味着 此算法 将耗费大量 redis 连接 (也意味着 性能上将会有损耗)

最后的最后 我们引出了 Redisson 具体使用情况 下篇文章再讲解

赏

谢谢你请我吃糖果

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

一个懂java皮毛的弟弟<br><br>如有其他意见欢迎讨论