msb_6557
/
opsli-boot
mirror of https://github.com/hiparker/opsli-boot.git
1
1
Code Issues Projects Releases Wiki Activity

Redis 分布式锁

Browse Source
v1.4.1
Parker 4 years ago
parent 1c5789269e
commit 37ff6866ff
6 changed files with 213 additions and 69 deletions
Show Stats Download Patch File Download Diff File

1
opsli-plugins/opsli-plugins-redis/pom.xml
Unescape View File

@ -20,7 +20,6 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- 集成Redis缓存 END -->
</dependencies>

157
opsli-plugins/opsli-plugins-redis/src/main/java/org/opsli/plugins/redis/RedisPlugin.java
Unescape View File

@ -1,21 +1,28 @@
package org.opsli.plugins.redis;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.opsli.plugins.redis.exception.RedisPluginException;
import org.opsli.plugins.redis.lock.RedisLock;
import org.opsli.plugins.redis.msg.RedisMsg;
import org.opsli.plugins.redis.pushsub.entity.BaseSubMessage;
import org.opsli.plugins.redis.scripts.RedisPluginScript;
import org.opsli.plugins.redis.scripts.RedisScriptCache;
import org.opsli.plugins.redis.scripts.enums.RedisScriptsEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
/**
@ -33,6 +40,7 @@ public class RedisPlugin {
@Autowired
private RedisScriptCache redisScriptCache;
// ===================== 基础相关 =====================
/**
@ -83,18 +91,23 @@ public class RedisPlugin {
* @param argv 多参数
* @return object
*/
public Object callScript(RedisScriptsEnum scriptsEnum, List<String> keys, List<String> argv) {
public Object callScript(RedisScriptsEnum scriptsEnum, List<String> keys, Object... argv) {
// 获得Script脚本
RedisPluginScript script = redisScriptCache.getScript(scriptsEnum);
if(script == null){
return false;
}
RedisScript<Object> of = RedisScript.of(script.getScript());
return redisTemplate.execute(of,keys,argv);
DefaultRedisScript<Long> lockScript = new DefaultRedisScript<>();
lockScript.setScriptText(script.getScript());
lockScript.setResultType(Long.class);
return redisTemplate.execute(lockScript, keys, argv);
}
// ===================== Redis 锁相关 =====================
/*
* 不建议在 Redis集群下使用，且使用时 需要加一个看门口 来自动续命 ，否则会出现锁重入
*
* 分布式锁需要考虑的问题
* 1、这把锁没有失效时间，一旦解锁操作失败，就会导致锁记录一直在tair中，其他线程无法再获得到锁。
* 2、这把锁只能是非阻塞的，无论成功还是失败都直接返回。
@ -103,52 +116,124 @@ public class RedisPlugin {
/**
* Redis 加分布式锁
* @param lockName 锁名称
* @param acquireTimeOut 尝试获取锁等待时间
* @param lockTimeOut 锁有效时间
* @param redisLock 锁
* @return identifier 很重要，解锁全靠他 唯一凭证
*/
public String lockWithTimeOut(String lockName, long acquireTimeOut, int lockTimeOut) {
String identifier = UUID.randomUUID().toString();
List<String> keys = Collections.singletonList("lock:" + lockName);
List<String> argv = Arrays.asList(identifier,
String.valueOf(lockTimeOut));
long acquireTimeEnd = System.currentTimeMillis() + acquireTimeOut;
boolean acquired = false;
// 尝试获得锁
while (!acquired && (System.currentTimeMillis() < acquireTimeEnd)) {
Long ret = (Long) this.callScript(RedisScriptsEnum.REDIS_LOCK, keys, argv);
if(ret == null){
break;
}
if (1 == ret){
acquired = true;
} else {
try {
Thread.sleep(10);
} catch (InterruptedException ignored) {
log.error(ignored.getMessage(),ignored);
public RedisLock tryLock(RedisLock redisLock) {
//String identifier = UUID.randomUUID().toString().replaceAll("-","");
String identifier = "aaa";
redisLock = this.tryLock(redisLock,identifier);
if(redisLock != null){
log.info("分布式锁 - 开启： 锁名称: "+redisLock.getLockName()+" 锁凭证: \""+redisLock.getIdentifier()+"\"");
// 启动看门狗
this.lockDog(redisLock);
}
return redisLock;
}
/**
* Redis 加分布式锁
* @param redisLock 锁
* @return identifier 很重要，解锁全靠他 唯一凭证
*/
private RedisLock tryLock(RedisLock redisLock,String identifier) {
try {
List<String> keys = Collections.singletonList("lock:" + redisLock.getLockName());
long acquireTimeEnd = System.currentTimeMillis() + redisLock.getAcquireTimeOut();
boolean acquired = false;
// 尝试获得锁
while (!acquired && (System.currentTimeMillis() < acquireTimeEnd)) {
Long ret = (Long) this.callScript(RedisScriptsEnum.REDIS_LOCK, keys, identifier,redisLock.getLockTimeOut());
if(ret == null){
break;
}
if (1 == ret){
acquired = true;
} else {
try {
Thread.sleep(10);
} catch (InterruptedException ignored) {
log.error(ignored.getMessage(),ignored);
}
}
}
redisLock.setIdentifier(identifier);
return acquired ? redisLock : null;
}catch (Exception e){
log.error(e.getMessage(),e);
return null;
}
return acquired ? identifier : null;
}
/**
* Redis 释放分布式锁
* @param lockName 锁名称
* @param identifier 唯一凭证
* @param redisLock 锁
* @return boolean
*/
public boolean unLock(String lockName, String identifier) {
List<String> keys = Collections.singletonList("lock:" + lockName);
List<String> argv = Collections.singletonList(identifier);
Long ret = (Long) this.callScript(RedisScriptsEnum.REDIS_UN_LOCK, keys, argv);
if(ret == null){
public boolean unLock(RedisLock redisLock) {
try {
List<String> keys = Collections.singletonList("lock:" + redisLock.getLockName());
Long ret = (Long) this.callScript(RedisScriptsEnum.REDIS_UN_LOCK, keys, redisLock.getIdentifier());
// 减去线程锁
redisLock.unThreadLock();
log.info("分布式锁 - 解除： 锁名称: "+redisLock.getLockName()+" 锁凭证: "+redisLock.getIdentifier());
if(ret == null){
return false;
}
if(1 == ret){
return true;
}
return false;
}catch (Exception e){
log.error(e.getMessage(),e);
}
return 1 == ret;
return false;
}
/**
* Redis 分布式锁 - 看门狗 自动续命使用
* @param redisLock 锁
* @return boolean
*/
private void lockDog(RedisLock redisLock) {
Thread t = new Thread(()->{
try {
// 倒计时前续命
long countDownTime = 3000L;
// 锁释放时间
long lockTimeOutEnd = System.currentTimeMillis() + redisLock.getLockTimeOut();
boolean dogFlag = true;
// 看门狗检测 当前线程是否还存活
while (dogFlag) {
int lock = redisLock.getThreadLock();
if(lock <= 0){
dogFlag = false;
// 再一次确定 解锁 防止线程差 最后加锁
this.unLock(redisLock);
break;
}
try {
Thread.sleep(10);
} catch (InterruptedException ignored) {
log.error(ignored.getMessage(),ignored);
}
// 如果 距离倒计时 前 2000 毫秒还没有动作 则执行续命
if((System.currentTimeMillis()+countDownTime) >= lockTimeOutEnd){
Object o = this.tryLock(redisLock,redisLock.getIdentifier());
if(o != null){
lockTimeOutEnd = System.currentTimeMillis() + redisLock.getLockTimeOut();
log.info("分布式锁 - 续命： 锁名称: "+redisLock.getLockName()+" 锁凭证: "+redisLock.getIdentifier());
}
}
}
}catch (Exception e){
log.error(e.getMessage(),e);
}
});
t.setName("分布式锁看门狗："+" 锁名称: "+redisLock.getLockName()+" 锁凭证: "+redisLock.getIdentifier());
t.start();
}

26
opsli-plugins/opsli-plugins-redis/src/main/java/org/opsli/plugins/redis/conf/RedisPluginConfig.java
Unescape View File

@ -1,5 +1,6 @@
package org.opsli.plugins.redis.conf;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
@ -13,6 +14,7 @@ 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.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import javax.annotation.Resource;
@ -29,6 +31,8 @@ import java.util.Set;
@Configuration
public class RedisPluginConfig {
private static final FastJsonRedisSerializer<Object> FAST_JSON_REDIS_SERIALIZER = new FastJsonRedisSerializer<>(Object.class);
@Resource
private LettuceConnectionFactory factory;
@ -40,24 +44,20 @@ public class RedisPluginConfig {
public RedisTemplate<String, Object> redisTemplate() {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(factory);
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key采用String的序列化方式
template.setKeySerializer(stringRedisSerializer);
template.setKeySerializer(RedisSerializer.string());
// hash的key也采用String的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value序列化方式采用jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash的value序列化方式采用jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.setHashKeySerializer(RedisSerializer.string());
// value序列化方式采用 json
template.setValueSerializer(FAST_JSON_REDIS_SERIALIZER);
// hash的value序列化方式采用 json
template.setHashValueSerializer(FAST_JSON_REDIS_SERIALIZER);
template.afterPropertiesSet();
// 开启事务
template.setEnableTransactionSupport(true);
//template.setEnableTransactionSupport(true);
return template;
}

43
opsli-plugins/opsli-plugins-redis/src/main/java/org/opsli/plugins/redis/lock/RedisLock.java
Unescape View File

@ -0,0 +1,43 @@
package org.opsli.plugins.redis.lock;
import lombok.Data;
import lombok.Value;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @BelongsProject: opsli-boot
* @BelongsPackage: org.opsli.plugins.redis.lock
* @Author: Parker
* @CreateTime: 2020-09-16 00:51
* @Description: Redis 锁
*/
@Data
public class RedisLock {
/** 锁名称 */
private String lockName;
/** 尝试获取锁等待时间 */
private Long acquireTimeOut;
/** 锁有效时间 */
private Long lockTimeOut;
/** 锁凭证 */
private String identifier;
/** 线程锁 */
private AtomicInteger atomicInteger = new AtomicInteger(1);
/** 获得线程锁 */
public int getThreadLock(){
return atomicInteger.get();
}
/** 解除线程锁 */
public int unThreadLock(){
return atomicInteger.decrementAndGet();
}
}

31
opsli-plugins/opsli-plugins-redis/src/main/java/org/opsli/plugins/redis/scripts/RedisLockScript.java
Unescape View File

@ -22,17 +22,26 @@ public class RedisLockScript implements RedisPluginScript {
@Override
public String getScript() {
return "local key = KEYS[1]\n" +
"local identifier = ARGV[1]\n" +
"local lockTimeOut = ARGV[2]\n" +
"\n" +
"if redis.call(\"SETNX\", key, identifier) == 1 then\n" +
" redis.call(\"EXPIRE\", key, lockTimeOut)\n" +
" return 1\n" +
"elseif redis.call(\"TTL\", key) == -1 then\n" +
" redis.call(\"EXPIRE\", key, lockTimeOut)\n" +
"end\n" +
"return 0";
return "-- 加锁脚本\n"+
"-- key1：要加锁的名称 argv1:当前线程或主机的地址 argv2：锁存活的时间ms \n"+
"local expire_time = tonumber(ARGV[2])\n"+
"if redis.call('exists', KEYS[1]) == 0 then\n"+
" -- 锁不存在，创建一把锁，存入hash类型的值\n"+
" redis.call('hset', KEYS[1], ARGV[1], 1)\n"+
" -- 设置锁的存活时间，防止死锁\n"+
" redis.call('pexpire', KEYS[1], expire_time)\n"+
" return 1\n"+
"end\n"+
"if redis.call('hexists', KEYS[1], ARGV[1]) == 1 then\n"+
" -- 表示是同一线程重入\n"+
" redis.call('hincrby', KEYS[1], ARGV[1], 1)\n"+
" -- 重新设置锁的过期时间\n"+
" redis.call('pexpire', KEYS[1], expire_time)\n"+
" return 1\n"+
"end\n"+
"-- 没抢到锁，返回锁的剩余有效时间ms\n"+
"return 0\n";
}
}

24
opsli-plugins/opsli-plugins-redis/src/main/java/org/opsli/plugins/redis/scripts/RedisUnLockScript.java
Unescape View File

@ -22,14 +22,22 @@ public class RedisUnLockScript implements RedisPluginScript {
@Override
public String getScript() {
return "local key = KEYS[1]\n" +
"local identifier = ARGV[1]\n" +
"\n" +
"if redis.call(\"GET\", key) == identifier then\n" +
" redis.call(\"DEL\", key)\n" +
" return 1\n" +
"end\n" +
"return 0";
return "-- 解锁脚本\n"+
"-- 判断是当前线程持有锁，避免解了其他线程加的锁\n"+
"if redis.call('hexists',KEYS[1],ARGV[1]) == 1 then\n"+
" -- 重入次数大于1，扣减次数\n"+
" --if tonumber(redis.call('hget',KEYS[1],ARGV[1])) > 1 then\n"+
" -- return redis.call('hincrby', KEYS[1], ARGV[1], -1);\n"+
" -- 重入次数等于1，删除该锁\n"+
" --else\n"+
" redis.call('del', KEYS[1]);\n"+
" return 1;\n"+
" --end\n"+
"-- 判断不是当前线程持有锁，返回解锁失败\n"+
"else\n"+
" return 0;\n"+
"end\n";
}
}

Write Preview
Loading…
Cancel
Save