经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 数据库/运维 » Redis » 查看文章
Redis加Lua脚本实现分布式锁
来源：cnblogs　　作者：coffee_baby　　时间：2024/2/28 9:07:48　　对本文有异议

先讲一下为什么使用分布式锁：

在传统的单体应用中，我们可以使用Java并发处理相关的API（如ReentrantLock或synchronized）来实现对共享资源的互斥控制，确保在高并发情况下同一时间只有一个线程能够执行特定方法。然而，随着业务的发展，单体应用逐渐演化为分布式系统，多线程、多进程分布在不同机器上，这导致了原有的单机部署下的并发控制策略失效。为了解决这一问题，我们需要引入一种跨JVM的互斥机制来管理共享资源的访问，这就是分布式锁所要解决的核心问题。

Lua介绍

Lua 是一种轻量小巧的脚本语言，用标准C语言编写并以源代码形式开放， 其设计目的是为了嵌入应用程序中，从而为应用程序提供灵活的扩展和定制功能。

为什么要用Lua呢

Redis采用单线程架构，可以保证单个命令的原子性，但是无法保证一组命令在高并发场景下的原子性。

在以下场景中：

  1. 当 事务1执行删除操作时，查询到的锁值确实相等。
  2. 在 事务1执行删除操作之前，锁的过期时间刚好到达，导致 Redis 自动释放了该锁。
  3. 事务2获取了这个已被释放的锁。
  4. 当 事务1执行删除操作时，会意外地删除掉 事务2持有的锁。

上面的删除情况也无法保证原子性，只能通过lua脚本实现

如果redis客户端通过lua脚本把3个命令一次性发送给redis服务器，那么这三个指令就不会被其他客户端指令打断。Redis 也保证脚本会以原子性(atomic)的方式执行： 当某个脚本正在运行的时候，不会有其他脚本或 Redis 命令被执行。

Lua脚本命令

在Redis中需要通过eval命令执行lua脚本

  1. EVAL script numkeys key [key ...] arg [arg ...]
  3. script：lua脚本字符串，这段Lua脚本不需要（也不应该）定义函数。
  4. numkeys：lua脚本中KEYS数组的大小
  5. key [key ...]：KEYS数组中的元素
  6. arg [arg ...]：ARGV数组中的元素

案列1：动态传参

  1. EVAL "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 5 8 10 30 40 50 60 70 
  2. # 输出：8 10 60 70
  4. EVAL "if KEYS[1] > ARGV[1] then return 1 else return 0 end" 1 10 20
  5. # 输出：0
  7. EVAL "if KEYS[1] > ARGV[1] then return 1 else return 0 end" 1 20 10
  8. # 输出：1

案列2：执行redis类库方法

  1. EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 bbb 20

可重入性

可重入性是指一个线程在持有锁的情况下，可以多次获取同一个锁而不会发生死锁或阻塞的特性。在可重入锁中，线程可以重复获取已经持有的锁，每次获取都会增加一个计数器，直到计数器归零时才会真正释放锁。
下面是一个示例代码来说明可重入性：

  1. public synchronized void a() {
  2.  b();
  3. }
  4. public synchronized void b() {
  5.  // pass
  6. }

假设线程X在方法a中获取了锁后，继续执行方法b。如果这是一个不可重入的锁，线程X在执行b方法时将会被阻塞，因为它已经持有了该锁并且无法再次获取。这种情况下，线程X必须等待自己释放锁后才能再次争抢该锁。
而对于可重入性的情况，当线程X持有了该锁后，在遇到加锁方法时会直接将加锁次数加1，并继续执行方法逻辑。当退出加锁方法时，加锁次数再减1。只有当加锁次数归零时，该线程才会真正释放该锁。
因此，可重入性的最大特点就是计数器的存在，用于统计加锁的次数。在分布式环境中实现可重入分布式锁时也需要考虑如何正确统计和管理加锁次数。

加锁脚本

Redis 提供了 Hash （哈希表）这种可以存储键值对数据结构。所以我们可以使用 Redis Hash 存储的锁的重入次数，然后利用 lua 脚本判断逻辑。

  1. if (redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1) 
  2. then
  3.     redis.call('hincrby', KEYS[1], ARGV[1], 1);
  4.     redis.call('expire', KEYS[1], ARGV[2]);
  5.     return 1;
  6. else
  7. 	return 0;
  8. end

