【TJXT】Day 11

目录前言1. 集群下的锁监视器2. 分布式锁简单实现流程图3. 分布式锁问题3.1. 锁误删问题3.2. 超时释放锁问题3.3. 其他问题4. Redisson5. 通用分布式锁组件5.1AOP切面5.1.1. 实现思路分析5.1.2. 定义注解5.1.3. 定义切面5.1.4. 使用锁5.2. 工厂模式切换锁类型5.2.1. 锁类型枚举5.2.2. 改动切面类5.2.3. 工厂代码优化5.2.4. 使用5.3. 锁失败策略模式5.3.1. 策略分析5.3.2. 策略实现5.4. 基于SPEL的动态锁名5.4.1. SPEL表达式5.4.2. 解析SPEL6.Lua脚本6.1. 官网教程6.2. 中文教程6.3. Redis的Lua脚本前言记录学习笔记大家可以参考参考1. 集群下的锁监视器04.027-Monitor-工作原理_哔哩哔哩_bilibilihttps://www.bilibili.com/video/BV16J411h7Rd/?p76vd_source1ff0c1b434581723cf696ccc2f59ceaaSynchronized中的重量级锁底层就是基于锁监视器Monitor来实现的。简单来说就是锁对象头会指向一个锁监视器而在监视器中则会记录一些信息比如_owner持有锁的线程_recursions锁重入次数因此每一个锁对象都会指向一个锁监视器而每一个锁监视器同一时刻只能被一个线程持有这样就实现了互斥效果。但前提是多个线程使用的是同一把锁。比如有三个线程来争抢锁资源线程1获取锁成功如图所示此时其它线程想要获取锁会发现监视器中的_owner已经有值了就会获取锁失败。由于咱们代码在锁对象是用户id的字符串常量因此同一个用户肯定是同一把锁线程是绝对安全的。但问题来了我们的服务将来肯定会多实例不是形成集群。每一个实例都会有一个自己的JVM运行环境因此即便是同一个用户如果并发的发起了多个请求由于请求进入了多个JVM就会出现多个锁对象用户id对象自然就有多个锁监视器。此时就会出现每个JVM内部都有一个线程获取锁成功的情况没有达到互斥的效果并发安全问题就可能再次发生了显然我们不能让每个实例去使用各自的JVM内部锁监视器而是应该在多个实例外部寻找一个锁监视器多个实例争抢同一把锁。称之为分布式锁。分布式锁必须要满足的特征多JVM实例都可以访问互斥能满足上述特征的组件有很多因此实现分布式锁的方式也非常多例如基于MySQL基于Redis基于Zookeeper基于ETCD但目前使用最广泛的还应该是基于Redis的分布式锁。2. 分布式锁但是添加锁和过期时间也不是原子操作当两者中间出现宕机还是会导致死锁问题简单实现流程图3. 分布式锁问题3.1. 锁误删问题3.2. 超时释放锁问题3.3. 其他问题除了上述问题以外分布式锁还会碰到一些其它问题锁的重入问题同一个线程多次获取锁的场景目前不支持可能会导致死锁锁失败的重试问题获取锁失败后要不要重试目前是直接失败不支持重试Redis主从的一致性问题由于主从同步存在延迟当线程在主节点获取锁后从节点可能未同步锁信息。如果此时主宕机会出现锁失效情况。此时会有其它线程也获取锁成功。从而出现并发安全问题。……当然上述问题并非无法解决只不过会比较麻烦。例如原子性问题可以利用Redis的LUA脚本来编写锁操作确保原子性超时问题利用WatchDog看门狗机制获取锁成功时开启一个定时任务在锁到期前自动续期避免超时释放。而当服务宕机后WatchDog跟着停止运行不会导致死锁。锁重入问题可以模拟Synchronized原理放弃setnx而是利用Redis的Hash结构来记录锁的持有者以及重入次数获取锁时重入次数1释放锁是重入次数-1次数为0则锁删除主从一致性问题可以利用Redis官网推荐的RedLock机制来解决这些解决方案实现起来比较复杂因此我们通常会使用一些开源框架来实现分布式锁而不是自己来编码实现。目前对这些解决方案实现的比较完善的一个第三方组件Redisson4. RedissonRedissonhttps://redisson.pro/docs/1. 引入依赖!--redisson-- dependency groupIdorg.redisson/groupId artifactIdredisson/artifactId /dependency2. 配置Configuration public class RedisConfig { Bean public RedissonClient redissonClient() { // 配置类 Config config new Config(); // 添加redis地址这里添加了单点的地址也可以使用config.useClusterServers()添加集群地址 config.useSingleServer() .setAddress(redis://192.168.150.101:6379) .setPassowrd(123321); // 创建客户端 return Redisson.create(config); } }#########################################################################Slf4j ConditionalOnClass({RedissonClient.class, Redisson.class}) Configuration EnableConfigurationProperties(RedisProperties.class) public class RedissonConfig { private static final String REDIS_PROTOCOL_PREFIX redis://; private static final String REDISS_PROTOCOL_PREFIX rediss://; Bean ConditionalOnMissingBean public LockAspect lockAspect(RedissonClient redissonClient){ return new LockAspect(redissonClient); } Bean ConditionalOnMissingBean public RedissonClient redissonClient(RedisProperties properties){ log.debug(尝试初始化RedissonClient); // 1.读取Redis配置 RedisProperties.Cluster cluster properties.getCluster(); RedisProperties.Sentinel sentinel properties.getSentinel(); String password properties.getPassword(); int timeout 3000; Duration d properties.getTimeout(); if(d ! null){ timeout Long.valueOf(d.toMillis()).intValue(); } // 2.设置Redisson配置 Config config new Config(); if(cluster ! null !CollectionUtil.isEmpty(cluster.getNodes())){ // 集群模式 config.useClusterServers() .addNodeAddress(convert(cluster.getNodes())) .setConnectTimeout(timeout) .setPassword(password); }else if(sentinel ! null !StrUtil.isEmpty(sentinel.getMaster())){ // 哨兵模式 config.useSentinelServers() .setMasterName(sentinel.getMaster()) .addSentinelAddress(convert(sentinel.getNodes())) .setConnectTimeout(timeout) .setDatabase(0) .setPassword(password); }else{ // 单机模式 config.useSingleServer() .setAddress(String.format(redis://%s:%d, properties.getHost(), properties.getPort())) .setConnectTimeout(timeout) .setDatabase(0) .setPassword(password); } // 3.创建Redisson客户端 return Redisson.create(config); } private String[] convert(ListString nodesObject) { ListString nodes new ArrayList(nodesObject.size()); for (String node : nodesObject) { if (!node.startsWith(REDIS_PROTOCOL_PREFIX) !node.startsWith(REDISS_PROTOCOL_PREFIX)) { nodes.add(REDIS_PROTOCOL_PREFIX node); } else { nodes.add(node); } } return nodes.toArray(new String[0]); } }这个配置上添加了条件注解ConditionalOnClass({RedissonClient.class, Redisson.class})也就是说只要引用了tj-common并且引用了Redisson依赖这套配置就会生效。不引入Redisson依赖配置自然不会生效从而实现按需引入。RedissonClient的配置无需自定义Redis地址而是直接基于SpringBoot中的Redis配置即可。而且不管是Redis单机、Redis集群、Redis哨兵模式都可以支持3. 基本用法Autowired private RedissonClient redissonClient; Test void testRedisson() throws InterruptedException { // 1.获取锁对象指定锁名称 RLock lock redissonClient.getLock(anyLock); try { // 2.尝试获取锁参数waitTime、leaseTime、时间单位 boolean isLock lock.tryLock(1, 10, TimeUnit.SECONDS); if (!isLock) { // 获取锁失败处理 .. } else { // 获取锁成功处理 } } finally { // 4.释放锁 lock.unlock(); } }利用Redisson获取锁时可以传3个参数waitTime获取锁的等待时间。当获取锁失败后可以多次重试直到waitTime时间耗尽。waitTime默认-1即失败后立刻返回不重试。leaseTime锁超时释放时间。默认是30同时会利用WatchDog来不断更新超时时间。需要注意的是如果手动设置leaseTime值会导致WatchDog失效。TimeUnit时间单位5. 通用分布式锁组件Redisson的分布式锁使用并不复杂基本步骤包括1创建锁对象2尝试获取锁3处理业务4释放锁但是除了第3步以外其它都是非业务代码对业务的侵入较多可以发现非业务代码格式固定每次获取锁总是在重复编码。我们可不可以对这部分代码进行抽取和简化呢5.1AOP切面5.1.1. 实现思路分析要优化这部分代码需要通过整个流程来分析可以发现只有红框部分是业务功能业务前、后都是固定的锁操作。既然如此我们完全可以基于AOP的思想将业务部分作为切入点将业务前后的锁操作作为环绕增强。但是我们该如何标记这些切入点呢不是每一个service方法都需要加锁因此我们不能直接基于类来确定切入点另外需要加锁的方法可能也较多我们不能基于方法名作为切入点这样太麻烦。因此最好的办法是把加锁的方法给标记出来利用标记来确定切入点。如何标记呢最常见的办法就是基于注解来标记了。同时加锁时还有一些参数比如锁的key名称、锁的waitTime、releaseTime等等都可以基于注解来传参。因此注解的核心作用是两个标记切入点传递锁参数综上我们计划利用注解来标记切入点传递锁参数。同时利用AOP环绕增强来实现加锁、释放锁等操作。5.1.2. 定义注解注解本身起到标记作用同时还要带上锁参数锁名称锁等待时间锁超时时间时间单位package com.tianji.promotion.utils; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import java.util.concurrent.TimeUnit; Retention(RetentionPolicy.RUNTIME) Target(ElementType.METHOD) public interface MyLock { String name(); long waitTime() default 1; long leaseTime() default -1; TimeUnit unit() default TimeUnit.SECONDS; }5.1.3. 定义切面接下来我们定义一个环绕增强的切面实现加锁、释放锁代码实现如下package com.tianji.promotion.utils; import lombok.RequiredArgsConstructor; import org.aspectj.lang.ProceedingJoinPoint; import org.aspectj.lang.annotation.Around; import org.aspectj.lang.annotation.Aspect; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.stereotype.Component; Component Aspect RequiredArgsConstructor public class MyLockAspect implements Ordered{ private final RedissonClient redissonClient; Around(annotation(myLock)) public Object tryLock(ProceedingJoinPoint pjp, MyLock myLock) throws Throwable { // 1.创建锁对象 RLock lock redissonClient.getLock(myLock.name()); // 2.尝试获取锁 boolean isLock lock.tryLock(myLock.waitTime(), myLock.leaseTime(), myLock.unit()); // 3.判断是否成功 if(!isLock) { // 3.1.失败快速结束 throw new BizIllegalException(请求太频繁); } try { // 3.2.成功执行业务 return pjp.proceed(); } finally { // 4.释放锁 lock.unlock(); } } Override public int getOrder() { return 0; } }注意Spring中的AOP切面有很多会按照Order排序按照Order值从小到大依次执行。Spring事务AOP的order值是Integer.MAX_VALUE优先级最低。我们的分布式锁一定要先于事务执行因此我们的切面一定要实现Ordered接口指定order值小于Integer.MAX_VALUE即可。5.1.4. 使用锁5.2. 工厂模式切换锁类型5.2.1. 锁类型枚举public enum MyLockType { RE_ENTRANT_LOCK, // 可重入锁 FAIR_LOCK, // 公平锁 READ_LOCK, // 读锁 WRITE_LOCK, // 写锁 ; }5.2.2. 改动切面类但是这样不够优雅像这种有很多种不同的类型根据不同的类型我们要创建不同的对象可以用一种工厂的模式。如果我们把每一种具体的这个类型和它对应的操作封装成键值对将来在这个地方要获取锁的时候直接根据这个类型取出他的这种操作然后直接去调用。那么怎么样去把这个类型和它对应的操作进行键值映射呢?要有一种抽象的思想其实所谓的操作就是一段行为行为在程序中就是方法或者讲函数。所以可以定义一个map map的key就是锁的类型值就是一个一段行为。这个行为是干什么这个行为的参数是一个字符串这个行为的返回值是RLock有参有返回值的行为叫function就是函数式接口。5.2.3. 工厂代码优化package com.tianji.promotion.utils; import org.redisson.api.RLock; import org.redisson.api.RedissonClient; import org.springframework.stereotype.Component; import java.util.EnumMap; import java.util.Map; import java.util.function.Function; import static com.tianji.promotion.utils.MyLockType.*; Component public class MyLockFactory { private final MapMyLockType, FunctionString, RLock lockHandlers; public MyLockFactory(RedissonClient redissonClient) { this.lockHandlers new EnumMap(MyLockType.class); this.lockHandlers.put(RE_ENTRANT_LOCK, redissonClient::getLock); this.lockHandlers.put(FAIR_LOCK, redissonClient::getFairLock); this.lockHandlers.put(READ_LOCK, name - redissonClient.getReadWriteLock(name).readLock()); this.lockHandlers.put(WRITE_LOCK, name - redissonClient.getReadWriteLock(name).writeLock()); } public RLock getLock(MyLockType lockType, String name){ return lockHandlers.get(lockType).apply(name); } }说明MyLockFactory内部持有了一个Mapkey是锁类型枚举值是创建锁对象的Function。注意这里不是存锁对象因为锁对象必须是多例的不同业务用不同锁对象同一个业务用相同锁对象。MyLockFactory内部的Map采用了EnumMap。只有当Key是枚举类型时可以使用EnumMap其底层不是hash表而是简单的数组。由于枚举项数量固定因此这个数组长度就等于枚举项个数然后按照枚举项序号作为角标依次存入数组。这样就能根据枚举项序号作为角标快速定位到数组中的数据。原本基于这个map去get一个具体的key那它就需要对这个key呢进行一个哈希运算然后呢对数组取模找到数组角标从而才能找到对应的这个操作那这样一来这个流程就变复杂。这个性能上也会受到影响。原来用的是哈希map需要把它改造成一个新的map叫ennum map。顾名思义就是一个枚举的一个map那它有什么特点在创建的时候不用你去指定具体的这个哈希map的大小。是你的这个Key的类型也就是说你把这个枚举的类型告诉他就可以了把class告诉他。因为你把枚举的类型告诉他你的枚举里边将来有几个枚举项那么这个map的大小是不是就应该是多少自己会去判断。那既然枚举项的数量是固定的可以认为这个map的大小也是固定的所以将来map不需要去做动态的扩容而且呢那枚举项由于是固定的那因此我们是不是可以把它们给个编号那这样一来map的内部就不用去做什么哈希运算了将来就对应数组角标零他就不需要做哈希运算然后取模之类。就找对应的那个角标位就能快速拿到这个操作因为lockfactory以及成为了bean,而且redissonclient也是bean,所以自动注入5.2.4. 使用5.3. 锁失败策略模式应该怎么样去实现这样一种根据场景不同他选择的这个处理逻辑也不同的这种效果就是策略模式。5.3.1. 策略分析5.3.2. 策略实现先确定抽象方法方法的参数有哪些要释放锁需要得到锁也就是RLock要根据等待时间时间单位等信息所以还需要前面编写的切面注解MyLock。然后在枚举内部类实现抽象方法。package com.tianji.promotion.utils; import com.tianji.common.exceptions.BizIllegalException; import org.redisson.api.RLock; public enum MyLockStrategy { SKIP_FAST(){ Override public boolean tryLock(RLock lock, MyLock prop) throws InterruptedException { return lock.tryLock(0, prop.leaseTime(), prop.unit()); } }, FAIL_FAST(){ Override public boolean tryLock(RLock lock, MyLock prop) throws InterruptedException { boolean isLock lock.tryLock(0, prop.leaseTime(), prop.unit()); if (!isLock) { throw new BizIllegalException(请求太频繁); } return true; } }, KEEP_TRYING(){ Override public boolean tryLock(RLock lock, MyLock prop) throws InterruptedException { lock.lock( prop.leaseTime(), prop.unit()); return true; } }, SKIP_AFTER_RETRY_TIMEOUT(){ Override public boolean tryLock(RLock lock, MyLock prop) throws InterruptedException { return lock.tryLock(prop.waitTime(), prop.leaseTime(), prop.unit()); } }, FAIL_AFTER_RETRY_TIMEOUT(){ Override public boolean tryLock(RLock lock, MyLock prop) throws InterruptedException { boolean isLock lock.tryLock(prop.waitTime(), prop.leaseTime(), prop.unit()); if (!isLock) { throw new BizIllegalException(请求太频繁); } return true; } }, ; public abstract boolean tryLock(RLock lock, MyLock prop) throws InterruptedException; }5.4. 基于SPEL的动态锁名在当前业务中我们的锁对象本来应该是当前登录用户是动态获取的。而加锁是基于注解参数添加的在编码时就需要指定。怎么办Spring中提供了一种表达式语法称为SPEL表达式可以执行java代码获取任意参数。思路我们可以让用户指定锁名称参数时不要写死而是基于SPEL表达式。在创建锁对象时解析SPEL表达式动态获取锁名称。5.4.1. SPEL表达式SPEL的表达式语法可以参考官网文档6. Spring Expression Language (SpEL)https://docs.spring.io/spring-framework/docs/3.0.x/reference/expressions.html中文文档8. Spring 表达式语言 (SpEL)https://itmyhome.com/spring/expressions.html首先在使用锁注解时锁名称可以利用SPEL表达式例如我们指定锁名称中要包含参数中的用户id则可以这样写而如果是通过UserContext.getUser()获取则可以利用下面的语法这里T(类名).方法名()就是调用静态方法。5.4.2. 解析SPEL在切面中我们需要基于注解中的锁名称做动态解析而不是直接使用名称其中获取锁名称用的是getLockName()这个方法/** * SPEL的正则规则 */ private static final Pattern pattern Pattern.compile(\\#\\{([^\\}]*)\\}); /** * 方法参数解析器 */ private static final ParameterNameDiscoverer parameterNameDiscoverer new DefaultParameterNameDiscoverer(); /** * 解析锁名称 * param name 原始锁名称 * param pjp 切入点 * return 解析后的锁名称 */ private String getLockName(String name, ProceedingJoinPoint pjp) { // 1.判断是否存在spel表达式 if (StringUtils.isBlank(name) || !name.contains(#)) { // 不存在直接返回 return name; } // 2.构建context,也就是SPEL表达式获取参数的上下文环境这里上下文就是切入点的参数列表 EvaluationContext context new MethodBasedEvaluationContext( TypedValue.NULL, resolveMethod(pjp), pjp.getArgs(), parameterNameDiscoverer); // 3.构建SPEL解析器 ExpressionParser parser new SpelExpressionParser(); // 4.循环处理因为表达式中可以包含多个表达式 Matcher matcher pattern.matcher(name); while (matcher.find()) { // 4.1.获取表达式 String tmp matcher.group(); String group matcher.group(1); // 4.2.这里要判断表达式是否以 T字符开头这种属于解析静态方法不走上下文 Expression expression parser.parseExpression(group.charAt(0) T ? group : # group); // 4.3.解析出表达式对应的值 Object value expression.getValue(context); // 4.4.用值替换锁名称中的SPEL表达式 name name.replace(tmp, ObjectUtils.nullSafeToString(value)); } return name; } private Method resolveMethod(ProceedingJoinPoint pjp) { // 1.获取方法签名 MethodSignature signature (MethodSignature)pjp.getSignature(); // 2.获取字节码 Class? clazz pjp.getTarget().getClass(); // 3.方法名称 String name signature.getName(); // 4.方法参数列表 Class?[] parameterTypes signature.getMethod().getParameterTypes(); return tryGetDeclaredMethod(clazz, name, parameterTypes); } private Method tryGetDeclaredMethod(Class? clazz, String name, Class? ... parameterTypes){ try { // 5.反射获取方法 return clazz.getDeclaredMethod(name, parameterTypes); } catch (NoSuchMethodException e) { Class? superClass clazz.getSuperclass(); if (superClass ! null) { // 尝试从父类寻找 return tryGetDeclaredMethod(superClass, name, parameterTypes); } } return null; }6.Lua脚本6.1. 官网教程Lua 5.4 Reference Manual - contentshttps://www.lua.org/manual/5.4/6.2. 中文教程Lua 教程_w3cschoolhttps://www.w3cschool.cn/lua/6.3. Redis的Lua脚本Redis Lua | Docshttps://redis.io/docs/latest/manual/programmability/lua-api/spring - 快速入门Redis调用Lua脚本及使用场景介绍 - 个人文章 - SegmentFault 思否https://segmentfault.com/a/1190000039287716实战篇-15.分布式锁-Lua脚本解决多条命令原子性问题_哔哩哔哩_bilibilihttps://www.bilibili.com/video/BV1cr4y1671t/?p62spm_id_frompageDrivervd_source1ff0c1b434581723cf696ccc2f59ceaa