Redis 分布式锁 2022-09-04 程序之旅,记录 1 条评论 1072 次阅读 ## Redis 分布式锁 ### 实现原理 共享资源互斥，实现资源串行化，在单体应用中常用的有：Synchronized、ReentrantLock。分布式锁是控制分布式系统之间同步访问共享资源的一种方式，利用 Redis 的单线程特性对共享资源进行串行化。 ### 实现方式 可以使用 Jedis 实现分布式锁的获取和释放，推荐使用 jedis 的 set 方法，其中设置 NX 保证互斥性，添加 EX 过期时间，也可是使用 setnx 命令实现，但是会存在并发问题。 释放锁可以使用 del 命令或者 redis + lua 脚本。del 命令也会存在并发问题，例如 A、B 客户端，A 在执行 jedis.del() 之前，A 锁突然过期，同时 B 客户端尝试加锁成功，然后 A 客户端执行 del 命令，解锁了 B 客户端刚加的锁。 ### 存在问题 除了上边并发的问题，还存在其他的问题。 1、单机：无法高可用。搭建集群 AP 模型。 2、主从复制：由于主从数据的不一致，有可能锁会重复获得。主库的锁没有复制到从库，从库就由于主库宕机而升级为主库，此时的主库没有锁，同一资源就能重复获得锁。该问题可以使用 RedLock (红锁)来解决，保证获取锁的时候需要在满足大于一半的缓存服务器 set 锁成功的情况下才能获取成功。 3、无法续租：超过 expireTime 后，不能继续使用。使用 Redisson 框架。 > CP 强一致性 > > AP 高可用 ### Redission 分布式锁的使用 Redisson 是架设在 Redis 基础上的一个 Java 驻内存数据网络。基于 NIO 的 Netty 框架上，生产环境使用分布式锁。 #### Maven jar 包 ```xml <dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> </dependency> ``` #### 配置 Redisson application-redis.yml ```yaml spring: redis: host: localhost port: 6379 password: 123456 #没有密码就保留空 timeout: 5000 # 连接超时时间 jedis: pool: max-active: 1000 # 池在给定时间可以分配的最大连接数。使用负值表示无限制。 max-idle: 50 #池中“空闲”连接的最大数量。使用负值表示空闲连接的数量不受限制 min-idle: 10 # 目标是池中要维护的最小空闲连接数。。 max-wait: -1 # 在池耗尽时引发异常之前，连接分配应阻止的最长时间。使用负值无限期阻塞。 redisson: pool: max-active: 1000 # 池在给定时间可以分配的最大连接数。使用负值表示无限制。 min-idle: 10 # 目标是池中要维护的最小空闲连接数。 tokenName: Authorization # 用于分布式锁的唯一标识,一般使用token如果没有找到,就找sessionId session: store-type: redis #设置session保存为默认redis的方式 ,可以解决分布式session不一致问题 ``` 使用文件 ```java import org.redisson.Redisson; import org.redisson.config.Config; public class RedissonManager { private static Config config = new Config(); // 声明 redisson 对象 private static Redisson redisson = null; // 实例化 redisson static { config.useClusterServers() // 集群扫描间隔 .setScanInterval(2000) // cluster 方式 // .addNodeAddress("redis://127.0.0.1:6379") // .addNodeAddress("redis://127.0.0.1:6379") .addNodeAddress("redis://127.0.0.1:6379"); redisson = (Redisson) Redisson.create(config); } public static Redisson getRedisson() { return redisson; } } ``` 锁的获取和释放 ```java public class DistributedRedisLock { //从配置类中获取redisson对象 private static Redisson redisson = RedissonManager.getRedisson(); private static final String LOCK_TITLE = "redisLock_"; // 加锁 public static boolean acquire(String lockName) { // 声明 key 对象 String key = LOCK_TITLE + lockName; // 获取锁对象 RLock myLock = redisson.getLock(key); //加锁，并且设置锁过期时间3秒，防止死锁的产生 uuid+threadId myLock.lock(3, TimeUnit.SECONDS); return true; } // 锁的释放 public static void reelease(String lockName) { String key = LOCK_TITLE + lockName; RLock myLock = redisson.getLock(key); myLock.unlock(); } } ``` #### 业务使用 ```java public String doSoming() { String key = "lock001"; DistributedRedisLock.acquire(key); // do something DistributedRedisLock.release(key); return "success"; } ``` ### Redisson 分布式锁的实现原理 #### 加锁的机制 Redisson 在加锁前会根据 hash 阶段选择一台机器(Redis 机器)，发送 lua 脚本到 redis 服务器上，脚本如下 ```lua //如果不存在加锁的key(serviceKey)就执行加锁的逻辑 "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + -- 无锁，加锁，1：加了一回 //pexpire命令:给指定的Key设置过期时间 "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + //判断Hash类型的加锁的key对应的的Map结构中，key对应的value是否存在 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + //将key对应的value值进行+1，重入锁 "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + //pexpire命令:给指定的Key设置过期时间 "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + //返回Key(serviceKey)的过期时间 "return redis.call('pttl', KEYS[1]);", ``` > lua 的作用，保证业务逻辑执行的原子性。也就是解决上边踢桃的 setnx 的并发问题。 名词解析 KEYS[1]：加锁的 key ARGV[1]：key 的生成时间， expire。 ARGV[2]：加锁的客户端 Id （UUID.randomUUID + ":" + threadId） #### 自动延迟机制 一旦加锁成功，就会启动一个 watch dog (看门狗），一个后端线程，会每隔 10 秒检查一下，如果还持有锁 key，会不断的延长锁的生存时间。可以通过 lockWatchdogTimeout 进行配置。 #### 释放锁的机制 ```lua -- 锁对应的hash不存在 if (redis.call('exists', KEYS[1]) == 0) then -- 通知抢锁。 redis.call('publish', KEYS[2], ARGV[1]); --结束 return 1; end; -- 如果锁不存在，不处理 if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil; end; --对其中的元素进行计数 -1 实现可重入 local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); -- 如果此时还有计数 if (counter > 0) then -- 刷新过期时间 redis.call('pexpire', KEYS[1], ARGV[2]); return 0; else -- 解锁，通知其他线程争抢锁。 redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil; ``` - KEYS[1] ：需要加锁的key，这里需要是字符串类型。 - KEYS[2] ：redis消息的ChannelName,一个分布式锁对应唯一的一个channelName: “redisson_lockchannel{” + getName() + “}” - ARGV[1] ：reids消息体，这里只需要一个字节的标记就可以，主要标记redis的key已经解锁，再结合 redis 的Subscribe，能唤醒其他订阅解锁消息的客户端线程申请锁。 - ARGV[2] ：锁的超时时间，防止死锁 - ARGV[3] ：锁的唯一标识，也就是刚才介绍的 id（UUID.randomUUID()） + “:” + threadId ### 总结 #### 分布式锁的特性 - 互斥性：只能一个客户端 - 同一性：只能同一个客户端 - 可重入：持续加锁 - 容错性：防止死锁 打赏: 微信, 支付宝 标签: 分布式, Redis 本作品采用 知识共享署名-相同方式共享 4.0 国际许可协议 进行许可。
微信
支付宝
学习了!