Redis之RedLock算法以及底層原理

自研redis分布式鎖存在的問題以及面試切入點
在這里插入圖片描述
lock加鎖關鍵邏輯

unlock解鎖的關鍵邏輯

使用Redis的分布式鎖

之前手寫的redis分布式鎖有什么缺點？？

Redis之父的RedLock算法
Redis也提供了Redlock算法，用來實現基于多個實例的分布式鎖。
鎖變量由多個實例維護，即使有實例發生了故障，鎖變量仍然是存在的，客戶端還是可以完成鎖操作。
Redlock算法是實現高可靠分布式鎖的一種有效解決方案，可以在實際開發中使用
官網
Redis分布式鎖
在這里插入圖片描述

在這里插入圖片描述
RedLock的設計理念
該方案也是基于（set 加鎖、Lua 腳本解鎖）進行改良的，所以redis之父antirez 只描述了差異的地方，大致方案如下。

假設我們有N個Redis主節點，例如 N = 5這些節點是完全獨立的，我們不使用復制或任何其他隱式協調系統，

為了取到鎖客戶端執行以下操作：
在這里插入圖片描述
該方案為了解決數據不一致的問題，直接舍棄了異步復制只使用 master 節點，同時由于舍棄了 slave，為了保證可用性，引入了 N 個節點，官方建議是 5。
本次教學演示用3臺實例來做說明。客戶端只有在滿足下面的這兩個條件時，才能認為是加鎖成功。

條件1：客戶端從超過半數（大于等于N/2+1）的Redis實例上成功獲取到了鎖；
條件2：客戶端獲取鎖的總耗時沒有超過鎖的有效時間。

解決方案與容錯公式
在這里插入圖片描述

RedLock的落地實現Redisson

github地址

https://github.com/redisson/redisson

https://redisson.pro/docs/configuration/#cluster-mode

在這里插入圖片描述
pom文件

 <dependency><groupId>org.redisson</groupId><artifactId>redisson</artifactId><version>3.19.1</version></dependency>

RedissonConfig配置類

package com.atguigu.redislock.config;import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;@Configuration
public class RedisConfig
{@Beanpublic RedisTemplate<String, Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory){RedisTemplate<String,Object> redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(lettuceConnectionFactory);//設置key序列化方式stringredisTemplate.setKeySerializer(new StringRedisSerializer());//設置value的序列化方式jsonredisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.setHashKeySerializer(new StringRedisSerializer());redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());redisTemplate.afterPropertiesSet();return redisTemplate;}@Beanpublic Redisson redisson(){Config config = new Config();config.useSingleServer().setAddress("redis://172.18.8.229:6379").setDatabase(0).setPassword("root");return (Redisson) Redisson.create(config);}
}

業務方法的改造

  @Autowiredprivate Redisson redisson;//V9版本public String saleV9(){String retMessage="";RLock redissonLock = redisson.getLock("redisLock");redissonLock.lock();try {//查詢庫存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//判斷庫存是否足夠Integer inventory =result==null?0: Integer.valueOf(result);//扣減庫存if(inventory>0){stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventory));retMessage="成功賣出一個商品，庫存剩余："+inventory;System.out.println(retMessage+"\t"+"服務端口號"+port);try {TimeUnit.SECONDS.sleep(120);} catch (InterruptedException e) {throw new RuntimeException(e);}}else{retMessage="商品賣完了";}}finally {redissonLock.unlock();}return retMessage+"\t"+"服務端口號"+port;}

Jemeter壓測
在這里插入圖片描述

這樣直接刪除鎖是有bug的

解決方案

