1.前言
随着微服务的快速推进,分布式架构也得到蓬勃的发展,那么如何保证多进程之间的并发则成为需要考虑的问题。因为服务是分布式部署模式,本地锁Reentrantlock和Synchnorized就无法使用了,当然很多同学脱口而出的基于Redis的setnx锁由于上手简单,所以也被广泛使用,但是Redis的setnx锁存在无法保证原子性,所以Redisson目前备受推崇,今天我们一起来了解一下,并且用十分优雅的方式实现它。
当然实现分布式锁的方式有很多,像基于数据库表主键、基于表字段版本号、基于Redis的SETNX、REDLOCK、REDISSON以及Zookeeper等方式来实现,本文对以上锁的实现以及优缺点不在讨论,有兴趣的可以移步至此:《分布式锁》
本文重点讲解一下Redisson分布式锁的实现
2.Redisson是如何基于Redis实现分布式锁的原理
先看一下最简单的实现方式:
@Test
void 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 acquired
if (ttl == null) {
return;
}
// 订阅
RFuture<RedissonLockEntry> future = subscribe(threadId);
commandExecutor.syncSubscription(future);
try {
while (true) {
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
break;
}
// waiting for message
if (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