服务器之家：专注于服务器技术及软件下载分享
分类导航

Mysql|Sql Server|Oracle|Redis|MongoDB|PostgreSQL|Sqlite|DB2|mariadb|Access|数据库技术|

服务器之家 - 数据库 - Redis - Redis分布式锁升级版RedLock及SpringBoot实现方法

Redis分布式锁升级版RedLock及SpringBoot实现方法

2021-07-25 22:06等不到的口琴 Redis

这篇文章主要介绍了Redis分布式锁升级版RedLock及SpringBoot实现,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下

分布式锁概览

在多线程的环境下，为了保证一个代码块在同一时间只能由一个线程访问，java中我们一般可以使用synchronized语法和reetrantlock去保证，这实际上是本地锁的方式。但是现在公司都是流行分布式架构，在分布式环境下，如何保证不同节点的线程同步执行呢？因此就引出了分布式锁，它是控制分布式系统之间互斥访问共享资源的一种方式。

在一个分布式系统中，多台机器上部署了多个服务，当客户端一个用户发起一个数据插入请求时，如果没有分布式锁机制保证，那么那多台机器上的多个服务可能进行并发插入操作，导致数据重复插入，对于某些不允许有多余数据的业务来说，这就会造成问题。而分布式锁机制就是为了解决类似这类问题，保证多个服务之间互斥的访问共享资源，如果一个服务抢占了分布式锁，其他服务没获取到锁，就不进行后续操作。大致意思如下图所示：

Redis分布式锁升级版RedLock及SpringBoot实现方法

分布式锁的特点

分布式锁一般有如下的特点：

  • 互斥性： 同一时刻只能有一个线程持有锁
  • 可重入性： 同一节点上的同一个线程如果获取了锁之后能够再次获取锁
  • 锁超时：和j.u.c中的锁一样支持锁超时，防止死锁
  • 高性能和高可用： 加锁和解锁需要高效，同时也需要保证高可用，防止分布式锁失效
  • 具备阻塞和非阻塞性：能够及时从阻塞状态中被唤醒

分布式锁的实现方式

我们一般实现分布式锁有以下几种方式：

  • 基于数据库
  • 基于redis
  • 基于zookeeper

redis普通分布式锁存在的问题

说到redis分布式锁,大部分人都会想到：setnx+lua（redis保证执行lua脚本时不执行其他操作，保证操作的原子性），或者知道set key value px milliseconds nx。后一种方式的核心实现命令如下：

?
1
2
3
4
5
6
7
8
9
- 获取锁（unique_value可以是uuid等）
set resource_name unique_value nx px 30000
 
- 释放锁（lua脚本中，一定要比较value，防止误解锁）
if redis.call("get",keys[1]) == argv[1] then
 return redis.call("del",keys[1])
else
 return 0
end

这种实现方式有3大要点（也是面试概率非常高的地方）：

  • set命令要用set key value px milliseconds nx；
  • value要具有唯一性；
  • 释放锁时要验证value值，不能误解锁；

事实上这类锁最大的缺点就是它加锁时只作用在一个redis节点上，即使redis通过sentinel保证高可用，如果这个master节点由于某些原因发生了主从切换，那么就会出现锁丢失的情况：

  1. 在redis的master节点上拿到了锁；
  2. 但是这个加锁的key还没有同步到slave节点；
  3. master故障，发生故障转移，slave节点升级为master节点；
  4. 导致锁丢失。

为了避免单点故障问题，redis作者antirez基于分布式环境下提出了一种更高级的分布式锁的实现方式：redlock。redlock也是redis所有分布式锁实现方式中唯一能让面试官高潮的方式。

redis高级分布式锁:redlock

antirez提出的redlock算法大概是这样的：

在redis的分布式环境中，我们假设有n个redis master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。我们确保将在n个实例上使用与在redis单实例下相同方法获取和释放锁。现在我们假设有5个redis master节点，同时我们需要在5台服务器上面运行这些redis实例，这样保证他们不会同时都宕掉。

为了取到锁，客户端应该执行以下操作:

  • 获取当前unix时间，以毫秒为单位。
  • 依次尝试从5个实例，使用相同的key和具有唯一性的value（例如uuid）获取锁。当向redis请求获取锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间ttl为10秒，则超时时间应该在5-50毫秒之间。这样可以避免服务器端redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试去另外一个redis实例请求获取锁。
  • 客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（n/2+1，这里是3个节点）的redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
  • 如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）。
  • 如果因为某些原因，获取锁失败（没有在至少n/2+1个redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的redis实例上进行解锁（即便某些redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。
  • 此处不讨论时钟漂移

Redis分布式锁升级版RedLock及SpringBoot实现方法

redlock源码

redisson已经有对redlock算法封装，接下来对其用法进行简单介绍，并对核心源码进行分析（假设5个redis实例）。

