注解方式优雅的实现Redisson分布式锁

1.前言

随着微服务的快速推进，分布式架构也得到蓬勃的发展，那么如何保证多进程之间的并发则成为需要考虑的问题。因为服务是分布式部署模式，本地锁Reentrantlock和Synchnorized就无法使用了，当然很多同学脱口而出的基于Redis的setnx锁由于上手简单，所以也被广泛使用，但是Redis的setnx锁存在无法保证原子性，所以Redisson目前备受推崇，今天我们一起来了解一下，并且用十分优雅的方式实现它。

当然实现分布式锁的方式有很多，像基于数据库表主键、基于表字段版本号、基于Redis的SETNX、REDLOCK、REDISSON以及Zookeeper等方式来实现，本文对以上锁的实现以及优缺点不在讨论，有兴趣的可以移步至此：《分布式锁》

本文重点讲解一下Redisson分布式锁的实现

2.Redisson是如何基于Redis实现分布式锁的原理

先看一下最简单的实现方式：

    @Testvoid test1() {// 1、创建配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");// 2、根据 Config 创建出 RedissonClient 实例RedissonClient redissonClient = Redisson.create(config);//获取锁RLock lock = redissonClient.getLock("xxx-lock");try {// 2.加锁lock.lock();} finally {// 3.解锁lock.unlock();System.out.println("Finally，释放锁成功");}}

通过上面这段代码，我们看一下Redisson是如何基于Redis实现分布式锁的
下面的原理分析来自：《分布式锁-8.基于 REDISSON 做分布式锁》

2.1 加锁原理

通过上面的这段简单的代码，可以看出其加锁的方法主要依赖于其lock()方法，对于应的源码如下：

可以看到，调用getLock()方法后实际返回一个RedissonLock对象，在RedissonLock对象的lock()方法主要调用tryAcquire()方法

由于leaseTime == -1，于是走tryLockInnerAsync()方法，这个方法才是关键
首先，看一下evalWriteAsync方法的定义

<T, R> RFuture evalWriteAsync(String key, Codec codec, RedisCommand evalCommandType, String script, List keys, Object … params);

最后两个参数分别是keys和params

evalWriteAsync具体如何调用的呢？

commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,"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; " +"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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

结合上面的参数声明，我们可以知道，这里

KEYS[1]就是getName()
ARGV[2]是getLockName(threadId)

假设前面获取锁时传的name是“abc”，假设调用的线程ID是Thread-1，假设成员变量UUID类型的id是6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c

那么KEYS[1]=abc，ARGV[2]=6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1

因此，这段代码想表达什么呢？

1、判断有没有一个叫“abc”的key

2、如果没有，则在其下设置一个字段为“6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1”，值为“1”的键值对，并设置它的过期时间

3、如果存在，则进一步判断“6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1”是否存在，若存在，则其值加1，并重新设置过期时间

4、返回“abc”的生存时间（毫秒）

这里用的数据结构是hash，hash的结构是： key 字段1 值1 字段2 值2 。。。

用在锁这个场景下，key就表示锁的名称，也可以理解为临界资源，字段就表示当前获得锁的线程

所有竞争这把锁的线程都要判断在这个key下有没有自己线程的字段，如果没有则不能获得锁，如果有，则相当于重入，字段值加1（次数）
算法原理如下图所示：

2.1 解锁原理

protected RFuture<Boolean> unlockInnerAsync(long threadId) {return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end;" +"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.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));}

我们还是假设name=abc，假设线程ID是Thread-1

同理，我们可以知道

KEYS[1]是getName()，即KEYS[1]=abc

KEYS[2]是getChannelName()，即KEYS[2]=redisson_lock__channel:{abc}

ARGV[1]是LockPubSub.unlockMessage，即ARGV[1]=0

ARGV[2]是生存时间

ARGV[3]是getLockName(threadId)，即ARGV[3]=6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1

因此，上面脚本的意思是：

1、判断是否存在一个叫“abc”的key

2、如果不存在，向Channel中广播一条消息，广播的内容是0，并返回1

3、如果存在，进一步判断字段6f0829ed-bfd3-4e6f-bba3-6f3d66cd176c:Thread-1是否存在

4、若字段不存在，返回空，若字段存在，则字段值减1