if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread())
{
redissonLock.unlock();
} }

 @Autowiredprivate Redisson redisson;//V9版本public String saleV9(){String retMessage="";RLock redissonLock = redisson.getLock("redisLock");redissonLock.lock();try {//查詢庫存信息String result = stringRedisTemplate.opsForValue().get("inventory001");//判斷庫存是否足夠Integer inventory =result==null?0: Integer.valueOf(result);//扣減庫存if(inventory>0){stringRedisTemplate.opsForValue().set("inventory001",String.valueOf(--inventory));retMessage="成功賣出一個商品，庫存剩余："+inventory;System.out.println(retMessage+"\t"+"服務端口號"+port);}else{retMessage="商品賣完了";}}finally {if(redissonLock.isLocked() && redissonLock.isHeldByCurrentThread()){redissonLock.unlock();}        }return retMessage+"\t"+"服務端口號"+port;}

Redisson源碼解析

加鎖
可重入
續命
解鎖
分析步驟
Redis分布式鎖過期了，但是業務邏輯還沒處理完怎么辦？（還記得之前說過的緩存續命么）
守護線程續命

在獲取鎖成功后，給鎖加一個watch dog，watchdog會啟動一個定時任務，在鎖沒有被釋放且快要過期的時候會續期。

在這里插入圖片描述
源碼分析

通過redissson新建出來的鎖key,默認是30s

加鎖的核心代碼
在這里插入圖片描述

Lua腳本加鎖

通過 exists 判斷，如果鎖不存在，則設置值和過期時間，加鎖成功。
通過 hexists 判斷，如果鎖已存在，并且鎖的是當前線程，則證明是重入鎖，加鎖成功。
如果鎖已存在，但鎖的不是當前線程，則證明有其他線程持有鎖。返回當前鎖的過期時間（代表了鎖 key 的剩余生存時間），加鎖失敗。
看門狗的鎖續期

客戶端A加鎖成功，就會啟動一個watch dog看門狗，他是一個后臺線程，會每隔10秒檢查一下，如果客戶端A還持有鎖key，那么就會不斷的延長鎖key的生存時間，默認每次續命又從30秒新開始

Lua腳本執行看門狗的鎖續期

解鎖方法

多機案例
理論參考

實戰演示：
docker啟動三個redis實例

server.port=9090
spring.application.name=redlockspring.swagger2.enabled=truespring.redis.database=0
spring.redis.password=
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=172.18.8.229:6382
spring.redis.single.address2=172.18.8.229:6383
spring.redis.single.address3=172.18.8.229:6384

redis三個實例對應的配置類

package com.atguigu.redis.redlock.config;import org.apache.commons.lang3.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;@Configuration
@EnableConfigurationProperties(RedisProperties.class)
public class CacheConfiguration {@AutowiredRedisProperties redisProperties;@BeanRedissonClient redissonClient1() {Config config = new Config();String node = redisProperties.getSingle().getAddress1();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 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);}
}

controller的演示方法

@RestController
@Slf4j
public class RedLockController
{public static final String CACHE_KEY_REDLOCK = "ATGUIGU_REDLOCK";@Autowired RedissonClient redissonClient1;@Autowired RedissonClient redissonClient2;@Autowired RedissonClient redissonClient3;@GetMapping(value = "/multilock")public String getMultiLock(){String taskThreadID = Thread.currentThread().getId()+"";RLock lock1 = redissonClient1.getLock(CACHE_KEY_REDLOCK);RLock lock2 = redissonClient2.getLock(CACHE_KEY_REDLOCK);RLock lock3 = redissonClient3.getLock(CACHE_KEY_REDLOCK);RedissonMultiLock redLock = new RedissonMultiLock(lock1, lock2, lock3);redLock.lock();try{log.info("come in biz multilock:{}",taskThreadID);try { TimeUnit.SECONDS.sleep(30); } catch (InterruptedException e) { e.printStackTrace(); }log.info("task is over multilock:{}",taskThreadID);}catch (Exception e){e.printStackTrace();log.error("multilock exception:{}",e.getCause()+"\t"+e.getMessage());}finally {redLock.unlock();log.info("釋放分布式鎖成功key:{}",CACHE_KEY_REDLOCK);}return "multilock task is over: "+taskThreadID;}
}