假设值为：KEYS:[lock], ARGV[uuid, expire]

如果锁不存在或者这是自己的锁，就通过hincrby（不存在就新增并加1，存在就加1）获取锁或者锁次数加1。

解锁脚本

  1. -- 判断 hash set 可重入 key 的值是否等于 0
  2. -- 如果为 nil 代表 自己的锁已不存在，在尝试解其他线程的锁，解锁失败
  3. -- 如果为 0 代表 可重入次数被减 1
  4. -- 如果为 1 代表 该可重入 key 解锁成功
  5. if(redis.call('hexists', KEYS[1], ARGV[1]) == 0) then 
  6.     return nil; 
  7. elseif(redis.call('hincrby', KEYS[1], ARGV[1], -1) > 0) then 
  8.     return 0; 
  9. else 
  10.     redis.call('del', KEYS[1]); 
  11.     return 1; 
  12. end;

如果锁不存在直接返回null，如果锁存在就对数量进行减一，如果减到等于0 就直接删除此锁

自动续期

有可能代码没执行完毕，锁就到期了。基于上面这种情况需要对锁进行续期。使用定时器加lua脚本进行对锁续期

  1. if(redis.call('hexists', KEYS[1], ARGV[1]) == 1) then 
  2.     redis.call('expire', KEYS[1], ARGV[2]); 
  3.     return 1; 
  4. else 
  5.     return 0; 
  6. end

Java代码实现


考虑到分布式锁可能使用多种方式实现，比如Redis、mysql、zookeeper,所以暂时做成一个工厂类，按需使用。

以下是完整代码：

  1. public class DistributedRedisLock implements Lock {
  3.     private StringRedisTemplate redisTemplate;
  5.     private String lockName;
  7.     private String uuid;
  9.     private long expire = 30;
  11.     public DistributedRedisLock(StringRedisTemplate redisTemplate, String lockName, String uuid) {
  12.         this.redisTemplate = redisTemplate;
  13.         this.lockName = lockName;
  14.         this.uuid = uuid + ":" + Thread.currentThread().getId();
  15.     }
  17.     @Override
  18.     public void lock() {
  19.         this.tryLock();
  20.     }
  22.     @Override
  23.     public void lockInterruptibly() throws InterruptedException {
  25.     }
  27.     @Override
  28.     public boolean tryLock() {
  29.         try {
  30.             return this.tryLock(-1L, TimeUnit.SECONDS);
  31.         } catch (InterruptedException e) {
  32.             e.printStackTrace();
  33.         }
  34.         return false;
  35.     }
  37.     /**
  38.      * 加锁方法
  39.      * @param time
  40.      * @param unit
  41.      * @return
  42.      * @throws InterruptedException
  43.      */
  44.     @Override
  45.     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  46.         if (time != -1){
  47.             this.expire = unit.toSeconds(time);
  48.         }
  49.         String script = "if redis.call('exists', KEYS[1]) == 0 or redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
  50.                 "then " +
  51.                 "   redis.call('hincrby', KEYS[1], ARGV[1], 1) " +
  52.                 "   redis.call('expire', KEYS[1], ARGV[2]) " +
  53.                 "   return 1 " +
  54.                 "else " +
  55.                 "   return 0 " +
  56.                 "end";
  57.         while (!this.redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuid, String.valueOf(expire))){
  58.             Thread.sleep(50);
  59.         }
  60.         // 加锁成功，返回之前，开启定时器自动续期
  61.         this.renewExpire();
  62.         return true;
  63.     }
  65.     /**
  66.      * 解锁方法
  67.      */
  68.     @Override
  69.     public void unlock() {
  70.         String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 0 " +
  71.                 "then " +
  72.                 "   return nil " +
  73.                 "elseif redis.call('hincrby', KEYS[1], ARGV[1], -1) == 0 " +
  74.                 "then " +
  75.                 "   return redis.call('del', KEYS[1]) " +
  76.                 "else " +
  77.                 "   return 0 " +
  78.                 "end";
  79.         Long flag = this.redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName), uuid);
  80.         if (flag == null){
  81.             throw new IllegalMonitorStateException("this lock doesn't belong to you!");
  82.         }
  83.     }
  85.     @Override
  86.     public Condition newCondition() {
  87.         return null;
  88.     }
  90.     private void renewExpire(){
  91.         String script = "if redis.call('hexists', KEYS[1], ARGV[1]) == 1 " +
  92.                 "then " +
  93.                 "   return redis.call('expire', KEYS[1], ARGV[2]) " +
  94.                 "else " +
  95.                 "   return 0 " +
  96.                 "end";
  97.         new Timer().schedule(new TimerTask() {
  98.             @Override
  99.             public void run() {
  100.                 if (redisTemplate.execute(new DefaultRedisScript<>(script, Boolean.class), Arrays.asList(lockName), uuid, String.valueOf(expire))) {
  101.                     renewExpire();
  102.                 }
  103.             }
  104.         }, this.expire * 1000 / 3);
  105.     }
  106. }

