Redis 分布式锁

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的JavaAPI并不能提供分布式锁的能力。为了解决这个问题就需要一种跨VM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题!

分布式锁的主流的实现方案：

基于数据库实现分布式锁
基于缓存（Redis等）
基于zookeeper

每一种分布式锁解决方案都有各自的优缺点：

性能：Redis最高
可靠性：zookeeper最高

这里，我们就基于redis实现分布式锁

设置锁和过期时间

命令：setnx KEY_NAME VALUE

将 key 的值设为 value，当且仅当 key 不存在。若给定的 key 已经存在，则 SETNX 不做任何动作。

如果锁一直不释放，可以设置key过期时间，过期后会释放锁

expire user 10

如果上锁后出现异常，无法设置过期时间：可以上锁的同时设置过期时间；

set user 10 nx ex [expire time]

java代码实现，我们根据之前 springboot整合redis搭建环境

@GetMapping("testLOck")
    public void testLock(){
        // 获取锁，setnx,并设置锁的过期时间
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock","111",3, TimeUnit.SECONDS);
        // 获取锁成功，查询num值
        if (lock){
            Object value = redisTemplate.opsForValue().get("num");
            //panduan
            if (StringUtils.isEmpty(value)){
                return;
            }
            // 有值就转换成
            int num = Integer.parseInt(value+"");
            // 把redis的num加1
            redisTemplate.opsForValue().set("num",++num);
            // 释放锁del
            redisTemplate.delete("lock");

        }else{
            //获取锁失败，每隔0.1秒再获取
            try{
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

使用ab工具模拟并发测试

centos7 安装

yum install httpd-tools

ab -n 1000 -c 100 http://10.211.28.124:8080/redisTest/testLock

UUID防止误删

@GetMapping("testLock")
    public void testLock(){
        //生成uuid
        String uuid = UUID.randomUUID().toString();

        // 获取锁，setnx,并设置锁的过期时间
        Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock",uuid,3, TimeUnit.SECONDS);
        // 获取锁成功，查询num值
        if (lock){
            Object value = redisTemplate.opsForValue().get("num");
            //panduan
            if (StringUtils.isEmpty(value)){
                return;
            }
            // 有值就转换成
            int num = Integer.parseInt(value+"");
            // 把redis的num加1
            redisTemplate.opsForValue().set("num",++num);
            // 释放锁del
            // 判断比较uuid的值是否一样
            String  lockUuid = (String) redisTemplate.opsForValue().get("lock");
            if (lockUuid.equals(uuid)){
                redisTemplate.delete("lock");
            }
        }else{
            //获取锁失败，每隔0.1秒再获取
            try{
                Thread.sleep(100);
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

这样做的问题是：删除操作缺乏原子性

LUA保证删除的原子性

代码如下

@GetMapping("testLockLua")
public void testLockLua() {
    //1 声明一个uuid ,将做为一个value 放入我们的key所对应的值中
    String uuid = UUID.randomUUID().toString();
    //2 定义一个锁：lua 脚本可以使用同一把锁，来实现删除！
    String skuId = "25"; // 访问skuId 为25号的商品 100008348542
    String locKey = "lock:" + skuId; // 锁住的是每个商品的数据

    // 3 获取锁
    Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);

    // 第一种： lock 与过期时间中间不写任何的代码。
    // redisTemplate.expire("lock",10, TimeUnit.SECONDS);//设置过期时间
    // 如果true
    if (lock) {
        // 执行的业务逻辑开始
        // 获取缓存中的num 数据
        Object value = redisTemplate.opsForValue().get("num");
        // 如果是空直接返回
        if (StringUtils.isEmpty(value)) {
            return;
        }
        // 不是空 如果说在这出现了异常！ 那么delete 就删除失败！ 也就是说锁永远存在！
        int num = Integer.parseInt(value + "");
        // 使num 每次+1 放入缓存
        redisTemplate.opsForValue().set("num", String.valueOf(++num));
        /*使用lua脚本来锁*/
        // 定义lua 脚本
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 使用redis执行lua执行
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(script);
        // 设置一下返回值类型 为Long
        // 因为删除判断的时候，返回的0,给其封装为数据类型。如果不封装那么默认返回String 类型，
        // 那么返回字符串与0 会有发生错误。
        redisScript.setResultType(Long.class);
        // 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
        redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
    } else {
        // 其他线程等待
        try {
            // 睡眠
            Thread.sleep(1000);
            // 睡醒了之后，调用方法。
            testLockLua();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}