1. redlock依赖

?
1
2
3
4
5
6
<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->
<dependency>
 <groupid>org.redisson</groupid>
 <artifactid>redisson</artifactid>
 <version>3.3.2</version>
</dependency>

2. redlock用法

首先，我们来看一下redission封装的redlock算法实现的分布式锁用法，非常简单，跟重入锁（reentrantlock）有点类似：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
config config = new config();
config.usesentinelservers().addsentineladdress("127.0.0.1:6369","127.0.0.1:6379", "127.0.0.1:6389")
 .setmastername("mastername")
 .setpassword("password").setdatabase(0);
redissonclient redissonclient = redisson.create(config);
// 还可以getfairlock(), getreadwritelock()
rlock redlock = redissonclient.getlock("redlock_key");
boolean islock;
try {
 islock = redlock.trylock();
 // 500ms拿不到锁, 就认为获取锁失败。10000ms即10s是锁失效时间。
 islock = redlock.trylock(500, 10000, timeunit.milliseconds);
 if (islock) {
 //todo if get lock success, do something;
 }
} catch (exception e) {
} finally {
 // 无论如何, 最后都要解锁
 redlock.unlock();
}

3. redlock唯一id

实现分布式锁的一个非常重要的点就是set的value要具有唯一性，redisson的value是怎样保证value的唯一性呢？答案是uuid+threadid。入口在redissonclient.getlock("redlock_key")，源码在redisson.java和redissonlock.java中：

?
1
2
3
4
protected final uuid id = uuid.randomuuid();
string getlockname(long threadid) {
 return id + ":" + threadid;
}

4. redlock获取锁

获取锁的代码为redlock.trylock()或者redlock.trylock(500, 10000, timeunit.milliseconds)，两者的最终核心源码都是下面这段代码，只不过前者获取锁的默认租约时间（leasetime）是lock_expiration_interval_seconds，即30s：

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
<t> rfuture<t> trylockinnerasync(long leasetime, timeunit unit, long threadid, redisstrictcommand<t> command) {
 internallockleasetime = unit.tomillis(leasetime);
 // 获取锁时向5个redis实例发送的命令
 return commandexecutor.evalwriteasync(getname(), longcodec.instance, command,
  // 首先分布式锁的key不能存在，如果确实不存在，那么执行hset命令（hset redlock_key uuid+threadid 1），并通过pexpire设置失效时间（也是锁的租约时间）
  "if (redis.call('exists', keys[1]) == 0) then " +
   "redis.call('hset', keys[1], argv[2], 1); " +
   "redis.call('pexpire', keys[1], argv[1]); " +
   "return nil; " +
  "end; " +
  // 如果分布式锁的key已经存在，并且value也匹配，表示是当前线程持有的锁，那么重入次数加1，并且设置失效时间
  "if (redis.call('hexists', keys[1], argv[2]) == 1) then " +
   "redis.call('hincrby', keys[1], argv[2], 1); " +
   "redis.call('pexpire', keys[1], argv[1]); " +
   "return nil; " +
  "end; " +
  // 获取分布式锁的key的失效时间毫秒数
  "return redis.call('pttl', keys[1]);",
  // 这三个参数分别对应keys[1]，argv[1]和argv[2]
  collections.<object>singletonlist(getname()), internallockleasetime, getlockname(threadid));
}

获取锁的命令中，

  • keys[1]就是collections.singletonlist(getname())，表示分布式锁的key，即redlock_key；
  • argv[1]就是internallockleasetime，即锁的租约时间，默认30s；
  • argv[2]就是getlockname(threadid)，是获取锁时set的唯一值，即uuid+threadid：

5. redlock释放锁

释放锁的代码为redlock.unlock()，核心源码如下：

?
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
protected rfuture<boolean> unlockinnerasync(long threadid) {
 // 向5个redis实例都执行如下命令
 return commandexecutor.evalwriteasync(getname(), longcodec.instance, rediscommands.eval_boolean,
  // 如果分布式锁key不存在，那么向channel发布一条消息
  "if (redis.call('exists', keys[1]) == 0) then " +
  "redis.call('publish', keys[2], argv[1]); " +
  "return 1; " +
  "end;" +
  // 如果分布式锁存在，但是value不匹配，表示锁已经被占用，那么直接返回
  "if (redis.call('hexists', keys[1], argv[3]) == 0) then " +
  "return nil;" +
  "end; " +
  // 如果就是当前线程占有分布式锁，那么将重入次数减1
  "local counter = redis.call('hincrby', keys[1], argv[3], -1); " +
  // 重入次数减1后的值如果大于0，表示分布式锁有重入过，那么只设置失效时间，还不能删除
  "if (counter > 0) then " +
  "redis.call('pexpire', keys[1], argv[2]); " +
  "return 0; " +
  "else " +
  // 重入次数减1后的值如果为0，表示分布式锁只获取过1次，那么删除这个key，并发布解锁消息
  "redis.call('del', keys[1]); " +
  "redis.call('publish', keys[2], argv[1]); " +
  "return 1; "+
  "end; " +
  "return nil;",
  // 这5个参数分别对应keys[1]，keys[2]，argv[1]，argv[2]和argv[3]
  arrays.<object>aslist(getname(), getchannelname()), lockpubsub.unlockmessage, internallockleasetime, getlockname(threadid));
 
}

