Redlock算法和底层源码分析

自写分布式锁

lock()加锁关键逻辑
加锁的Lua脚本，通过redis里面的hash数据模型，加锁和可重入性都要保证
加锁不成，需要while进行重试并自旋
自动续期，加个钟

8.0代码===>加锁实际上就是在redis中，给Key键设置一个值，为避免死锁，并给定一个过期时间，自旋，续期

unlock解锁关键逻辑
考虑可重入性的递减，加锁几次就要减锁几次
最后到零了，直接del删除
将Key键删除。但也不能乱删，不能说客户端1的请求将客户端2的锁给删除掉，只能自己删除自己的锁

 @Overridepublic void unlock(){String script =             //Lua解锁脚本"if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 0 then " +"   return nil " +"elseif redis.call('HINCRBY',KEYS[1],ARGV[1],-1) == 0 then " +"   return redis.call('del',KEYS[1]) " +"else " +"   return 0 " +"end";// nil = false 1 = true 0 = falseSystem.out.println("lockName: "+lockName);System.out.println("uuidValue: "+uuidValue);System.out.println("expireTime: "+expireTime);Long flag = stringRedisTemplate.execute(new DefaultRedisScript<>(script, Long.class), Arrays.asList(lockName),uuidValue,String.valueOf(expireTime));if(flag == null){throw new RuntimeException("This lock doesn't EXIST");}}

红锁算法

官网：https://redis.io/docs/manual/patterns/distributed-locks/

线程 1 首先获取锁成功，将键值对写入 redis 的 master 节点，在 redis 将该键值对同步到 slave 节点之前，master 发生了故障；redis 触发故障转移，其中一个 slave 升级为新的 master，此时新上位的master并不包含线程1写入的键值对，因此线程 2 尝试获取锁也可以成功拿到锁， 此时相当于有两个线程获取到了锁，可能会导致各种预期之外的情况发生，例如最常见的脏数据。

我们加的是排它独占锁，同一时间只能有一个建redis锁成功并持有锁，严禁出现2个以上的请求线程拿到锁。危险的

为此，Redis也提供了Redlock算法，用来实现基于多个实例的分布式锁。

锁变量由多个实例维护，即使有实例发生了故障，锁变量仍然是存在的，客户端还是可以完成锁操作。

设计理念

该方案也是基于（set 加锁、Lua 脚本解锁）进行改良的，所以redis之父antirez 只描述了差异的地方，大致方案如下

假设我们有N个Redis主节点，例如 N = 5这些节点是完全独立的，我们不使用复制或任何其他隐式协调系统，

为了取到锁客户端执行以下操作：

1	获取当前时间，以毫秒为单位；
2	依次尝试从5个实例，使用相同的 key 和随机值（例如 UUID）获取锁。当向Redis 请求获取锁时，客户端应该设置一个超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为 10 秒，则超时时间应该在 5-50 毫秒之间。这样可以防止客户端在试图与一个宕机的 Redis 节点对话时长时间处于阻塞状态。如果一个实例不可用，客户端应该尽快尝试去另外一个 Redis 实例请求获取锁；
3	客户端通过当前时间减去步骤 1 记录的时间来计算获取锁使用的时间。当且仅当从大多数（N/2+1，这里是 3 个节点）的 Redis 节点都取到锁，并且获取锁使用的时间小于锁失效时间时，锁才算获取成功；
4	如果取到了锁，其真正有效时间等于初始有效时间减去获取锁所使用的时间（步骤 3 计算的结果）。
5	如果由于某些原因未能获得锁（无法在至少 N/2 + 1 个 Redis 实例获取锁、或获取锁的时间超过了有效时间），客户端应该在所有的 Redis 实例上进行解锁（即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。

总结

这个锁的算法实现了多redis实例的情况，相对于单redis节点来说，优点在于 防止了 单节点故障造成整个服务停止运行的情况且在多节点中锁的设计，及多节点同时崩溃等各种意外情况有自己独特的设计方法。

Redisson 分布式锁支持 MultiLock 机制可以将多个锁合并为一个大锁，对一个大锁进行统一的申请加锁以及释放锁。

最低保证分布式锁的有效性及安全性的要求如下：

1.互斥；任何时刻只能有一个client获取锁

2.释放死锁；即使锁定资源的服务崩溃或者分区，仍然能释放锁

3.容错性；只要多数redis节点（一半以上）在使用，client就可以获取和释放锁

网上讲的基于故障转移实现的redis主从无法真正实现Redlock:

因为redis在进行主从复制时是异步完成的，比如在clientA获取锁后，主redis复制数据到从redis过程中崩溃了，导致没有复制到从redis中，然后从redis选举出一个升级为主redis,造成新的主redis没有clientA 设置的锁，这时clientB尝试获取锁，并且能够成功获取锁，导致互斥失效；

该方案为了解决数据不一致的问题，直接舍弃了异步复制只使用 master 节点，同时由于舍弃了 slave，为了保证可用性，引入了 N 个节点，官方建议是 5。本次教学演示用3台实例来做说明。

客户端只有在满足下面的这两个条件时，才能认为是加锁成功。

条件1：客户端从超过半数（大于等于N/2+1）的Redis实例上成功获取到了锁；

条件2：客户端获取锁的总耗时没有超过锁的有效时间。

解决方案

落地实现

Redisson是java的redis客户端之一，提供了一些api方便操作redis

官网：https://redisson.org/ https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95

redisson解决分布式锁：https://github.com/redisson/redisson/wiki/8.-Distributed-locks-and-synchronizers

修改为9.0版(redisson)

• 这里使用redisson 则是使用官网写好的分布式锁代码，前面8.0版我们自己写的分布式锁就是借鉴这里的
• redisson底层用Lua脚本编写，已经保证原子性
• redisson实现自动续期：额外起一个线程，定期检查线程是否还持有锁，如果有则延长过期时间，使用“看门狗”定期检查(每1/3的锁时间检查1次），如果线程还持有锁，则刷新过期时间

<!--redisson-->
<dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.13.4</version>
</dependency>	

RedisConfig

@Configuration     //不写默认会有，这里使用@bean引入，重写了RedisConfig
public class RedisConfig{       //springboot连接redis以及redission@Beanpublic RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory){RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(lettuceConnectionFactory);   //使用lettuce作为Redis的Java客户端连接工厂//设置key序列化方式stringredisTemplate.setKeySerializer(new StringRedisSerializer());//设置value的序列化方式jsonredisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.afterPropertiesSet();return redisTemplate;}//单Redis节点模式，使用redission连接redis并创建redission客户端@Beanpublic Redisson redisson(){Config config = new Config();config.useSingleServer().setAddress("redis://192.168.248.134:6379").setDatabase(0).setPassword("123456");return (Redisson) Redisson.create(config);}
}

InventoryController

@RestController
@Api(tags = "redis分布式锁测试")
public class InventoryController{@Autowiredprivate InventoryService inventoryService;@ApiOperation("扣减库存，一次卖一个")@GetMapping(value = "/inventory/sale")public String sale(){return inventoryService.sale();}@ApiOperation("扣减库存saleByRedisson，一次卖一个")@GetMapping(value = "/inventory/saleByRedisson")public String saleByRedisson(){return inventoryService.saleByRedisson();}
}

InventoryService

@Service
@Slf4j
public class InventoryService2{@Autowiredprivate StringRedisTemplate stringRedisTemplate;@Value("${server.port}")private String port;@Autowiredprivate DistributedLockFactory distributedLockFactory;   //V9.1,引入Redisson对应的官网推荐RedLock算法实现类@Autowiredprivate Redisson redisson;public String saleByRedisson(){String retMessage = "";String key = "zzyyRedisLock";RLock redissonLock = redisson.getLock(key);redissonLock.lock();    //加锁try{//1 查询库存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//2 判断库存是否足够Integer inventoryNumber = result == null ? 0 : Integer.parseInt(result);//3 扣减库存if(inventoryNumber > 0) {stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventoryNumber));retMessage = "成功卖出一个商品，库存剩余: "+inventoryNumber;System.out.println(retMessage);}else{retMessage = "商品卖完了，o(╥﹏╥)o";}}finally {if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()){   //如果被当前线程持有锁redissonLock.unlock();                 //解锁}}return retMessage+"\t"+"服务端口号："+port;}
}

源码分析

源码分析1

只要涉及锁，就要符合规范，继承Lock接口

源码分析2

加锁方法 tryLockInnerAsync

• 通过exists判断，如果锁不存在，则设置值和过期时间，加锁成功
• 通过hexists判断，如果锁已存在，并且锁的是当前线程，则证明是重入锁，加锁成功
• 如果锁已存在，但锁的不是当前线程，则证明有其他线程持有锁。返回当前锁的过期时间(代表了锁key的剩余生存时间)，加锁失败

源码分析3
这里面初始化了一个定时器，dely 的时间是 internalLockLeaseTime/3。

在 Redisson 中，internalLockLeaseTime 是 30s，也就是每隔 10s 续期一次，每次 30s。

多机案例

理论参考来源：redis之父提出了Redlock算法解决这个问题

RedLock锁实现：见设计理念

步骤

准备

docker走起3台redis的master机器，本次设置3台master各自独立无从属关系

docker run -p 6381:6379 --name redis-master-1 -d redis
docker run -p 6382:6379 --name redis-master-2 -d redis
docker run -p 6383:6379 --name redis-master-3 -d redis

进入上一步刚启动的redis容器实例

docker exec -it redis-master-1 /bin/bash   或者 docker exec -it redis-master-1 redis-clidocker exec -it redis-master-2 /bin/bash   或者 docker exec -it redis-master-2 redis-clidocker exec -it redis-master-3 /bin/bash   或者 docker exec -it redis-master-3 redis-cli

POM

建Module：redis_redlock

<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.19.1</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.30</version></dependency><!--swagger--><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger2</artifactId><version>2.9.2</version></dependency><!--swagger-ui--><dependency><groupId>io.springfox</groupId><artifactId>springfox-swagger-ui</artifactId><version>2.9.2</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.4</version><scope>compile</scope></dependency><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>5.8.11</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId></exclude></excludes></configuration></plugin></plugins></build>

YML

server.port=9090
spring.application.name=redlock
spring.swagger2.enabled=truespring.redis.database=0    #指定redis的几号数据库
spring.redis.password=123456
spring.redis.timeout=3000
spring.redis.mode=singlespring.redis.pool.conn-timeout=3000
spring.redis.pool.so-timeout=3000
spring.redis.pool.size=10spring.redis.single.address1=192.168.248.134:6381
spring.redis.single.address2=192.168.248.134:6382
spring.redis.single.address3=192.168.248.134:6383

主启动

@SpringBootApplication
public class RedisRedlockApplication{public static void main(String[] args){SpringApplication.run(RedisRedlockApplication.class, args);}
}

业务类

import lombok.Data;
@Data
public class RedisPoolProperties {private int maxIdle;private int minIdle;private int maxActive;private int maxWait;private int connTimeout;private int soTimeout;/*** 池大小*/private  int size;
}
//
import lombok.Data;
@Data
public class RedisSingleProperties {private  String address1;private  String address2;private  String address3;
}

配置

//CacheConfiguration
@Configuration
//启用配置属性绑定application.yml中读取并绑定到RedisProperties类的实例上
@EnableConfigurationProperties(RedisProperties.class)  
public class CacheConfiguration {@AutowiredRedisProperties redisProperties;  //读取的所有关于Redis的配置信息@Bean    //Spring会使用该方法配置和初始化一个RedissonClient实例RedissonClient redissonClient1() {Config config = new Config();  //创建Config对象，这是Redisson配置的基础String node = redisProperties.getSingle().getAddress1();     //设置单机模式下的服务器地址node = node.startsWith("redis://") ? node : "redis://" + node;  //确保地址以redis://开头SingleServerConfig serverConfig = config.useSingleServer()  //指明将要配置的是单机模式的连接.setAddress(node)                                   //设置Redis服务器的地址.setTimeout(redisProperties.getPool().getConnTimeout())    //设置了连接超时时间.setConnectionPoolSize(redisProperties.getPool().getSize())  //连接池的大小.setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());  //最小空闲连接数量if (StringUtils.isNotBlank(redisProperties.getPassword())) {serverConfig.setPassword(redisProperties.getPassword());     //密码 }return Redisson.create(config);      //根据上述配置创建一个Redisson客户端实例}@BeanRedissonClient redissonClient2() {Config config = new Config();String node = redisProperties.getSingle().getAddress2();node = node.startsWith("redis://") ? node : "redis://" + node;SingleServerConfig serverConfig = config.useSingleServer().setAddress(node).setTimeout(redisProperties.getPool().getConnTimeout()).setConnectionPoolSize(redisProperties.getPool().getSize()).setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());if (StringUtils.isNotBlank(redisProperties.getPassword())) {serverConfig.setPassword(redisProperties.getPassword());}return Redisson.create(config);}@BeanRedissonClient redissonClient3() {Config config = new Config();String node = redisProperties.getSingle().getAddress3();node = node.startsWith("redis://") ? node : "redis://" + node;SingleServerConfig serverConfig = config.useSingleServer().setAddress(node).setTimeout(redisProperties.getPool().getConnTimeout()).setConnectionPoolSize(redisProperties.getPool().getSize()).setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());if (StringUtils.isNotBlank(redisProperties.getPassword())) {serverConfig.setPassword(redisProperties.getPassword());}return Redisson.create(config);}/*** 单机* @return*//*@Beanpublic Redisson redisson(){Config config = new Config();config.useSingleServer().setAddress("redis://192.168.111.147:6379").setDatabase(0);return (Redisson) Redisson.create(config);}*/
}

controller

@RestController
@Slf4j
public class RedLockController {public static final String CACHE_KEY_REDLOCK = "ATGUIGU_REDLOCK";  //分布式锁的缓存键@AutowiredRedissonClient redissonClient1;     //字段注入 根据类型 RedissonClient找到所有bean，在名称一样的前提下//在根据字段名称匹配  @Qualifier指定的名称，所以这里最好加上@Qualifier("redissonClient1")// 根据redissonClient1这个名字 找到对应bean实例//这样就可以在这使用 这个实例了@AutowiredRedissonClient redissonClient2;@AutowiredRedissonClient redissonClient3;boolean isLockBoolean;@GetMapping(value = "/multiLock")public String getMultiLock() throws InterruptedException{String uuid =  IdUtil.simpleUUID();String uuidValue = uuid+":"+Thread.currentThread().getId();//分别从三个Redisson客户端获取同一锁键CACHE_KEY_REDLOCK对应的锁对象RLock lock1 = redissonClient1.getLock(CACHE_KEY_REDLOCK); RLock lock2 = redissonClient2.getLock(CACHE_KEY_REDLOCK);RLock lock3 = redissonClient3.getLock(CACHE_KEY_REDLOCK);//使用这三个锁实例创建一个RedissonMultiLock对象redLock，这允许跨多个Redis实例或槽位实现分布式的多锁协调RedissonMultiLock redLock = new RedissonMultiLock(lock1, lock2, lock3);redLock.lock();  //尝试获取所有锁。这一步骤是阻塞的，直到所有锁都获取成功或超时try{System.out.println(uuidValue+"\t"+"---come in biz multiLock");  //模拟业务处理try { TimeUnit.SECONDS.sleep(30); } catch (InterruptedException e) { e.printStackTrace(); }System.out.println(uuidValue+"\t"+"---task is over multiLock");} catch (Exception e) {e.printStackTrace();log.error("multiLock exception ",e);} finally {redLock.unlock();  //释放所有锁log.info("释放分布式锁成功key:{}", CACHE_KEY_REDLOCK);}return "multiLock task is over  "+uuidValue;}
}

测试

http://localhost:9090/multilock

命令
ttl ATGUIGU_REDLOCK
HGETALL ATGUIGU_REDLOCK
shutdown
docker start redis-master-1
docker exec -it redis-master-1 redis-cli