您现在的位置是：首页 > 数据库

当前栏目

基于redis的分布式锁

Redis 分布式分布式

2023-03-14 22:51:37 时间

为什么要使用分布式锁？

因为服务器使用了集群方案。词穷。。。

怎么使用分布式锁？

需求

实现一个查询数据库，在大于0的情况下减库存这样小小的功能。

测试：模拟100并发并看结果

基础代码

没有任何锁

 
    @RequestMapping("/reduce_stock")
    public String reduceStock() {
        //查数据库（redis）中库存数量
        Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        //判断库存
        if (stock > 0) {
            System.out.println("消费库存成功--->" + stock);
            //更新库存
            stock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
        } else {
            System.out.println("消费库存失败。。。");
        }
 
 
        return "helloworld";
    }

用测压工具测压结果：出现并发问题

有锁：给方法添加synchronized关键字

    @RequestMapping("/reduce_stock")
    public synchronized    String  reduceStock() {
        //查数据库（redis）中库存数量
        Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        //判断库存
        if (stock > 0) {
            System.out.println("消费库存成功--->" + stock);
            //更新库存
            stock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
        } else {
            System.out.println("消费库存失败。。。");
        }
        return "helloworld";
    }

单机测试结果：没有问题

分布式测试结果：出现线程安全问题

分析，如下图所示：

两个微服务，synchronized关键字只能锁住一个微服务，跨微服务是锁不住的。

就像你家的屋子A复制一份为B，A是否锁门和B是否锁门是没有关系的。

基于redis的分布式锁（理论+实操）

理论

基于redis的setnx命令实现分布式锁

setnx命令的特点是：当你第一次设置的时候会返回1，后面在设置的时候就会返回0（即修改失败），如下图所示

手写基于redis分布式锁（此处逻辑、理论大于实操）

一代代码

分析

---逻辑：先获取锁，如果获取锁，就继续；否则就不执行

---问题：容易出现死锁。如果我获取锁成功后在执行业务逻辑的过程中出现异常，则释放锁的过程就没有了，不释放锁就会引起死锁

   @RequestMapping("/reduce_stock")
    public String reduceStock() {
 
        //key的名称
        String lockKey = "lock";
        //setnx key value    加锁逻辑
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1");
        if (!aBoolean){
            return "fail";
        }
 
        //查数据库（redis）中库存数量
        Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        //判断库存
        if (stock > 0) {
            System.out.println("消费库存成功--->" + stock);
            //更新库存
            stock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
        } else {
            System.out.println("消费库存失败。。。");
        }
 
        // del key  释放锁逻辑
        stringRedisTemplate.delete(lockKey);
 
        return "helloworld";
    }

二代代码

分析：

---优点：在finally中释放锁，解决了死锁的问题