DistributedLockClient

  1. @Component
  2. public class DistributedLockClient {
  3.     @Autowired
  4.     private StringRedisTemplate redisTemplate;
  6.     private String uuid;
  8.     public DistributedLockClient() {
  9.         this.uuid = UUID.randomUUID().toString();
  10.     }
  12.     public DistributedRedisLock getRedisLock(String lockName){
  13.         return new DistributedRedisLock(redisTemplate, lockName, uuid);
  14.     }
  15. }

使用及测试：

在业务代码中使用：

  1. public void deduct() {
  2.     DistributedRedisLock redisLock = this.distributedLockClient.getRedisLock("lock");
  3.     redisLock.lock();
  5.     try {
  6.         // 1. 查询库存信息
  7.         String stock = redisTemplate.opsForValue().get("stock").toString();
  9.         // 2. 判断库存是否充足
  10.         if (stock != null && stock.length() != 0) {
  11.             Integer st = Integer.valueOf(stock);
  12.             if (st > 0) {
  13.                 // 3.扣减库存
  14.                 redisTemplate.opsForValue().set("stock", String.valueOf(--st));
  15.             }
  16.         }
  17.     } finally {
  18.         redisLock.unlock();
  19.     }
  20. }

测试可重入性：

红锁算法

在Redis集群状态下可能出现的问题如下：

  1. 客户端A从主节点（master）获取到了锁。
  2. 在主节点将锁同步到从节点（slave）之前，主节点发生宕机。
  3. 从节点被晋升为主节点。
  4. 客户端B获取了同一个资源，但是客户端A已经在另一个锁上获取了锁。
    在这种情况下，由于主节点宕机导致从节点晋升为新的主节点，可能会出现客户端B误认为资源未被锁定而获取了另一个锁的情况。这可能导致数据不一致性或竞争条件的发生。
    为了避免这种问题

安全失效！

解决集群下锁失效，参照redis官方网站针对redlock文档：https://redis.io/topics/distlock

实现步骤：

  1. 客户端向N个Redis节点发送请求获取锁。
  2. 每个Redis节点生成一个独立的随机值作为锁值，并设置相同的过期时间。
  3. 客户端等待大部分节点（如大多数节点的一半以上）返回获取成功的响应。
  4. 如果大部分节点返回获取成功，则认定为成功获取了分布式锁；否则认定为未获取到分布式锁。

原文链接：https://www.cnblogs.com/dupengpeng/p/18037650

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Redis热门文章

  • Redis-复制
  • Redis
  • 关于SpringBoot整合redis使用Lettuce客户端超时问题
  • RedisTemplate常用方法封装
  • RedisTemplate中opsForValue和opsForList方法的使用详解
  • Redis7.0部署集群的实现步骤
  • Java&nbsp;Redis&nbsp;Redisson配置教程详解
  • Go语言开发redis封装及简单使用详解
  • SpringBoot Redis批量存取数据的操作
  • .NetCore下基于FreeRedis实现的Redis6.0客户端缓存之缓存键条件优雅过滤

    • Redis推荐文章

  • Redis系列：使用Stream实现消息队列 （图文总结+Go案例） - Hello-Brand
  • 记一次 redis 事件注册不当导致的内存泄露
  • Java Redis多限流
  • 【Azure Developer】一个复制Redis Key到另一个Redis服务的工具(redis_copy_net8)
  • Redis 高阶应用
  • 【长文】带你搞明白Redis
  • StackExchange.Redis跑起来，为什么这么溜？
  • 【进阶篇】基于 Redis 实现分布式锁的全过程
  • Higress 基于自定义插件访问 Redis
  • 美团二面：如何保证Redis与Mysql双写一致性？连续两个面试问到了！

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号