redis实现的分布式锁轮子

下面利用springboot + jedis + aop的组合来实现一个简易的分布式锁。

1. 自定义注解

自定义一个注解，被注解的方法会执行获取分布式锁的逻辑

?
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
@target(elementtype.method)
@retention(retentionpolicy.runtime)
@documented
@inherited
public @interface redislock {
 /**
 * 业务键
 *
 * @return
 */
 string key();
 /**
 * 锁的过期秒数,默认是5秒
 *
 * @return
 */
 int expire() default 5;
 
 /**
 * 尝试加锁，最多等待时间
 *
 * @return
 */
 long waittime() default long.min_value;
 /**
 * 锁的超时时间单位
 *
 * @return
 */
 timeunit timeunit() default timeunit.seconds;
}

2. aop拦截器实现

在aop中我们去执行获取分布式锁和释放分布式锁的逻辑，代码如下：

?
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
@aspect
@component
public class lockmethodaspect {
 @autowired
 private redislockhelper redislockhelper;
 @autowired
 private jedisutil jedisutil;
 private logger logger = loggerfactory.getlogger(lockmethodaspect.class);
 
 @around("@annotation(com.redis.lock.annotation.redislock)")
 public object around(proceedingjoinpoint joinpoint) {
 jedis jedis = jedisutil.getjedis();
 methodsignature signature = (methodsignature) joinpoint.getsignature();
 method method = signature.getmethod();
 
 redislock redislock = method.getannotation(redislock.class);
 string value = uuid.randomuuid().tostring();
 string key = redislock.key();
 try {
  final boolean islock = redislockhelper.lock(jedis,key, value, redislock.expire(), redislock.timeunit());
  logger.info("islock : {}",islock);
  if (!islock) {
  logger.error("获取锁失败");
  throw new runtimeexception("获取锁失败");
  }
  try {
  return joinpoint.proceed();
  } catch (throwable throwable) {
  throw new runtimeexception("系统异常");
  }
 } finally {
  logger.info("释放锁");
  redislockhelper.unlock(jedis,key, value);
  jedis.close();
 }
 }
}

3. redis实现分布式锁核心类

?
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
@component
public class redislockhelper {
 private long sleeptime = 100;
 /**
 * 直接使用setnx + expire方式获取分布式锁
 * 非原子性
 *
 * @param key
 * @param value
 * @param timeout
 * @return
 */
 public boolean lock_setnx(jedis jedis,string key, string value, int timeout) {
 long result = jedis.setnx(key, value);
 // result = 1时，设置成功，否则设置失败
 if (result == 1l) {
  return jedis.expire(key, timeout) == 1l;
 } else {
  return false;
 }
 }
 
 /**
 * 使用lua脚本，脚本中使用setnex+expire命令进行加锁操作
 *
 * @param jedis
 * @param key
 * @param uniqueid
 * @param seconds
 * @return
 */
 public boolean lock_with_lua(jedis jedis,string key, string uniqueid, int seconds) {
 string lua_scripts = "if redis.call('setnx',keys[1],argv[1]) == 1 then" +
  "redis.call('expire',keys[1],argv[2]) return 1 else return 0 end";
 list<string> keys = new arraylist<>();
 list<string> values = new arraylist<>();
 keys.add(key);
 values.add(uniqueid);
 values.add(string.valueof(seconds));
 object result = jedis.eval(lua_scripts, keys, values);
 //判断是否成功
 return result.equals(1l);
 }
 
 /**
 * 在redis的2.6.12及以后中,使用 set key value [nx] [ex] 命令
 *
 * @param key
 * @param value
 * @param timeout
 * @return
 */
 public boolean lock(jedis jedis,string key, string value, int timeout, timeunit timeunit) {
 long seconds = timeunit.toseconds(timeout);
 return "ok".equals(jedis.set(key, value, "nx", "ex", seconds));
 }
 