5、若减完以后，字段值仍大于0，则返回0

6、减完后，若字段值小于或等于0，则广播一条消息，广播内容是0，并返回1；

可以猜测，广播0表示资源可用，即通知那些等待获取锁的线程现在可以获得锁了

2.3 等待

上面的加锁，解锁均是可以获取到锁资源的情况，那么当无法立即获取锁资源时，就需要等待

@Override
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {long threadId = Thread.currentThread().getId();Long ttl = tryAcquire(leaseTime, unit, threadId);// lock acquiredif (ttl == null) {return;}//    订阅RFuture<RedissonLockEntry> future = subscribe(threadId);commandExecutor.syncSubscription(future);try {while (true) {ttl = tryAcquire(leaseTime, unit, threadId);// lock acquiredif (ttl == null) {break;}// waiting for messageif (ttl >= 0) {getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {getEntry(threadId).getLatch().acquire();}}} finally {unsubscribe(future, threadId);}
//        get(lockAsync(leaseTime, unit));
}protected static final LockPubSub PUBSUB = new LockPubSub();protected RFuture<RedissonLockEntry> subscribe(long threadId) {return PUBSUB.subscribe(getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}protected void unsubscribe(RFuture<RedissonLockEntry> future, long threadId) {PUBSUB.unsubscribe(future.getNow(), getEntryName(), getChannelName(), commandExecutor.getConnectionManager().getSubscribeService());
}

这里会订阅Channel，当资源可用时可以及时知道，并抢占，防止无效的轮询而浪费资源

当资源可用用的时候，循环去尝试获取锁，由于多个线程同时去竞争资源，所以这里用了信号量，对于同一个资源只允许一个线程获得锁，其它的线程阻塞

3.Redisson分布式锁常规使用

本章讲主要讲述加锁的常规使用，Redisson分布式锁是基于Redis的Rlock锁，实现了JavaJUC包下的Lock接口。

3.1 添加maven依赖

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

3.2 REDISSON的牛刀小试

还是原理中的那段代码，稍作修改

    @GetMapping("test1")public String test1() {// 1、创建配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");// 2、根据 Config 创建出 RedissonClient 实例RedissonClient redissonClient = Redisson.create(config);//获取锁RLock lock = redissonClient.getLock("xxx-lock");try {// 2.加锁lock.lock();System.out.println(new Date()+"获取锁成功");//业务代码Thread.sleep(1000 * 3);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {// 3.解锁lock.unlock();System.out.println("Finally，释放锁成功");}System.out.println("finish");return "finish";}

在这里插入图片描述

上面这段代码做的事情很简单：

getLock获取锁，lock.lock进行加锁，会出现的问题就是lock拿不到锁一直等待，会进入阻塞状态，显然这样是不好的。

1.TryLock

返回boolean类型，和Reentrantlock的tryLock是一个意思，尝试获取锁，获取到就返回true，获取失败就返回false，不会使获不到锁的线程一直处于等待状态，返回false可以继续执行下面的业务逻辑，当然Ression锁内部也涉及到watchDog看门狗机制，主要作用就是给快过期的锁进行续期，主要用途就是使拿到锁的有限时间让业务执行完，再进行锁释放。

为了避免频繁的去书写创建redis连接的代码，所以，我们将获取锁和释放锁的过程简单封装一下

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;@Component
public class LockUtil {static Map<String, RLock> lockMap = new ConcurrentHashMap<>();/*** 获取redisson客户端** @return*/public static final RedissonClient getClient() {// 1、创建配置Config config = new Config();config.useSingleServer().setAddress("redis://127.0.0.1:6379");// 2、根据 Config 创建出 RedissonClient 实例RedissonClient redissonClient = Redisson.create(config);return redissonClient;}/*** 获取锁** @param lockName* @return*/public static boolean getLock(String lockName) {//获取锁RLock lock = getClient().getLock(lockName);try {if (lock.tryLock(2, 10, TimeUnit.SECONDS)) {lockMap.put(lockName, lock);return true;}return false;} catch (InterruptedException e) {throw new RuntimeException(e);}}public static boolean getLock(String lockName, long waitTime, long leaseTime, TimeUnit timeUnit) {//获取锁RLock lock = getClient().getLock(lockName);try {if (lock.tryLock(waitTime, leaseTime, timeUnit)) {lockMap.put(lockName, lock);return true;}return false;} catch (InterruptedException e) {throw new RuntimeException(e);}}/*** 解锁** @param lockName*/public static void unLock(String lockName) {RLock lock = lockMap.get(lockName);if (Objects.nonNull(lock) && lock.isHeldByCurrentThread()) {lock.unlock();lockMap.remove(lockName);}}
}

使用方式如下：

    @GetMapping("test2")public void test2() {try {if (LockUtil.getLock("ninesun")) {//执行业务代码System.out.println("业务代码");}} catch (Exception e) {System.out.println("获取锁失败");e.printStackTrace();} finally {//释放锁LockUtil.unLock("ninesun");}}

为了使我们实现的方式更加优雅，下面我们通过注解来实现

2.自定义注解实现锁机制

通常我们都会将redisson实例注入到方法类里面，然后调用加锁方法进行加锁，如果其他业务方法也需要加锁执行，将会产生很多重复代码，由此采用AOP切面的方式，只需要通过注解的方式就能将方法进行加锁处理。

2.1 添加切面依赖

        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-aop</artifactId></dependency>

2.2 自定义注解

import java.lang.annotation.*;
import java.util.concurrent.TimeUnit;/*** @author ninesun* @ClassName RedissonDistributedLock* @description: TODO* @date 2023年11月27日* @version: 1.0*/
@Documented
@Inherited
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
public @interface RedissonDistributedLock {String key() default "";int leaseTime() default 10;boolean autoRelease() default true;String errorDesc() default "系统正常处理，请稍后提交";int waitTime() default 1;TimeUnit timeUnit() default TimeUnit.SECONDS;
}

2.3 切面类实现

import com.example.demo.Utils.LockUtil;
import com.example.demo.annoation.RedissonDistributedLock;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.DefaultParameterNameDiscoverer;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;/*** @author ninesun* @ClassName RedisonDistributedLockHandler* @description: TODO* @date 2023年11月27日* @version: 1.0*/
@Aspect
@Component
public class RedisonDistributedLockHandler {private static final Logger log = LoggerFactory.getLogger(RedisonDistributedLockHandler.class);public RedisonDistributedLockHandler() {}@Around("@annotation(distributedLock)")public Object around(ProceedingJoinPoint joinPoint, RedissonDistributedLock distributedLock) throws Throwable {String lockName = this.getRedisKey(joinPoint, distributedLock);int leaseTime = distributedLock.leaseTime();String errorDesc = distributedLock.errorDesc();int waitTime = distributedLock.waitTime();TimeUnit timeUnit = distributedLock.timeUnit();Object var8;try {boolean lock = LockUtil.getLock(lockName, leaseTime, waitTime, timeUnit);if (!lock) {throw new RuntimeException(errorDesc);}var8 = joinPoint.proceed();} catch (Throwable var12) {log.error("执行业务方法异常", var12);throw var12;} finally {LockUtil.unLock(lockName);}return var8;}/*** 获取加锁的key** @param joinPoint* @param distributedLock* @return*/private String getRedisKey(ProceedingJoinPoint joinPoint, RedissonDistributedLock distributedLock) {String key = distributedLock.key();Object[] parameterValues = joinPoint.getArgs();MethodSignature signature = (MethodSignature) joinPoint.getSignature();Method method = signature.getMethod();DefaultParameterNameDiscoverer nameDiscoverer = new DefaultParameterNameDiscoverer();String[] parameterNames = nameDiscoverer.getParameterNames(method);if (StringUtils.isEmpty(key)) {if (parameterNames != null && parameterNames.length > 0) {StringBuffer sb = new StringBuffer();int i = 0;for (int len = parameterNames.length; i < len; ++i) {sb.append(parameterNames[i]).append(" = ").append(parameterValues[i]);}key = sb.toString();} else {key = "redissionLock";}return key;} else {SpelExpressionParser parser = new SpelExpressionParser();Expression expression = parser.parseExpression(key);if (parameterNames != null && parameterNames.length != 0) {EvaluationContext evaluationContext = new StandardEvaluationContext();for (int i = 0; i < parameterNames.length; ++i) {evaluationContext.setVariable(parameterNames[i], parameterValues[i]);}try {Object expressionValue = expression.getValue(evaluationContext);return expressionValue != null && !"".equals(expressionValue.toString()) ? expressionValue.toString() : key;} catch (Exception var13) {return key;}} else {return key;}}}
}

2.4具体使用

    @GetMapping("test3")@RedissonDistributedLock(key = "'updateUserInfo:'+#id", errorDesc = "请勿重复提交")public void test3(@RequestParam(value = "id") String id) {//业务代码}

方法头加自定义注解

key参数代表需要加锁的key
errorDesc获取锁失败提示报错信息

在这里插入图片描述

上面的演示示例是单机模式，我们线上使用的可能是redis集群以及哨兵模式，这个只需控制我们redis的连接方式即可。

3.3 分布式集群

1.集群模式

这个需要我们redis中开启cluster nodes

       Config config = new Config();config.useClusterServers().setScanInterval(2000) // cluster state scan interval in milliseconds.addNodeAddress("redis://127.0.0.1:7000", "redis://127.0.0.1:7001").addNodeAddress("redis://127.0.0.1:7002");RedissonClient redisson = Redisson.create(config);

2.哨兵模式

在使用哨兵模式时，需要创建SentinelServersConfig对象，并将其设置为Config对象的配置信息。代码创建SentinelServersConfig对象的方式如下：

SentinelServersConfig sentinelConfig = new SentinelServersConfig();
sentinelConfig.setMasterName("mymaster");
sentinelConfig.addSentinelAddress("redis://127.0.0.1:26379");
sentinelConfig.addSentinelAddress("redis://127.0.0.1:26380");
sentinelConfig.addSentinelAddress("redis://127.0.0.1:26381");
config.useSentinelServers().setMasterName("mymaster").addSentinelAddress("redis://127.0.0.1:26379").addSentinelAddress("redis://127.0.0.1:26380").addSentinelAddress("redis://127.0.0.1:26381");

根据Redisson的官方文档，可以根据自己的需要来调整Redisson的各种参数，以达到最优的性能表现。以下是一些常用的配置参数及其说明。

connectTimeout：连接超时时间，单位：毫秒
timeout：读写超时时间，单位：毫秒
retryAttempts：连接失败重试次数，-1表示不限制重试次数
retryInterval：重试时间间隔，单位：毫秒
threads：响应请求线程数，最大为16

3.Redisson配置了集群不生效

3.4 Redisson配置序列化

为了提高Redisson的性能表现，Redisson在数据存储时使用了高效的序列化机制。在Redisson中，默认使用的是JDK序列化机制，但是考虑到JDK的序列化机制在序列化性能、序列化结果可读性、可靠性等方面存在一些问题，因此Redisson提供了多种序列化方式供用户选择。

常用的序列化方式有三种：JDK序列化、FastJSON序列化和Kryo序列化。其中，Kryo序列化是性能最高的一种序列化方式，但是需要注意的是，Kryo序列化与JDK序列化不兼容，因此在使用Kryo序列化时需要注意操作系统的类型及JDK的版本。

如果要对Redisson的序列化机制进行定制，可以通过以下方式来实现。

// 基于Jackson序列化
SerializationConfig serialConfig = config.getCodec().getSerializationConfig();
serialConfig.setJacksonObjectMapper(new ObjectMapper());// 基于FastJSON序列化
SerializationConfig serialConfig = config.getCodec().getSerializationConfig();
serialConfig.setSerializer("com.alibaba.fastjson.JSON").setDecoder("com.alibaba.fastjson.JSON");// 基于Kryo序列化
SerializationConfig serialConfig = config.getCodec().getSerializationConfig();
Kryo kryo = new Kryo();
kryo.register(User.class);
kryo.register(Order.class);
kryo.register(Item.class);
kryo.register(ArrayList.class);
kryo.register(LinkedList.class);
kryo.register(RedisCommand.class);
UnicornKryoPool pool = new UnicornKryoPoolImpl(kryo);
serialConfig.setKryoPool(pool);

具体使用方式如下：

        //使用json序列化方式Codec codec = new JsonJacksonCodec();config.setCodec(codec);

至此单机模式下的基于Redission和注解实现的幂等控制就实现了，后面会将redis集群以及哨兵模式下的实现方式进行实现。

git地址：https://gitee.com/ninesuntec/distributed-locks.git