RedLock底层源码分析
一、Redlock红锁算法
https://redis.io/docs/manual/patterns/distributed-locks/官网说明
1、为什么要学习这个?怎么产生的?
一个很直接的问题,当我使用redis锁的那台机器挂了,出现了单点故障了,程序该何去何从?
官网上的说明
再翻译一下就是,客户端A获取到了master中的锁了,在从节点slave同步master之前,master挂了,这个时候slave就会从机上位成为master,但是它就没有客户端A获取的那个锁,此时客户端B过来了一看没有锁,直接获取一个把锁加上,这样AB加的就是同一把锁了(一锁多写),要是A完成了自己的业务把锁给删除了,B完成业务之后一看我tm锁没了。
我们加的是排他独占锁,同一时间只能有一个建redis锁成功并持有锁,严禁出现2个及以上的线程拿到锁,这是危险的操作。
2、Redlock算法设计理念
解释:这个方案解决了数据不一致的问题,直接舍弃了集群或者哨兵的模式,只使用master,官方建议使用5个,案例中使用3台机器演示,不存在主从关系,大家都是master。
2.1、容错公式
需要奇数个机器,N=2X+1(N是最终需要的机器,X是容错的机器数)
什么是容错?
失败了多少个机器实例后我还是可以容忍的,所谓的容忍就是数据一致性还是可以Ok的,CP数据一致性还是可以满足。
为啥是奇数?
因为可以用最少的机器,最多的产出效果。
举个例子:
- 使用奇数:容错机器是1个,则最终需要2*1+1=3个实例
- 使用偶数:容错机器是1个,则最终需要2*2+1=4个实例
3、落地实现Redisson
官网
官网的例子:
RLock lock = redisson.getLock("myLock");
// traditional lock method
lock.lock();
// or acquire lock and automatically unlock it after 10 seconds
lock.lock(10, TimeUnit.SECONDS);
// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
try {
...
} finally {
lock.unlock();
}
}
多重锁定
RLock lock1 = redisson1.getLock("lock1");
RLock lock2 = redisson2.getLock("lock2");
RLock lock3 = redisson3.getLock("lock3");
RLock multiLock = anyRedisson.getMultiLock(lock1, lock2, lock3);
// traditional lock method
multiLock.lock();
// or acquire lock and automatically unlock it after 10 seconds
multiLock.lock(10, TimeUnit.SECONDS);
// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
boolean res = multiLock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
try {
...
} finally {
multiLock.unlock();
}
}
二、使用Redisson进行编码改造上一节的案例
Redisson官网:https://redisson.org/
怎么使用,官网查看,quick start
先导依赖
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.24.3</version>
</dependency>
然后写RedisConfig
@Bean
public Redisson redisson()
{
Config config = new Config();
config.useSingleServer().setAddress("redis://192.168.111.27:6379").setDatabase(0).setPassword("123456");
return (Redisson) Redisson.create(config);
}
再去写service
//使用Redisson对应的官网推荐的RedLock算法实现类
@Autowired
private Redisson redisson;
public String saleByRedisson() {
String retMessage = "";
RLock redissonLock = redisson.getLock("redisLock");
redissonLock.lock();
try {
// 查询库存信息
String result = stringRedisTemplate.opsForValue().get("inventory001");
// 判断库存够不够,如果为空则设置为0,有则转化为integger
Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);
if (inventoryNumber > 0){
// 减扣库存,每次减少一个
stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));
retMessage = "成功卖出一个商品,库存剩余:"+inventoryNumber;
System.out.println(retMessage+"\t"+"服务端口号:"+port);
}else {
retMessage = "商品卖完了,去别处看看吧(0-0)ll";
}
} finally {
redissonLock.unlock();
}
return retMessage+"\t"+"服务端口号:"+port;
}
新加一个controller
@ApiOperation("扣减库存saleByRedisson,一次卖一个")
@GetMapping(value = "/inventory/saleByRedisson")
public String saleByRedisson()
{
return inventoryService.saleByRedisson();
}
好了,测试一下,单机高并发测试
一切正常,但是真的就是一切正常吗?
当然不是一帆风顺,目前会造成解锁的时候找不着锁,也就是这个线程的锁被别人删除了,所以在释放锁时要进行判断,只能删除自己的锁。
但是又区别于上一次我们的判断,上一次是因为A线程的业务没有干完,锁过期了,B线程拿到了锁,但是A的活又干完了A以为这是自己的锁就删除了,然后B来删除的时候没有锁了。
我们这次使用的是Redisson中redissonLock正在持有锁,并且正是该线程持有才能释放锁。这里的判断和解锁是原子性的,底层帮我们做了
InvrntoryService进行部分修改
finally {
//只能删除自己的,进行判断Redisson中redissonLock正在持有锁,并且正是该线程持有才能释放锁
if (redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()){
redissonLock.unlock();
}
}
再次上测试
没有问题
三、Redisdon源码解析
分布式锁的要求,加锁、可重入、续期、解锁
守护线程的“续命”
额外起一个线程,定期检查线程是否还持有锁,如果有则延长过期时间。
Redisson里面就实现了这个方案,使用“看门狗”定期检查(每1/3的锁时间检查1次),如果线程还持有锁,则刷新过期时间。
在获取锁成功后,给锁加一个watchdog, watchdog会起一个定时任务,在锁没有被释放且快要过期的时候会续期。
找到RedissonLock源码文件
这个过期时间就是30秒
所以通过redisson创建的锁默认过期时间就是30秒
再来看一下续期的源码,包括尝试加锁,加锁后的看门狗缓存续期操作
点进tryLockInnerAsync方法看到加锁的源代码
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
this.internalLockLeaseTime = unit.toMillis(leaseTime);
return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, command,
"if (redis.call('exists', KEYS[1]) == 0) then redis.call('hincrby', KEYS[1], ARGV[2], 1); redis.call('pexpire', KEYS[1], ARGV[1]); return nil; end; 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; return redis.call('pttl', KEYS[1]);", Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
}
可以看到,源代码为了保证原子性也是用的Lua脚本,分析一下这个lua脚本。
if (redis.call('exists', KEYS[1]) == 0) then 先看存不存在这个锁
redis.call('hincrby', KEYS[1], ARGV[2], 1); 不存在就hincrby设置它的值
redis.call('pexpire', KEYS[1], ARGV[1]); 并且加过期时间
return nil;
end;
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then 如果锁已经存在而且是当前线程的
redis.call('hincrby', KEYS[1], ARGV[2], 1); 就进行可重入操作,就是通过hincrby加1
redis.call('pexpire', KEYS[1], ARGV[1]); 再续个期
return nil;
end;
return redis.call('pttl', KEYS[1]);如果锁存在,而且不是当前的线程所持有,就返回这个所的ttl,这时加锁失败
watch dog自动延期机制
自动续期lua脚本分析
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then redis.call('pexpire', KEYS[1], ARGV[1]); return 1; end; return 0;", Collections.singletonList(this.getName()), this.internalLockLeaseTime, this.getLockName(threadId));
}
if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then 判断是否存在
redis.call('pexpire', KEYS[1], ARGV[1]); 存在就续期
return 1;
end;
return 0;
unlock脚本分析
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
return this.evalWriteAsync(this.getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end; 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;", Arrays.asList(this.getName(), this.getChannelName()), LockPubSub.UNLOCK_MESSAGE, this.internalLockLeaseTime, this.getLockName(threadId));
}
提取lua脚本
if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then
return nil;等于0就是没有这把锁,不是同一个线程返回nil
end;
local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); 这里自定义一个变量代表先释放一次
if (counter > 0) then 释放一次之后counter还大于0代表它是可重入锁需要刷新过期时间
redis.call('pexpire', KEYS[1], ARGV[2]);
return 0;
else 如果剩余次数小于0就删除key并发布锁释放的订阅消息,解锁成功。
redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]);
return 1;
end;
return nil;
多机案例
这里的多机,不是集群,也不是哨兵模式,而是多个主节点每一个都是master。
RedLock算法实现了多redis实例的情况,相对于单节点来说,其优点在于防止因单点故障造成整个服务停止运行的事故发生,且在多节点中锁的设计,以及多节点同时崩溃等各种意外情况都有自己的独特设计方法。
Redisson分布式锁还支持MultiLock(多重锁)机制可以将多个锁合并成一个大锁,对一个大锁进行统一的申请加锁以及释放锁。
最低保证分布式锁的有效性以及安全性的要求:
- 互斥:任何时候都只能有一个client获取锁;
- 释放死锁:即使锁定资源的服务器崩溃或者分区,仍然可以释放锁;
- 容错性:只要多数redis节点(一半以上)在使用,client就可以获取和释放锁;
网上讲的基于故障转移实现的redis主从无法真正实现Redlock:
因为redis在进行主从复制时是异步完成的,比如在clientA获取锁后,主redis复制数据到从redis过程中崩溃了,导致没有复制到从redis中,然后从redis选举出一个升级为主redis,造成新的主redis没有clientA设置的锁,这是clientB尝试获取锁,并且能够成功获取锁,导致互斥失效;
但是现在去官网找RedLock,第8.4节,会发现,这玩意被弃用了,官网推荐去使用RLock或者RFencedLock
现在,我们要使用官网中的8.3节的多重锁。
官网示例代码:
RLock lock1 = redisson1.getLock("lock1");
RLock lock2 = redisson2.getLock("lock2");
RLock lock3 = redisson3.getLock("lock3");
RLock multiLock = anyRedisson.getMultiLock(lock1, lock2, lock3);
// traditional lock method
multiLock.lock();
// or acquire lock and automatically unlock it after 10 seconds
multiLock.lock(10, TimeUnit.SECONDS);
// or wait for lock aquisition up to 100 seconds
// and automatically unlock it after 10 seconds
boolean res = multiLock.tryLock(100, 10, TimeUnit.SECONDS);
if (res) {
try {
...
} finally {
multiLock.unlock();
}
}
开始案例
我们使用docker起3个master。
然后分别启动他们
使用命令:
docker exec -it master01 /bin/bash 启动后进去再连接
或者
docker exec -it master01 redis-cli 直接连接启动
OK启动完成。
接下来我们再去idea新建一个mode,redis_redLock.
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.zm</groupId>
<artifactId>redis_distributed_lock2</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.6.10</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<lombok.version>1.16.18</lombok.version>
</properties>
<dependencies>
<!--SpringBoot通用依赖模块-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!--SpringBoot与Redis整合依赖-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
<!--swagger2-->
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.9.2</version>
</dependency>
<!--redisson-->
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.13.4</version>
</dependency>
<!--通用基础配置boottest/lombok/hutool-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.8.8</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
application.properties
server.port=9090
spring.application.name=redLock
spring.swagger2.enabled=true
spring.redis.database=0
spring.redis.password=123456
spring.redis.timeout=3000
spring.session.redis.flush-mode=single
spring.redis.pool.conn-timeout=3000
spring.redis.pool.so-timeout=3000
spring.redis.pool.size=10
spring.redis.single.address1=192.168.111.27:6381
spring.redis.single.address2=192.168.111.27:6382
spring.redis.single.address3=192.168.111.27:6383
主启动类
@SpringBootApplication
public class RedLockApplication9090 {
public static void main(String[] args) {
SpringApplication.run(RedLockApplication9090.class,args);
}
}
RedisSingleProperties单机配置类,此类中就定义那三台IP地址。
@Data
public class RedisSingleProperties {
private String address1;
private String address2;
private String address3;
}
RedisPoolProperties池化技术,定义一些超时时间和池的大小变量。
@Data
public class RedisPoolProperties {
private int maxIdle;
private int minIdle;
private int maxActive;
private int maxWait;
private int connTimeout = 10000;
private int soTimeout;
//池的大小
private int size;
}
RedisProperties读取application配置文件,顺便也将池配置和单机信息配置也注入。
@ConfigurationProperties(prefix = "spring.redis",ignoreInvalidFields = false)
@Data
public class RedisProperties {
private int database;
//等待节点回复命令的时间,该时间从命令发送成功时开始计时
private int timout = 3000;
private String password;
private String mode;
//池配置
private RedisPoolProperties pool;
//单机信息配置
private RedisSingleProperties single;
}
CacheConfiguration主配置文件,创建redissonClient实例
@Configuration
@EnableConfigurationProperties(RedisProperties.class)
public class CacheConfiguration {
@Autowired
RedisProperties redisProperties;
@Bean
RedissonClient redissonClient1(){
Config config = new Config();
String address1 = redisProperties.getSingle().getAddress1();
address1 = address1.startsWith("redis://") ? address1 : "redis://" + address1;
SingleServerConfig serverConfig = config.useSingleServer()
.setAddress(address1)
.setTimeout(redisProperties.getPool().getConnTimeout())
.setConnectionPoolSize(redisProperties.getPool().getSize())
.setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());
if (StringUtil.isNotBlank(redisProperties.getPassword())){
serverConfig.setPassword(redisProperties.getPassword());
}
return Redisson.create(config);
}
@Bean
RedissonClient redissonClient2(){
Config config = new Config();
String address2 = redisProperties.getSingle().getAddress2();
address2 = address2.startsWith("redis://") ? address2 : "redis://" + address2;
SingleServerConfig serverConfig = config.useSingleServer()
.setAddress(address2)
.setTimeout(redisProperties.getPool().getConnTimeout())
.setConnectionPoolSize(redisProperties.getPool().getSize())
.setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());
if (StringUtil.isNotBlank(redisProperties.getPassword())){
serverConfig.setPassword(redisProperties.getPassword());
}
return Redisson.create(config);
}
@Bean
RedissonClient redissonClient3(){
Config config = new Config();
String address3 = redisProperties.getSingle().getAddress3();
address3 = address3.startsWith("redis://") ? address3 : "redis://" + address3;
SingleServerConfig serverConfig = config.useSingleServer()
.setAddress(address3)
.setTimeout(redisProperties.getPool().getConnTimeout())
.setConnectionPoolSize(redisProperties.getPool().getSize())
.setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());
if (StringUtil.isNotBlank(redisProperties.getPassword())){
serverConfig.setPassword(redisProperties.getPassword());
}
return Redisson.create(config);
}
}
本次就不再写service层了,就直接在controller中写逻辑代码了。
RedLockController.java
package com.zm.redLock.controller;
import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonMultiLock;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.concurrent.TimeUnit;
@RestController
@Slf4j
public class RedLockController {
public static final String CACHE_KEY_REDROCK = "ATGUIGU_REDLOCK";
@Autowired
RedissonClient redissonClient1;
@Autowired
RedissonClient redissonClient2;
@Autowired
RedissonClient redissonClient3;
@GetMapping("/getMultiLock")
public String getMultiLock(){
long threadID = Thread.currentThread().getId();
RLock lock1 = redissonClient1.getLock(CACHE_KEY_REDROCK);
RLock lock2 = redissonClient2.getLock(CACHE_KEY_REDROCK);
RLock lock3 = redissonClient3.getLock(CACHE_KEY_REDROCK);
RedissonMultiLock redLock = new RedissonMultiLock(lock1, lock2, lock3);
redLock.lock();
try {
System.out.println("进入业务逻辑:多重锁"+threadID);
//故意停30秒
try { TimeUnit.SECONDS.sleep(30); } catch (InterruptedException e) { e.printStackTrace(); }
System.out.println("多重锁业务逻辑结束...");
} catch (Exception e) {
e.printStackTrace();
log.error("multilock exception:{}",e.getCause()+"\t"+e.getMessage());
} finally {
redLock.unlock();
System.out.println("释放锁成功!!!");
}
return "多重锁已经完成:"+threadID;
}
}
启动进行测试,刚开始它会一直转圈,因为我们暂停进程30秒,让它自动续期。
看一下自动续期的效果
可以看到确实续期了30秒,这个是第二台master,可以看出三个master是同步的。
30秒续期之后的时间已结束。
查看后台
现在,我们将其中一台机器手动挂机,然后再给它打开,它默认的时间是30秒,但是马上就跟上了大部队,与其他的master进行时间同步。容错性贼强!