简单的redis分布式锁实现
1.需求
我们公司想实现一个简单的分布式锁，用于服务启动初始化执行init方法的时候，只执行一次，避免重复执行加载缓存规则的代码，还有预防高并发流程发起部分，产品超发，多发问题。所以结合网上信息，自己简单实现了一个redis分布式锁，可以进行单次资源锁定，排队锁定(没有实现权重，按照时间长短争夺锁信息)，还有锁定业务未完成，需要延期锁等简单方法，死锁则是设置过期时间即可。期间主要用到的技术为redis，延时线程池，LUA脚本，比较简单，此处记录一下，方便下次学习查看。
2.具体实现
整体配置相对简单，主要是编写redisUtil工具类，实现redis的简单操作，编写分布式锁类SimpleDistributeLock，主要内容都在此锁的实现类中，SimpleDistributeLock实现类主要实现方法如下：
- 1.一次抢夺加锁方法 tryLock
- 2.连续排队加锁方法tryContinueLock，此方法中间有调用线程等待Thread.sleep方法防止防止StackOverFlow异常，比较耗费资源，后续应该需要优化处理
- 3.重入锁tryReentrantLock，一个资源调用过程中，处于加锁状态仍然可以再次加锁，重新刷新其过期时间
- 4.刷新锁过期时间方法resetLockExpire
- 5.释放锁方法，注意，释放过程中需要传入加锁的value信息，以免高并发情况下多线程锁信息被其他线程释放锁操作误删
2.1 redis基本操作工具类redisUtil
package cn.git.redis;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.dao.DataAccessException;
import org.springframework.data.geo.*;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.util.CollectionUtils;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
* @program: bank-credit-sy
* @description: 封装redis的工具类
* @author: lixuchun
* @create: 2021-01-23 11:53
*/
public class RedisUtil {
/**
* 模糊查询匹配
*/
private static final String FUZZY_ENQUIRY_KEY = "*";
@Autowired
@Qualifier("redisTemplate")
private RedisTemplate<String, Object> redisTemplate;
public void setRedisTemplate(RedisTemplate<String, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}
/**
* 指定缓存失效时间
*
* @param key 键
* @param time 时间(秒)
* @return
*/
public boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据key 获取过期时间
*
* @param key 键 不能为null
* @return 时间(秒) 返回0代表为永久有效
*/
public long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/**
* 判断key是否存在
*
* @param key 键
* @return true 存在 false不存在
*/
public boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
return false;
}
}
/**
* 删除缓存
*
* @param key 可以传一个值 或多个
*/
@SuppressWarnings("unchecked")
public void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete(CollectionUtils.arrayToList(key));
}
}
}
/**
* 普通缓存获取
*
* @param key 键
* @return 值
*/
public Object get(String key) {
return key == null ? null : redisTemplate.opsForValue().get(key);
}
/**
* 普通缓存放入
*
* @param key 键
* @param value 值
* @return true成功 false失败
*/
public boolean set(String key, Object value) {
try {
redisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 普通缓存放入并设置时间
*
* @param key 键
* @param value 值
* @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
* @return true成功 false 失败
*/
public boolean set(String key, Object value, long time) {
try {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 如果不存在，则设置对应key,value 键值对,并且设置过期时间
* @param key 锁key
* @param value 锁值
* @param time 时间单位second
* @return 设定结果
*/
/**
public Boolean setNxEx(String key, String value, long time) {
Boolean setResult = (Boolean) redisTemplate.execute((RedisCallback) connection -> {
RedisStringCommands.SetOption setOption = RedisStringCommands.SetOption.ifAbsent();
// 设置过期时间
Expiration expiration = Expiration.seconds(time);
// 执行setnx操作
Boolean result = connection.set(key.getBytes(StandardCharsets.UTF_8),
value.getBytes(StandardCharsets.UTF_8), expiration, setOption);
return result;
});
return setResult;
}
**/
/**
* 如果不存在，则设置对应key,value 键值对,并且设置过期时间
* @param key 锁key
* @param value 锁值
* @param time 时间单位second
* @return 设定结果
*/
public Boolean setNxEx(String key, String value, long time) {
return redisTemplate.opsForValue().setIfAbsent(key, value, time, TimeUnit.SECONDS);
}
/**
* 递增
*
* @param key 键
* @return
*/
public long incr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递增因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, delta);
}
/**
* 递减
*
* @param key 键
* @return
*/
public long decr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递减因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, -delta);
}
/**
* HashGet
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return 值
*/
public Object hget(String key, String item) {
return redisTemplate.opsForHash().get(key, item);
}
/**
* 获取hashKey对应的所有键值
*
* @param key 键
* @return 对应的多个键值
*/
public Map<Object, Object> hmget(String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* 获取hashKey对应的所有键值
*
* @param key 键
* @return 对应的多个键值
*/
public List<Object> hmget(String key, List<Object> itemList) {
return redisTemplate.opsForHash().multiGet(key, itemList);
}
/**
* 获取key对应的hashKey值
*
* @param key 键
* @param hashKey 键
* @return 对应的键值
*/
public Object hmget(String key, String hashKey) {
return redisTemplate.opsForHash().get(key, hashKey);
}
/**
* HashSet
*
* @param key 键
* @param map 对应多个键值
* @return true 成功 false 失败
*/
public boolean hmset(String key, Map<String, Object> map) {
try {
redisTemplate.opsForHash().putAll(key, map);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* HashSet 并设置时间
*
* @param key 键
* @param map 对应多个键值
* @param time 时间(秒)
* @return true成功 false失败
*/
public boolean hmset(String key, Map<Object, Object> map, long time) {
try {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value) {
try {
redisTemplate.opsForHash().put(key, item, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value, long time) {
try {
redisTemplate.opsForHash().put(key, item, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 删除hash表中的值
*
* @param key 键 不能为null
* @param item 项 可以使多个 不能为null
*/
public void hdel(String key, Object... item) {
redisTemplate.opsForHash().delete(key, item);
}
/**
* 判断hash表中是否有该项的值
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return true 存在 false不存在
*/
public boolean hHasKey(String key, String item) {
return redisTemplate.opsForHash().hasKey(key, item);
}
/**
* hash递增 如果不存在,就会创建一个 并把新增后的值返回
*
* @param key 键
* @param item 项
* @param by 要增加几(大于0)
* @return
*/
public double hincr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, by);
}
/**
* hash递减
*
* @param key 键
* @param item 项
* @param by 要减少记(小于0)
* @return
*/
public double hdecr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, -by);
}
/**
* 根据key获取Set中的所有值
*
* @param key 键
* @return
*/
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 根据value从一个set中查询,是否存在
*
* @param key 键
* @param value 值
* @return true 存在 false不存在
*/
public boolean sHasKey(String key, Object value) {
try {
return redisTemplate.opsForSet().isMember(key, value);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将数据放入set缓存
*
* @param key 键
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 将set数据放入缓存
*
* @param key 键
* @param time 时间(秒)
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSetAndTime(String key, long time, Object... values) {
try {
Long count = redisTemplate.opsForSet().add(key, values);
if (time > 0) {
expire(key, time);
}
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 获取set缓存的长度
*
* @param key 键
* @return
*/
public long sGetSetSize(String key) {
try {
return redisTemplate.opsForSet().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 移除值为value的
*
* @param key 键
* @param values 值 可以是多个
* @return 移除的个数
*/
public long setRemove(String key, Object... values) {
try {
Long count = redisTemplate.opsForSet().remove(key, values);
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 获取list缓存的内容
*
* @param key 键
* @param start 开始
* @param end 结束 0 到 -1代表所有值
* @return
*/
public List<Object> lGet(String key, long start, long end) {
try {
return redisTemplate.opsForList().range(key, start, end);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 获取list缓存的长度
*
* @param key 键
* @return
*/
public long lGetListSize(String key) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 通过索引 获取list中的值
*
* @param key 键
* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
* @return
*/
public Object lGetIndex(String key, long index) {
try {
return redisTemplate.opsForList().index(key, index);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, Object value) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, Object value, long time) {
try {
redisTemplate.opsForList().rightPush(key, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, List<Object> value) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, List<Object> value, long time) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据索引修改list中的某条数据
*
* @param key 键
* @param index 索引
* @param value 值
* @return
*/
public boolean lUpdateIndex(String key, long index, Object value) {
try {
redisTemplate.opsForList().set(key, index, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 移除N个值为value
*
* @param key 键
* @param count 移除多少个
* @param value 值
* @return 移除的个数
*/
public long lRemove(String key, long count, Object value) {
try {
Long remove = redisTemplate.opsForList().remove(key, count, value);
return remove;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
public void testAdd(Double X, Double Y, String accountId) {
Long addedNum = redisTemplate.opsForGeo()
.add("cityGeoKey", new Point(X, Y), accountId);
System.out.println(addedNum);
}
public Long addGeoPoin() {
Point point = new Point(123.05778991994906, 41.188314667658965);
Long addedNum = redisTemplate.opsForGeo().geoAdd("cityGeoKey", point, 3);
return addedNum;
}
public void testNearByPlace() {
Distance distance = new Distance(100, Metrics.KILOMETERS);
RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands
.GeoRadiusCommandArgs
.newGeoRadiusArgs()
.includeDistance()
.includeCoordinates()
.sortAscending()
.limit(5);
GeoResults<RedisGeoCommands.GeoLocation<Object>> results = redisTemplate.opsForGeo()
.radius("cityGeoKey", "北京", distance, args);
System.out.println(results);
}
public GeoResults<RedisGeoCommands.GeoLocation<Object>> testGeoNearByXY(Double X, Double Y) {
Distance distance = new Distance(100, Metrics.KILOMETERS);
Circle circle = new Circle(X, Y, Metrics.KILOMETERS.getMultiplier());
RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands
.GeoRadiusCommandArgs
.newGeoRadiusArgs()
.includeDistance()
.includeCoordinates()
.sortAscending();
GeoResults<RedisGeoCommands.GeoLocation<Object>> results = redisTemplate.opsForGeo()
.radius("cityGeoKey", circle, distance, args);
System.err.println(results);
return results;
}
/**
* @Description: 执行lua脚本，只对key进行操作
* @Param: [redisScript, keys]
* @return: java.lang.Long
* @Date: 2021/2/21 15:00
*/
public Long executeLua(RedisScript<Long> redisScript, List keys) {
return redisTemplate.execute(redisScript, keys);
}
/**
* @Description: 执行lua脚本，只对key进行操作
* @Param: [redisScript, keys, value]
* @return: java.lang.Long
* @Date: 2021/2/21 15:00
*/
public Long executeLuaCustom(RedisScript<Long> redisScript, List keys, Object ...value) {
return redisTemplate.execute(redisScript, keys, value);
}
/**
* @Description: 执行lua脚本，只对key进行操作
* @Param: [redisScript, keys, value]
* @return: java.lang.Long
* @Date: 2021/2/21 15:00
*/
public Boolean executeBooleanLuaCustom(RedisScript<Boolean> redisScript, List keys, Object ...value) {
return redisTemplate.execute(redisScript, keys, value);
}
/**
* 时间窗口限流
* @param key key
* @param timeWindow 时间窗口
* @return
*/
public Integer rangeByScore(String key, Integer timeWindow) {
// 获取当前时间戳
Long currentTime = System.currentTimeMillis();
Set<Object> rangeSet = redisTemplate.opsForZSet().rangeByScore(key, currentTime - timeWindow, currentTime);
if (ObjectUtil.isNotNull(rangeSet)) {
return rangeSet.size();
} else {
return 0;
}
}
/**
* 新增Zset
* @param key
*/
public String addZset(String key) {
String value = IdUtil.simpleUUID();
Long currentTime = System.currentTimeMillis();
redisTemplate.opsForZSet().add(key, value, currentTime);
return value;
}
/**
* 删除Zset
* @param key
*/
public void removeZset(String key, String value) {
// 参数存在校验
if (ObjectUtil.isNotNull(redisTemplate.opsForZSet().score(key, value))) {
redisTemplate.opsForZSet().remove(key, value);
}
}
/**
* 通过前缀key值获取所有key内容(hash)
* @param keyPrefix 前缀key
* @param fieldArray 查询对象列信息
*/
public List<Object> getPrefixKeys(String keyPrefix, byte[][] fieldArray) {
if (StrUtil.isBlank(keyPrefix)) {
return null;
}
keyPrefix = keyPrefix.concat(FUZZY_ENQUIRY_KEY);
// 所有完整key值
Set<String> keySet = redisTemplate.keys(keyPrefix);
List<Object> objectList = redisTemplate.executePipelined(new RedisCallback<Object>() {
/**
* Gets called by {@link RedisTemplate} with an active Redis connection. Does not need to care about activating or
* closing the connection or handling exceptions.
*
* @param connection active Redis connection
* @return a result object or {@code null} if none
* @throws DataAccessException
*/
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
for (String key : keySet) {
connection.hMGet(key.getBytes(StandardCharsets.UTF_8), fieldArray);
}
return null;
}
});
return objectList;
}
}
2.2 SimpleDistributeLock实现
具体锁以及解锁业务实现类，具体如下所示
package cn.git.common.lock;
import cn.git.redis.RedisUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;
import java.util.Collections;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* 简单分布式锁
* 可以实现锁的重入，锁自动延期
* @program: bank-credit-sy
* @author: lixuchun
* @create: 2022-04-25
*/
@Slf4j
@Component
public class SimpleDistributeLock {
/**
* 活跃的锁集合
*/
private volatile static CopyOnWriteArraySet ACTIVE_KEY_SET = new CopyOnWriteArraySet();
/**
* 定时线程池，续期使用
*/
private static ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(5);
/**
* 解锁脚本, 脚本参数 KEYS[1]: 传入的key, ARGV[1]: 传入的value
* // 如果没有key,直接返回1
* if redis.call('EXISTS',KEYS[1]) == 0 then
* return 1
* else
* // 如果key存在,并且value与传入的value相等,删除key,返回1，如果值不等,返回0
* if redis.call('GET',KEYS[1]) == ARGV[1] then
* return redis.call('DEL',KEYS[1])
* else
* return 0
* end
* end
*/
private static final String UNLOCK_SCRIPT = "if redis.call('EXISTS',KEYS[1]) == 0 then return 1 else if redis.call('GET',KEYS[1]) == ARGV[1] then return redis.call('DEL',KEYS[1]) else return 0 end end";
/**
* lua脚本参数介绍 KEYS[1]：传入的key ARGV[1]：传入的value ARGV[2]：传入的过期时间
* // 如果成功设置keys，value值，然后设定过期时间，直接返回1
* if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then
* redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
* return 1
* else
* // 如果key存在，并且value值相等，则重置过期时间，直接返回1，值不等则返回0
* if redis.call('GET', KEYS[1]) == ARGV[1] then
* redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
* return 1
* else
* return 0
* end
* end
*/
private static final String REENTRANT_LOCK_LUA = "if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2])) return 1 else if redis.call('GET', KEYS[1]) == ARGV[1] then redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2])) return 1 else return 0 end end";
/**
* 续期脚本
* // 如果key存在，并且value值相等，则重置过期时间，直接返回1，值不等则返回0
* if redis.call('EXISTS',KEYS[1]) == 1 and redis.call('GET',KEYS[1]) == ARGV[1] then
* redis.call('EXPIRE',KEYS[1],tonumber(ARGV[2]))
* return 1
* else
* return 0
* end
*/
public static final String EXPIRE_LUA = "if redis.call('EXISTS',KEYS[1]) == 1 and redis.call('GET',KEYS[1]) == ARGV[1] then redis.call('EXPIRE',KEYS[1], tonumber(ARGV[2])) return 1 else return 0 end";
/**
* 释放锁失败标识
*/
private static final long RELEASE_OK_FLAG = 0L;
@Autowired
private RedisUtil redisUtil;
/**
* 加锁方法
* @param lockTypeEnum 锁信息
* @param customKey 自定义锁定key
* @return true 成功，false 失败
*/
public String tryLock(LockTypeEnum lockTypeEnum, String customKey) {
// 锁对应值信息
String lockValue = IdUtil.simpleUUID();
// 对自定义key进行加锁操作,value值与key值相同
boolean result = redisUtil.setNxEx(lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey),
lockValue,
lockTypeEnum.getExpireTime().intValue());
if (result) {
log.info("[{}]加锁成功!", lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey));
return lockValue;
}
return null;
}
/**
* 进行加锁，加锁失败，再次进行加锁直到加锁成功
* @param lockTypeEnum 分布式锁类型设定enum
* @param customKey 自定义key
* @return
*/
public String tryContinueLock(LockTypeEnum lockTypeEnum, String customKey) {
// 锁对应值信息
String lockValue = IdUtil.simpleUUID();
while (true) {
// 对自定义key进行加锁操作,value值与key值相同
boolean result = redisUtil.setNxEx(lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey),
lockValue,
lockTypeEnum.getExpireTime().intValue());
try {
// 休眠1秒，防止StackOverFlow
Thread.sleep(1000);
} catch (Exception e) {
e.printStackTrace();
}
if (result) {
log.info("[{}]加锁成功!", lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey));
return lockValue;
}
}
}
/**
* 重入锁
* @param lockTypeEnum 锁定类型
* @param value 锁定值，一般为线程id或者uuid
* @param customKey 自定义key
* @return
*/
public boolean tryReentrantLock(LockTypeEnum lockTypeEnum, String value, String customKey) {
// 设置释放锁定key,value值
String lockKey = lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey);
// 设置重入锁脚本信息
DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>();
// Boolean 对应 lua脚本返回的0,1
defaultRedisScript.setResultType(Boolean.class);
// 设置重入锁脚本信息
defaultRedisScript.setScriptText(REENTRANT_LOCK_LUA);
// 进行重入锁执行
Boolean executeResult = redisUtil.executeBooleanLuaCustom(defaultRedisScript,
Collections.singletonList(lockKey),
value,
lockTypeEnum.getExpireTime().intValue());
if (executeResult) {
// 设置当前key为激活状态
ACTIVE_KEY_SET.add(lockKey);
// 设置定时任务，进行续期操作
resetLockExpire(lockTypeEnum, customKey, value, lockTypeEnum.getExpireTime());
}
return executeResult;
}
/**
* 进行续期操作
* @param lockTypeEnum 锁定类型
* @param customKey 自定义key
* @param value 锁定值，一般为线程id或者uuid
* @param expireTime 过期时间 单位秒，
*/
public void resetLockExpire(LockTypeEnum lockTypeEnum, String customKey, String value, long expireTime) {
// 续期的key信息
String resetKey = lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey);
// 校验当前key是否还在执行过程中
if (!ACTIVE_KEY_SET.contains(resetKey)) {
return;
}
// 时间设定延迟执行时间delay，默认续期时间是过期时间的1/3，在获取锁之后每expireTime/3时间进行一次续期操作
long delay = expireTime <= 3 ? 1 : expireTime / 3;
EXECUTOR_SERVICE.schedule(() -> {
log.info("自定义key[{}],对应值[{}]开始执行续期操作!", resetKey, value);
// 执行续期操作，如果续期成功则再次添加续期任务，如果续期成功，进行下一次定时任务续期
DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>();
// Boolean 对应 lua脚本返回的0,1
defaultRedisScript.setResultType(Boolean.class);
// 设置重入锁脚本信息
defaultRedisScript.setScriptText(EXPIRE_LUA);
// 进行重入锁执行
boolean executeLua = redisUtil.executeBooleanLuaCustom(defaultRedisScript,
Collections.singletonList(resetKey),
value,
lockTypeEnum.getExpireTime().intValue());
if (executeLua) {
log.info("执行key[{}],value[{}]续期成功，进行下一次续期操作", resetKey, value);
resetLockExpire(lockTypeEnum, customKey, value, expireTime);
} else {
// 续期失败处理，移除活跃key信息
ACTIVE_KEY_SET.remove(resetKey);
}
}, delay, TimeUnit.SECONDS);
}
/**
* 解锁操作
* @param lockTypeEnum 锁定类型
* @param customKey 自定义key
* @param releaseValue 释放value
* @return true 成功，false 失败
*/
public boolean releaseLock(LockTypeEnum lockTypeEnum, String customKey, String releaseValue) {
// 各个模块服务启动时间差，预留5秒等待时间，防止重调用
if (ObjectUtil.isNotNull(lockTypeEnum.getLockedWaitTimeMiles())) {
try {
Thread.sleep(lockTypeEnum.getLockedWaitTimeMiles());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 设置释放锁定key,value值
String releaseKey = lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey);
// 释放锁定资源
RedisScript<Long> longDefaultRedisScript = new DefaultRedisScript<>(UNLOCK_SCRIPT, Long.class);
Long result = redisUtil.executeLuaCustom(longDefaultRedisScript,
Collections.singletonList(releaseKey),
releaseValue);
// 根据返回结果判断是否成功成功匹配并删除 Redis 键值对，若果结果不为空和0，则验证通过
if (ObjectUtil.isNotNull(result) && result != RELEASE_OK_FLAG) {
// 当前key释放成功，从活跃生效keySet中移除
ACTIVE_KEY_SET.remove(releaseKey);
return true;
}
return false;
}
}
注意，LUA脚本执行过程中有时候会有执行失败情况，这些情况下异常信息很难捕捉，所以可以在LUA脚本中设置日志打印，但是需要注意，需要配置redis配置文件，打开日志信息，此处以重入锁为例子，具体配置以及脚本信息如下：
- 1.redis配置日志级别，日志存储位置信息
# 日志级别，可以设置为 debug、verbose、notice、warning，默认为 notice
loglevel notice
# 日志文件路径
logfile "/path/to/redis-server.log"
- 2.配置LUA脚本信息
local function log(level, message)
redis.log(level, "[DISTRIBUTED_LOCK]: " .. message)
end
if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then
log(redis.LOG_NOTICE, "Successfully acquired lock with key: " .. KEYS[1])
local expire_result = redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
if expire_result == 1 then
log(redis.LOG_NOTICE, "Set expiration of " .. ARGV[2] .. " seconds on lock.")
else
log(redis.LOG_WARNING, "Failed to set expiration on lock with key: " .. KEYS[1])
end
return 1
else
local current_value = redis.call('GET', KEYS[1])
if current_value == ARGV[1] then
log(redis.LOG_NOTICE, "Lock already held by this client; renewing expiration.")
local expire_result = redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
if expire_result == 1 then
log(redis.LOG_NOTICE, "Renewed expiration of " .. ARGV[2] .. " seconds on lock.")
else
log(redis.LOG_WARNING, "Failed to renew expiration on lock with key: " .. KEYS[1])
end
return 1
else
log(redis.LOG_DEBUG, "Lock is held by another client; not acquiring.")
return 0
end
end
2.3 锁枚举类实现
此处使用BASE_PRODUCT_TEST_LOCK作为测试的锁类型
package cn.git.common.lock;
import lombok.Getter;
/**
* 分布式锁类型设定enum
* @program: bank-credit-sy
* @author: lixuchun
* @create: 2022-04-25
*/
@Getter
public enum LockTypeEnum {
/**
* 分布式锁类型详情
*/
DISTRIBUTE_TASK_LOCK("DISTRIBUTE_TASK_LOCK", 120L, "xxlJob初始化分布式锁", 5000L),
CACHE_INIT_LOCK("CACHE_INIT_LOCK", 120L, "缓存平台初始化缓存信息分布式锁", 5000L),
RULE_INIT_LOCK("RULE_INIT_LOCK", 120L, "规则引擎规则加载初始化", 5000L),
SEQUENCE_LOCK("SEQUENCE_LOCK", 120L, "序列信息月末初始化!", 5000L),
UAA_ONLINE_NUMBER_LOCK("UAA_ONLINE_LOCK", 20L, "登录模块刷新在线人数", 5000L),
BASE_SERVER_IDEMPOTENCE("BASE_IDEMPOTENCE_LOCK", 15L, "基础业务幂等性校验"),
WORK_FLOW_WEB_SERVICE_LOCK("WORK_FLOW_WEB_SERVICE_LOCK", 15L, "流程webService服务可用ip地址获取锁", 5000L),
BASE_PRODUCT_TEST_LOCK("BASE_PRODUCT_TEST_LOCK", 10L, "产品测试分布式锁", null),
;
/**
* 锁类型
*/
private String lockType;
/**
* 即过期时间,单位为second
*/
private Long expireTime;
/**
* 枷锁成功后，默认等待时间，时间应小于过期时间，单位毫秒
*/
private Long lockedWaitTimeMiles;
/**
* 描述信息
*/
private String lockDesc;
/**
* 构造方法
* @param lockType 类型
* @param lockTime 锁定时间
* @param lockDesc 锁描述
*/
LockTypeEnum(String lockType, Long lockTime, String lockDesc) {
this.lockDesc = lockDesc;
this.expireTime = lockTime;
this.lockType = lockType;
}
/**
* 构造方法
* @param lockType 类型
* @param lockTime 锁定时间
* @param lockDesc 锁描述
* @param lockedWaitTimeMiles 锁失效时间
*/
LockTypeEnum(String lockType, Long lockTime, String lockDesc, Long lockedWaitTimeMiles) {
this.lockDesc = lockDesc;
this.expireTime = lockTime;
this.lockType = lockType;
this.lockedWaitTimeMiles = lockedWaitTimeMiles;
}
}
3. 测试
测试分为两部分，模拟多线程清库存产品，10个产品，1000个线程进行争夺，具体实现如下
3.1 测试代码部分
package cn.git.foreign;
import cn.git.api.client.EsbCommonClient;
import cn.git.api.dto.P043001009DTO;
import cn.git.common.lock.LockTypeEnum;
import cn.git.common.lock.SimpleDistributeLock;
import cn.git.foreign.dto.QueryCreditDTO;
import cn.git.foreign.manage.ForeignCreditCheckApiImpl;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* @description: 个人/法人 征信查询
* @program: bank-credit-sy
* @author: lixuchun
* @create: 2022-07-12 09:03:23
*/
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ForeignApplication.class)
public class DistributionLockTest {
@Autowired
private SimpleDistributeLock distributeLock;
/**
* 产品信息
*/
private Product product = new Product("0001", 10, 0, "iphone");
/**
* @description: 产品信息
* @program: bank-credit-sy
* @author: lixuchun
* @create: 2024-04-03
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public static class Product {
/**
* id
*/
private String id;
/**
* 库存
*/
private Integer stock;
/**
* 已售
*/
private Integer sold;
/**
* 名称
*/
private String name;
}
/**
* 释放锁
*/
@Test
public void releaseLock() {
distributeLock.releaseLock(LockTypeEnum.BASE_PRODUCT_TEST_LOCK, "0001", "xxxx");
}
/**
* 分布式锁模拟测试
*/
@Test
public void testLock() throws InterruptedException {
// 20核心线程，最大线程也是100，非核心线程空闲等待时间10秒，队列最大1000
ThreadPoolExecutor executor = new ThreadPoolExecutor(100,
100,
10,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10000));
// 模拟1000个请求
CountDownLatch countDownLatch = new CountDownLatch(1000);
// 模拟10000个人抢10个商品
for (int i = 0; i < 1000; i++) {
executor.execute(() -> {
// 加锁
// soldByLock();
// 不加锁扣减库存
normalSold();
countDownLatch.countDown();
});
}
countDownLatch.await();
executor.shutdown();
// 输出产品信息
System.out.println(JSONObject.toJSONString(product));
}
/**
* 加锁减库存
*/
public void soldByLock() {
// 设置加锁value信息
String lockValue = IdUtil.simpleUUID();
try {
boolean isLocked = distributeLock.tryReentrantLock(LockTypeEnum.BASE_PRODUCT_TEST_LOCK, lockValue, product.getId());
if (isLocked) {
// 加锁成功，开始减库存信息
if (product.getStock() > 0) {
product.setStock(product.getStock() - 1);
product.setSold(product.getSold() + 1);
System.out.println(StrUtil.format("减库存成功，剩余库存[{}]", product.getStock()));
} else {
System.out.println("库存不足");
}
}
// 暂停1000毫秒，模拟业务处理
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
distributeLock.releaseLock(LockTypeEnum.BASE_PRODUCT_TEST_LOCK, product.getId(), lockValue);
}
}
/**
* 不加锁减库存
*/
public void normalSold() {
// 获取线程id
long id = Thread.currentThread().getId();
// 暂停1000毫秒，模拟业务处理
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
// 开始库存计算
if (product.getStock() > 0) {
product.setStock(product.getStock() - 1);
product.setSold(product.getSold() + 1);
System.out.println(StrUtil.format("线程[{}]减库存成功，剩余库存[{}]", id, product.getStock()));
} else {
System.out.println("库存不足");
}
}
}
3.2 无锁库存处理情况
无锁情况下，发生产品超发情况，卖出11个产品，具体如下图
3.3 加锁处理情况
多次实验，没有发生产品超发情况，具体测试结果如下：
