redis分布式鎖實現(xiàn)示例
更新時間：2024年11月06日 11:55:37 作者：小大力
本文主要介紹了redis分布式鎖實現(xiàn)示例,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
1.需求
我們公司想實現(xiàn)一個簡單的分布式鎖，用于服務啟動初始化執(zhí)行init方法的時候，只執(zhí)行一次，避免重復執(zhí)行加載緩存規(guī)則的代碼，還有預防高并發(fā)流程發(fā)起部分，產(chǎn)品超發(fā)，多發(fā)問題。所以結(jié)合網(wǎng)上信息，自己簡單實現(xiàn)了一個redis分布式鎖，可以進行單次資源鎖定，排隊鎖定(沒有實現(xiàn)權(quán)重，按照時間長短爭奪鎖信息)，還有鎖定業(yè)務未完成，需要延期鎖等簡單方法，死鎖則是設(shè)置過期時間即可。期間主要用到的技術(shù)為redis，延時線程池，LUA腳本，比較簡單，此處記錄一下，方便下次學習查看。
2.具體實現(xiàn)
整體配置相對簡單，主要是編寫redisUtil工具類，實現(xiàn)redis的簡單操作，編寫分布式鎖類SimpleDistributeLock，主要內(nèi)容都在此鎖的實現(xiàn)類中，SimpleDistributeLock實現(xiàn)類主要實現(xiàn)方法如下：
- 1.一次搶奪加鎖方法 tryLock
- 2.連續(xù)排隊加鎖方法tryContinueLock，此方法中間有調(diào)用線程等待Thread.sleep方法防止防止StackOverFlow異常，比較耗費資源，后續(xù)應該需要優(yōu)化處理
- 3.重入鎖tryReentrantLock，一個資源調(diào)用過程中，處于加鎖狀態(tài)仍然可以再次加鎖，重新刷新其過期時間
- 4.刷新鎖過期時間方法resetLockExpire
- 5.釋放鎖方法，注意，釋放過程中需要傳入加鎖的value信息，以免高并發(fā)情況下多線程鎖信息被其他線程釋放鎖操作誤刪
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;
}
}
/**
* 根據(jù)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;
}
}
/**
* 普通緩存放入并設(shè)置時間
*
* @param key 鍵
* @param value 值
* @param time 時間(秒) time要大于0 如果time小于等于0 將設(shè)置無限期
* @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;
}
}
/**
* 如果不存在，則設(shè)置對應key,value 鍵值對,并且設(shè)置過期時間
* @param key 鎖key
* @param value 鎖值
* @param time 時間單位second
* @return 設(shè)定結(jié)果
*/
/**
public Boolean setNxEx(String key, String value, long time) {
Boolean setResult = (Boolean) redisTemplate.execute((RedisCallback) connection -> {
RedisStringCommands.SetOption setOption = RedisStringCommands.SetOption.ifAbsent();
// 設(shè)置過期時間
Expiration expiration = Expiration.seconds(time);
// 執(zhí)行setnx操作
Boolean result = connection.set(key.getBytes(StandardCharsets.UTF_8),
value.getBytes(StandardCharsets.UTF_8), expiration, setOption);
return result;
});
return setResult;
}
**/
/**
* 如果不存在，則設(shè)置對應key,value 鍵值對,并且設(shè)置過期時間
* @param key 鎖key
* @param value 鎖值
* @param time 時間單位second
* @return 設(shè)定結(jié)果
*/
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 并設(shè)置時間
*
* @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表中放入數(shù)據(jù),如果不存在將創(chuàng)建
*
* @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表中放入數(shù)據(jù),如果不存在將創(chuàng)建
*
* @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遞增 如果不存在,就會創(chuàng)建一個 并把新增后的值返回
*
* @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);
}
/**
* 根據(jù)key獲取Set中的所有值
*
* @param key 鍵
* @return
*/
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 根據(jù)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;
}
}
/**
* 將數(shù)據(jù)放入set緩存
*
* @param key 鍵
* @param values 值 可以是多個
* @return 成功個數(shù)
*/
public long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 將set數(shù)據(jù)放入緩存
*
* @param key 鍵
* @param time 時間(秒)
* @param values 值 可以是多個
* @return 成功個數(shù)
*/
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 移除的個數(shù)
*/
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緩存的內(nèi)容
*
* @param key 鍵
* @param start 開始
* @param end 結(jié)束 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倒數(shù)第二個元素，依次類推
* @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;
}
}
/**
* 根據(jù)索引修改list中的某條數(shù)據(jù)
*
* @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 移除的個數(shù)
*/
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: 執(zhí)行l(wèi)ua腳本，只對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: 執(zhí)行l(wèi)ua腳本，只對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: 執(zhí)行l(wèi)ua腳本，只對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) {
// 參數(shù)存在校驗
if (ObjectUtil.isNotNull(redisTemplate.opsForZSet().score(key, value))) {
redisTemplate.opsForZSet().remove(key, value);
}
}
/**
* 通過前綴key值獲取所有key內(nèi)容(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實現(xiàn)
具體鎖以及解鎖業(yè)務實現(xiàn)類，具體如下所示
package cn.git.common.lock;
import cn.git.common.exception.ServiceException;
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;
/**
* 簡單分布式鎖
* 可以實現(xiàn)鎖的重入，鎖自動延期
* @program: bank-credit-sy
* @author: lixuchun
* @create: 2022-04-25
*/
@Slf4j
@Component
public class SimpleDistributeLock {
/**
* 活躍的鎖集合
*/
private volatile static CopyOnWriteArraySet ACTIVE_KEY_SET = new CopyOnWriteArraySet();
/**
* 定時線程池，續(xù)期使用
*/
private static ScheduledExecutorService EXECUTOR_SERVICE = Executors.newScheduledThreadPool(5);
/**
* 解鎖腳本, 腳本參數(shù) 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腳本參數(shù)介紹 KEYS[1]：傳入的key ARGV[1]：傳入的value ARGV[2]：傳入的過期時間
* // 如果成功設(shè)置keys，value值，然后設(shè)定過期時間，直接返回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";
/**
* 續(xù)期腳本
* // 如果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;
/**
* 最大重試時間間隔,單位毫秒
*/
private static final int MAX_RETRY_DELAY_MS = 2000;
@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 分布式鎖類型設(shè)定enum
* @param customKey 自定義key
* @return
*/
public String tryContinueLock(LockTypeEnum lockTypeEnum, String customKey) {
// 鎖對應值信息
String lockValue = IdUtil.simpleUUID();
// 設(shè)置最大重試次數(shù)
int maxRetries = 10;
// 初始重試間隔，可調(diào)整
int retryIntervalMs = 100;
for (int attempt = 1; attempt <= maxRetries; attempt++) {
// 對自定義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;
}
/**
* 如果未能獲取鎖，計算下一次重試間隔（可使用指數(shù)退避策略）, MAX_RETRY_DELAY_MS 為最大重試間隔
* 這行代碼用于計算下一次重試前的等待間隔（delay）。這里采用了指數(shù)退避策略，這是一種常用的重試間隔設(shè)計方法，旨在隨著重試次數(shù)的增加逐步增大等待間隔，同時限制其增長上限。
* 1. (1 << (attempt - 1))：這是一個二進制左移運算，相當于將 1 左移 attempt - 1 位。對于整數(shù) attempt，該表達式的結(jié)果等于 2^(attempt - 1)。隨著 attempt 增加，結(jié)果值按指數(shù)級增長（1, 2, 4, 8, ...），符合指數(shù)退避策略的要求。
* 2. * retryIntervalMs：將上述結(jié)果乘以基礎(chǔ)重試間隔 retryIntervalMs，得到實際的等待時間（單位為毫秒）。
* 3. Math.min(..., MAX_RETRY_DELAY_MS)：確保計算出的 delay 值不超過預設(shè)的最大重試間隔 MAX_RETRY_DELAY_MS。這樣做可以防止在極端情況下因等待時間過長而導致系統(tǒng)響應緩慢或其他問題。
*/
int delay = Math.min((1 << (attempt - 1)) * retryIntervalMs, MAX_RETRY_DELAY_MS);
/**
* 使用 try-catch 塊包裹線程休眠操作，以處理可能拋出的 InterruptedException 異常。
* 1. Thread.sleep(delay)：讓當前線程進入休眠狀態(tài)，暫停執(zhí)行指定的 delay 時間（之前計算得出的重試間隔）。在此期間，線程不會消耗 CPU 資源，有助于減輕系統(tǒng)壓力。
* 2. catch (InterruptedException e)：捕獲在休眠過程中被中斷時拋出的 InterruptedException。線程中斷通常用于請求線程提前結(jié)束其當前任務或進入某個特定狀態(tài)。
* 3. Thread.currentThread().interrupt();：當捕獲到 InterruptedException 時，恢復線程的中斷狀態(tài)。這是因為在處理中斷時，Thread.sleep() 方法會清除中斷狀態(tài)。通過重新設(shè)置中斷狀態(tài)，通知后續(xù)代碼（如其他 catch 子句或 finally 子句）或外部代碼當前線程已被中斷。
* 4. throw new RuntimeException(e);：將捕獲到的 InterruptedException 包裝成一個新的 RuntimeException 并拋出。這樣做是為了向上層代碼傳遞中斷信號，并保留原始異常堆棧信息以供調(diào)試。根據(jù)具體應用需求，可以選擇拋出自定義InterruptedException`。
*/
try {
Thread.sleep(delay);
} catch (InterruptedException e) {
// 保持中斷狀態(tài)
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
}
throw new ServiceException("Failed to acquire lock after " + maxRetries + " attempts");
}
/**
* 重入鎖
* @param lockTypeEnum 鎖定類型
* @param value 鎖定值，一般為線程id或者uuid
* @param customKey 自定義key
* @return
*/
public boolean tryReentrantLock(LockTypeEnum lockTypeEnum, String value, String customKey) {
// 設(shè)置釋放鎖定key,value值
String lockKey = lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey);
// 設(shè)置重入鎖腳本信息
DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>();
// Boolean 對應 lua腳本返回的0,1
defaultRedisScript.setResultType(Boolean.class);
// 設(shè)置重入鎖腳本信息
defaultRedisScript.setScriptText(REENTRANT_LOCK_LUA);
// 進行重入鎖執(zhí)行
Boolean executeResult = redisUtil.executeBooleanLuaCustom(defaultRedisScript,
Collections.singletonList(lockKey),
value,
lockTypeEnum.getExpireTime().intValue());
if (executeResult) {
// 設(shè)置當前key為激活狀態(tài)
ACTIVE_KEY_SET.add(lockKey);
// 設(shè)置定時任務，進行續(xù)期操作
resetLockExpire(lockTypeEnum, customKey, value, lockTypeEnum.getExpireTime());
}
return executeResult;
}
/**
* 進行續(xù)期操作
* @param lockTypeEnum 鎖定類型
* @param customKey 自定義key
* @param value 鎖定值，一般為線程id或者uuid
* @param expireTime 過期時間 單位秒，
*/
public void resetLockExpire(LockTypeEnum lockTypeEnum, String customKey, String value, long expireTime) {
// 續(xù)期的key信息
String resetKey = lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey);
// 校驗當前key是否還在執(zhí)行過程中
if (!ACTIVE_KEY_SET.contains(resetKey)) {
return;
}
// 時間設(shè)定延遲執(zhí)行時間delay，默認續(xù)期時間是過期時間的1/3，在獲取鎖之后每expireTime/3時間進行一次續(xù)期操作
long delay = expireTime <= 3 ? 1 : expireTime / 3;
EXECUTOR_SERVICE.schedule(() -> {
log.info("自定義key[{}],對應值[{}]開始執(zhí)行續(xù)期操作!", resetKey, value);
// 執(zhí)行續(xù)期操作，如果續(xù)期成功則再次添加續(xù)期任務，如果續(xù)期成功，進行下一次定時任務續(xù)期
DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>();
// Boolean 對應 lua腳本返回的0,1
defaultRedisScript.setResultType(Boolean.class);
// 設(shè)置重入鎖腳本信息
defaultRedisScript.setScriptText(EXPIRE_LUA);
// 進行重入鎖執(zhí)行
boolean executeLua = redisUtil.executeBooleanLuaCustom(defaultRedisScript,
Collections.singletonList(resetKey),
value,
lockTypeEnum.getExpireTime().intValue());
if (executeLua) {
log.info("執(zhí)行key[{}],value[{}]續(xù)期成功，進行下一次續(xù)期操作", resetKey, value);
resetLockExpire(lockTypeEnum, customKey, value, expireTime);
} else {
// 續(xù)期失敗處理，移除活躍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秒等待時間，防止重調(diào)用
if (ObjectUtil.isNotNull(lockTypeEnum.getLockedWaitTimeMiles())) {
try {
Thread.sleep(lockTypeEnum.getLockedWaitTimeMiles());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 設(shè)置釋放鎖定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);
// 根據(jù)返回結(jié)果判斷是否成功成功匹配并刪除 Redis 鍵值對，若果結(jié)果不為空和0，則驗證通過
if (ObjectUtil.isNotNull(result) && result != RELEASE_OK_FLAG) {
// 當前key釋放成功，從活躍生效keySet中移除
ACTIVE_KEY_SET.remove(releaseKey);
return true;
}
return false;
}
}
注意，LUA腳本執(zhí)行過程中有時候會有執(zhí)行失敗情況，這些情況下異常信息很難捕捉，所以可以在LUA腳本中設(shè)置日志打印，但是需要注意，需要配置redis配置文件，打開日志信息，此處以重入鎖為例子，具體配置以及腳本信息如下：
- 1.redis配置日志級別，日志存儲位置信息
# 日志級別，可以設(shè)置為 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 鎖枚舉類實現(xiàn)
此處使用BASE_PRODUCT_TEST_LOCK作為測試的鎖類型
package cn.git.common.lock;
import lombok.Getter;
/**
* 分布式鎖類型設(shè)定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, "規(guī)則引擎規(guī)則加載初始化", 5000L),
SEQUENCE_LOCK("SEQUENCE_LOCK", 120L, "序列信息月末初始化!", 5000L),
UAA_ONLINE_NUMBER_LOCK("UAA_ONLINE_LOCK", 20L, "登錄模塊刷新在線人數(shù)", 5000L),
BASE_SERVER_IDEMPOTENCE("BASE_IDEMPOTENCE_LOCK", 15L, "基礎(chǔ)業(yè)務冪等性校驗"),
WORK_FLOW_WEB_SERVICE_LOCK("WORK_FLOW_WEB_SERVICE_LOCK", 15L, "流程webService服務可用ip地址獲取鎖", 5000L),
BASE_PRODUCT_TEST_LOCK("BASE_PRODUCT_TEST_LOCK", 10L, "產(chǎn)品測試分布式鎖", null),
;
/**
* 鎖類型
*/
private String lockType;
/**
* 即過期時間,單位為second
*/
private Long expireTime;
/**
* 枷鎖成功后，默認等待時間，時間應小于過期時間，單位毫秒
*/
private Long lockedWaitTimeMiles;
/**
* 描述信息
*/
private String lockDesc;
/**
* 構(gòu)造方法
* @param lockType 類型
* @param lockTime 鎖定時間
* @param lockDesc 鎖描述
*/
LockTypeEnum(String lockType, Long lockTime, String lockDesc) {
this.lockDesc = lockDesc;
this.expireTime = lockTime;
this.lockType = lockType;
}
/**
* 構(gòu)造方法
* @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. 測試
測試分為兩部分，模擬多線程清庫存產(chǎn)品，10個產(chǎn)品，1000個線程進行爭奪，具體實現(xiàn)如下
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: 2024-04-07 08:03:23
*/
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = ForeignApplication.class)
public class DistributionLockTest {
@Autowired
private SimpleDistributeLock distributeLock;
/**
* 產(chǎn)品信息
*/
private Product product = new Product("0001", 10, 0, "iphone");
/**
* @description: 產(chǎn)品信息
* @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();
// 輸出產(chǎn)品信息
System.out.println(JSONObject.toJSONString(product));
}
/**
* 加鎖減庫存
*/
public void soldByLock() {
// 設(shè)置加鎖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毫秒，模擬業(yè)務處理
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毫秒，模擬業(yè)務處理
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 無鎖庫存處理情況
無鎖情況下，發(fā)生產(chǎn)品超發(fā)情況，賣出11個產(chǎn)品，具體如下圖
3.3 加鎖處理情況
多次實驗，沒有發(fā)生產(chǎn)品超發(fā)情況，具體測試結(jié)果如下：
4. 其他實現(xiàn)
還可以使用Redisson客戶端進行分布式鎖實現(xiàn)，這樣更加簡單安全，其有自己的看門狗機制，續(xù)期加鎖解鎖都更加方便，簡單操作過程實例代碼如下
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import java.util.concurrent.TimeUnit;
/**
* @description: Redisson客戶端實現(xiàn)
* @program: bank-credit-sy
* @author: lixuchun
* @create: 2022-07-12 09:03:23
*/
public class RedissonLockExample {
public static void main(String[] args) {
// 配置Redisson客戶端
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6379");
RedissonClient redisson = Redisson.create(config);
// 獲取鎖對象
RLock lock = redisson.getLock("myLock");
try {
// 嘗試獲取鎖，最多等待100秒，鎖定之后10秒自動釋放
// 鎖定之后會自動續(xù)期10秒
if (lock.tryLock(100, 10, TimeUnit.SECONDS)) {
try {
// 處理業(yè)務邏輯
} finally {
// 釋放鎖
lock.unlock();
}
}
} catch (InterruptedException e) {
// 處理中斷異常
Thread.currentThread().interrupt();
} finally {
// 關(guān)閉Redisson客戶端
redisson.shutdown();
}
}
}
其中鎖定api是否開啟看門狗，整理如下
// 開始拿鎖，失敗阻塞重試
RLock lock = redissonClient.getLock("guodong");
// 具有Watch Dog 自動延期機制 默認續(xù)30s 每隔30/3=10 秒續(xù)到30s
lock.lock();
// 嘗試拿鎖10s后停止重試,返回false 具有Watch Dog 自動延期機制 默認續(xù)30s
boolean res1 = lock.tryLock(10, TimeUnit.SECONDS);
// 嘗試拿鎖10s后，沒有Watch Dog
lock.lock(10, TimeUnit.SECONDS);
// 沒有Watch Dog ，10s后自動釋放
lock.lock(10, TimeUnit.SECONDS);
// 嘗試拿鎖100s后停止重試,返回false 沒有Watch Dog ，10s后自動釋放
boolean res2 = lock.tryLock(100, 10, TimeUnit.SECONDS);到此這篇關(guān)于redis分布式鎖實現(xiàn)示例的文章就介紹到這了,更多相關(guān)Redis分布式鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Redis list 類型學習筆記與總結(jié),本文著重講解了關(guān)于List的一些常用方法,比如lpush 方法、lrange 方法、rpush 方法、linsert 方法、 lset 方法等,需要的朋友可以參考下2015-06-06
這篇文章主要介紹了linux?redis-連接命令解讀，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-06-06
redis.conf中使用requirepass不生效的原因及解決方法
本文主要介紹了如何啟用requirepass，以及啟用requirepass為什么不會生效，從代碼層面分析了不生效的原因，以及解決方法,需要的朋友可以參考下2023-07-07
redis是一款基于BSD協(xié)議,開源的非關(guān)系型數(shù)據(jù)庫(nosql數(shù)據(jù)庫)這篇文章主要給大家介紹了關(guān)于K8S部署Redis(單機、集群)的超詳細步驟,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2024-05-05
NoSQL和Redis簡介及Redis在Windows下的安裝和使用教程
這篇文章主要介紹了NoSQL和Redis簡介及Redis在Windows下的安裝和使用教程,本文同時講解了python操作redis,并給出了操作實例,需要的朋友可以參考下2015-01-01
Redis連接命令主要是用于客戶端與服務器建立連接的,Redis是一種流行的內(nèi)存數(shù)據(jù)庫,支持多種數(shù)據(jù)結(jié)構(gòu),其中鍵命令是核心操作之一,在Redis中,鍵（Key）是用來存儲數(shù)據(jù)的主要元素,每個鍵都有一個唯一的名稱,本文給大家介紹了Redis中的連接命令與鍵命令操作2024-09-09
Geospatial是地理位置類型,我們可以用來查詢附近的人、計算兩人之間的距離等,這篇文章主要介紹了Redis中三種特殊數(shù)據(jù)類型命令詳解,需要的朋友可以參考下2024-05-05
Redis是一種高級key-value數(shù)據(jù)庫。它跟memcached類似，不過數(shù)據(jù)可以持久化，而且支持的數(shù)據(jù)類型很豐富。今天我們就來看看如何實現(xiàn)Redis持久化,需要的朋友可以參考下2022-10-10
這篇文章主要介紹了基于redis+lua進行限流,通過實例代碼詳細介紹了lua+redis進行限流的做法，開發(fā)環(huán)境使用idea+redis+lua，本文給大家介紹的非常詳細，需要的朋友可以參考下2022-07-07
在高并發(fā)系統(tǒng)中,Redis作為核心緩存組件,通常扮演著重要的"守門員"角色,當大量緩存同時失效時,會導致請求如洪水般直接涌向數(shù)據(jù)庫,造成數(shù)據(jù)庫瞬間壓力劇增甚至宕機,這種現(xiàn)象被形象地稱為"緩存雪崩",本文給大家介紹了Redis緩存雪崩的5種應對措施,需要的朋友可以參考下2025-04-04