 /**
 * 自定义获取锁的超时时间
 *
 * @param jedis
 * @param key
 * @param value
 * @param timeout
 * @param waittime
 * @param timeunit
 * @return
 * @throws interruptedexception
 */
 public boolean lock_with_waittime(jedis jedis,string key, string value, int timeout, long waittime,timeunit timeunit) throws interruptedexception {
 long seconds = timeunit.toseconds(timeout);
 while (waittime >= 0) {
  string result = jedis.set(key, value, "nx", "ex", seconds);
  if ("ok".equals(result)) {
  return true;
  }
  waittime -= sleeptime;
  thread.sleep(sleeptime);
 }
 return false;
 }
 /**
 * 错误的解锁方法—直接删除key
 *
 * @param key
 */
 public void unlock_with_del(jedis jedis,string key) {
 jedis.del(key);
 }
 
 /**
 * 使用lua脚本进行解锁操纵，解锁的时候验证value值
 *
 * @param jedis
 * @param key
 * @param value
 * @return
 */
 public boolean unlock(jedis jedis,string key,string value) {
 string luascript = "if redis.call('get',keys[1]) == argv[1] then " +
  "return redis.call('del',keys[1]) else return 0 end";
 return jedis.eval(luascript, collections.singletonlist(key), collections.singletonlist(value)).equals(1l);
 }
}

4. controller层控制

定义一个testcontroller来测试我们实现的分布式锁

?
1
2
3
4
5
6
7
8
@restcontroller
public class testcontroller {
 @redislock(key = "redis_lock")
 @getmapping("/index")
 public string index() {
 return "index";
 }
}

站在巨人的肩膀上

1.redlock：redis分布式锁最牛逼的实现

2.基于redis的分布式锁实现

到此这篇关于redis分布式锁升级版redlock及springboot实现的文章就介绍到这了,更多相关redis分布式锁redlock内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家！

原文链接：https://www.cnblogs.com/Courage129/archive/2021/02/01/14355562.html

  • springboot
  • redis
  • 分布式锁
  • RedLock

    • 延伸 · 阅读

    精彩推荐
    • Redis关于Redis数据库入门详细介绍

      关于Redis数据库入门详细介绍

      大家好，本篇文章主要讲的是关于Redis数据库入门详细介绍，感兴趣的同学赶快来看一看吧，对你有帮助的话记得收藏一下，方便下次浏览...

      沃尔码6982022-01-24
    • Redisredis缓存存储Session原理机制

      redis缓存存储Session原理机制

      这篇文章主要为大家介绍了redis缓存存储Session原理机制详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪...

      程序媛张小妍9252021-11-25
    • Redis如何使用Redis锁处理并发问题详解

      如何使用Redis锁处理并发问题详解

      这篇文章主要给大家介绍了关于如何使用Redis锁处理并发问题的相关资料，文中通过示例代码介绍的非常详细，对大家学习或者使用Redis具有一定的参考学习...

      haofly4522019-11-26
    • Redis《面试八股文》之 Redis十六卷

      《面试八股文》之 Redis十六卷

      redis 作为我们最常用的内存数据库，很多地方你都能够发现它的身影，比如说登录信息的存储，分布式锁的使用，其经常被我们当做缓存去使用。...

      moon聊技术8182021-07-26
    • RedisRedis Template实现分布式锁的实例代码

      Redis Template实现分布式锁的实例代码

      这篇文章主要介绍了Redis Template实现分布式锁,需要的朋友可以参考下 ...

      晴天小哥哥2592019-11-18
    • RedisRedis集群的5种使用方式，各自优缺点分析

      Redis集群的5种使用方式，各自优缺点分析

      Redis 多副本，采用主从(replication)部署结构，相较于单副本而言最大的特点就是主从实例间数据实时同步，并且提供数据持久化和备份策略。...

      优知学院4082021-08-10
    • RedisRedis 6.X Cluster 集群搭建

      Redis 6.X Cluster 集群搭建

      码哥带大家完成在 CentOS 7 中安装 Redis 6.x 教程。在学习 Redis Cluster 集群之前，我们需要先搭建一套集群环境。机器有限，实现目标是一台机器上搭建 6 个节...

      码哥字节15752021-04-07
    • Redis详解三分钟快速搭建分布式高可用的Redis集群

      详解三分钟快速搭建分布式高可用的Redis集群

      这篇文章主要介绍了详解三分钟快速搭建分布式高可用的Redis集群，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，...

      万猫学社4502021-07-25
    最近更新
    编辑推荐
    阅读排行
    热门标签
    16384 　 Hashes 　 Sorted-Sets 　 服务器管理命令 　 管线 　 操作 　 主键失效 　 LRU 　 机制 　 应用场景 　 redis-py 　 禁用命令 　 危险命令 　 速率限制 　 Redis简介 　 用户积分 　 Mac OS 　 集群部署 　 Redis安装 　 Redis配置 　 远程方法调用 　 数据结构 　 Redis教程 　 redis-dump 　 还原数据 　 java脚本 　 计数 　 使用场景 　 计数器模式 　 数据大小 　
    © 2019-2023 服务器之家(www.zzvips.com) 版权所有关于我们联系我们版权申明网站地图