---问题：引起锁失效问题。看下面的代码，先加锁，如果加锁失败，返回，但是此时代码也会去执行finally中释放锁的功能，从而使别人加的锁失效。

    @RequestMapping("/reduce_stock")
    public String reduceStock() {
 
        //key的名称
        String lockKey = "lock";
 
        try {
            //setnx key value    加锁逻辑
            Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1");
            if (!aBoolean) {
                return "fail";
            }
 
            //查数据库（redis）中库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断库存
            if (stock > 0) {
                System.out.println("消费库存成功--->" + stock);
                //更新库存
                stock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
            } else {
                System.out.println("消费库存失败。。。");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
 
            // del key  释放锁逻辑
            stringRedisTemplate.delete(lockKey);
        }
 
 
        return "helloworld";
    }

三代代码

分析：

--优点：解决了锁失效问题

--问题：没有解决因为宕机而引起的死锁，如下图所示，微服务8082获取锁后在执行业务逻辑时系统宕机后就会引起死锁

    @RequestMapping("/reduce_stock")
    public String reduceStock() {
 
        //key的名称
        String lockKey = "lock";
 
 
        //setnx key value    加锁逻辑
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1");
        if (!aBoolean) {
            return "fail";
        }
 
        try {
 
 
            //查数据库（redis）中库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断库存
            if (stock > 0) {
                System.out.println("消费库存成功--->" + stock);
                //更新库存
                stock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
            } else {
                System.out.println("消费库存失败。。。");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
 
            // del key  释放锁逻辑
            stringRedisTemplate.delete(lockKey);
        }
 
 
        return "helloworld";
    }

四代代码

分析：

--优点：加锁逻辑时设置过期时间，可以解决三代代码的死锁问题，系统中断了我到时间就自动释放锁

@RequestMapping("/reduce_stock")
    public String reduceStock() {
 
        //key的名称
        String lockKey = "lock";
 
 
        //setnx key value    加锁逻辑
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1",30, TimeUnit.SECONDS);
        if (!aBoolean) {
            return "fail";
        }
 
        try {
 
 
            //查数据库（redis）中库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断库存
            if (stock > 0) {
                System.out.println("消费库存成功--->" + stock);
                //更新库存
                stock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
            } else {
                System.out.println("消费库存失败。。。");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
 
            // del key  释放锁逻辑
            stringRedisTemplate.delete(lockKey);
        }
 
 
        return "helloworld";
    }

五代代码

分析：

--优点：解决了四代代码的锁失效问题

--缺点：如下图所示，如果我设置失效时间是30，而我业务逻辑时间是35，在30-35之间是有两个线程同时访问，这与独占锁是矛盾的，所以此处存在问题。

@RequestMapping("/reduce_stock")
    public String reduceStock() {
 
        //key的名称
        String lockKey = "lock";
 
        //value的值
        String clientId = UUID.randomUUID().toString();
 
 
        //setnx key value    加锁逻辑
        Boolean aBoolean = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);
        if (!aBoolean) {
            return "fail";
        }
 
        try {
 
 
            //查数据库（redis）中库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断库存
            if (stock > 0) {
                System.out.println("消费库存成功--->" + stock);
                //更新库存
                stock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
            } else {
                System.out.println("消费库存失败。。。");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
 
            //只能释放自己加的锁
            if (clientId.equals(stringRedisTemplate.opsForValue().get("lock"))) {
                // del key  释放锁逻辑
                stringRedisTemplate.delete(lockKey);
            }
        }
 
 
        return "helloworld";
    }

瓶颈

我们现在的瓶颈就是超时时间的设置。

如果设置短了会出现五代代码的问题；如果设置长了，你不能保证业务逻辑一定会比你设置的时间短，就算你设置的时间长，10分钟，那万一系统中断10分钟内不能有业务处理，也是不可取的。

如果我们能动态修改这个超时时间，那就无敌了

其实还有一个问题，这短代码的逻辑是获取锁失败后直接返回，其实应该继续尝试获取

基于redisson的分布式锁

原理

实践

<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.10.0</version>
</dependency>

@Bean
    public Redisson redisson() {
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.10.30.146:6379").setDatabase(0).setPassword("123456");
        return (Redisson) Redisson.create(config);
    }

@RestController
public class DistributedLockController {
 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
 
    @Autowired
    private Redisson redisson;
 
 
    @RequestMapping("/reduce_stock")
    public String reduceStock() {
 
        //key的名称
        String lockKey = "lock";
 
        RLock lock = redisson.getLock(lockKey);
 
        lock.lock();
        try {
            //查数据库（redis）中库存数量
            Integer stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            //判断库存
            if (stock > 0) {
                System.out.println("消费库存成功--->" + stock);
                //更新库存
                stock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", String.valueOf(stock));
            } else {
                System.out.println("消费库存失败。。。");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
 
 
        return "helloworld";
    }
 
 
}

问题

向redis集群写数据的步骤是：

1）向master节点写数据

2） master节点返回

3）master节点同步到子节点

如果线程t1 获取锁，写入一个数据 1） 2）成功后此时master 节点掉线了，在子节点中选一个master,但是这个master是没有t1写的数据，此时此刻t2是可以获取到锁的，这个是redis做分布式锁的瑕疵。

redis是高性能分布式锁，zk是高可靠分布式锁，看你看重性能还是一致性了。

猜你喜欢

鲜为人知但很有用的 HTML 属性
在 Go 里用 CGO？这 7 个问题你要关注！
数据孤岛是业务效率的无声杀手
9款优秀的去中心化通讯软件 Matrix 的客户端
翻转再翻转！有意思的水平横向溢出滚动
发现 Linux SpaceFM 文件管理器的威力
图像处理工具Python扩展库，你了解吗？
求职数据分析，项目经验该怎么写
自定义计数器小技巧！CSS 实现长按点赞累加动画
在OKR中，我看到了数据驱动业务的未来
2023展望：新的一年将给大数据分析领域带来什么？
过五关！React高频面试题指南
阿里云ADB基于Hudi构建Lakehouse的实践
火山引擎云原生大数据在金融行业的实践
OpenHarmony富设备移植指南（二）—从postmarketOS获取移植资源
《数据成熟度指数》报告：64%的企业领袖认为大多数员工“不懂数据”
OpenHarmony 小型系统兼容性测试指南
肯睿中国（Cloudera）：2023年企业数字战略三大趋势预测
适用于 Linux 的十大命令行游戏
软件开发中的十个认知偏差

zl程序教程