redis分布式鎖實(shí)現(xiàn)示例
1.需求
我們公司想實(shí)現(xiàn)一個(gè)簡單的分布式鎖,用于服務(wù)啟動(dòng)初始化執(zhí)行init方法的時(shí)候,只執(zhí)行一次,避免重復(fù)執(zhí)行加載緩存規(guī)則的代碼,還有預(yù)防高并發(fā)流程發(fā)起部分,產(chǎn)品超發(fā),多發(fā)問題。所以結(jié)合網(wǎng)上信息,自己簡單實(shí)現(xiàn)了一個(gè)redis分布式鎖,可以進(jìn)行單次資源鎖定,排隊(duì)鎖定(沒有實(shí)現(xiàn)權(quán)重,按照時(shí)間長短爭奪鎖信息),還有鎖定業(yè)務(wù)未完成,需要延期鎖等簡單方法,死鎖則是設(shè)置過期時(shí)間即可。期間主要用到的技術(shù)為redis,延時(shí)線程池,LUA腳本,比較簡單,此處記錄一下,方便下次學(xué)習(xí)查看。
2.具體實(shí)現(xiàn)
整體配置相對(duì)簡單,主要是編寫redisUtil工具類,實(shí)現(xiàn)redis的簡單操作,編寫分布式鎖類SimpleDistributeLock,主要內(nèi)容都在此鎖的實(shí)現(xiàn)類中,SimpleDistributeLock實(shí)現(xiàn)類主要實(shí)現(xiàn)方法如下:
- 1.一次搶奪加鎖方法 tryLock
- 2.連續(xù)排隊(duì)加鎖方法tryContinueLock,此方法中間有調(diào)用線程等待Thread.sleep方法防止防止StackOverFlow異常,比較耗費(fèi)資源,后續(xù)應(yīng)該需要優(yōu)化處理
- 3.重入鎖tryReentrantLock,一個(gè)資源調(diào)用過程中,處于加鎖狀態(tài)仍然可以再次加鎖,重新刷新其過期時(shí)間
- 4.刷新鎖過期時(shí)間方法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; } /** * 指定緩存失效時(shí)間 * * @param key 鍵 * @param time 時(shí)間(秒) * @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 獲取過期時(shí)間 * * @param key 鍵 不能為null * @return 時(shí)間(秒) 返回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 可以傳一個(gè)值 或多個(gè) */ @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è)置時(shí)間 * * @param key 鍵 * @param value 值 * @param time 時(shí)間(秒) 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è)置對(duì)應(yīng)key,value 鍵值對(duì),并且設(shè)置過期時(shí)間 * @param key 鎖key * @param value 鎖值 * @param time 時(shí)間單位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è)置過期時(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è)置對(duì)應(yīng)key,value 鍵值對(duì),并且設(shè)置過期時(shí)間 * @param key 鎖key * @param value 鎖值 * @param time 時(shí)間單位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 項(xiàng) 不能為null * @return 值 */ public Object hget(String key, String item) { return redisTemplate.opsForHash().get(key, item); } /** * 獲取hashKey對(duì)應(yīng)的所有鍵值 * * @param key 鍵 * @return 對(duì)應(yīng)的多個(gè)鍵值 */ public Map<Object, Object> hmget(String key) { return redisTemplate.opsForHash().entries(key); } /** * 獲取hashKey對(duì)應(yīng)的所有鍵值 * * @param key 鍵 * @return 對(duì)應(yīng)的多個(gè)鍵值 */ public List<Object> hmget(String key, List<Object> itemList) { return redisTemplate.opsForHash().multiGet(key, itemList); } /** * 獲取key對(duì)應(yīng)的hashKey值 * * @param key 鍵 * @param hashKey 鍵 * @return 對(duì)應(yīng)的鍵值 */ public Object hmget(String key, String hashKey) { return redisTemplate.opsForHash().get(key, hashKey); } /** * HashSet * * @param key 鍵 * @param map 對(duì)應(yīng)多個(gè)鍵值 * @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è)置時(shí)間 * * @param key 鍵 * @param map 對(duì)應(yīng)多個(gè)鍵值 * @param time 時(shí)間(秒) * @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 項(xiàng) * @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 項(xiàng) * @param value 值 * @param time 時(shí)間(秒) 注意:如果已存在的hash表有時(shí)間,這里將會(huì)替換原有的時(shí)間 * @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 項(xiàng) 可以使多個(gè) 不能為null */ public void hdel(String key, Object... item) { redisTemplate.opsForHash().delete(key, item); } /** * 判斷hash表中是否有該項(xiàng)的值 * * @param key 鍵 不能為null * @param item 項(xiàng) 不能為null * @return true 存在 false不存在 */ public boolean hHasKey(String key, String item) { return redisTemplate.opsForHash().hasKey(key, item); } /** * hash遞增 如果不存在,就會(huì)創(chuàng)建一個(gè) 并把新增后的值返回 * * @param key 鍵 * @param item 項(xiàng) * @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 項(xiàng) * @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從一個(gè)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 值 可以是多個(gè) * @return 成功個(gè)數(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 時(shí)間(秒) * @param values 值 可以是多個(gè) * @return 成功個(gè)數(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 值 可以是多個(gè) * @return 移除的個(gè)數(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時(shí), 0 表頭,1 第二個(gè)元素,依次類推;index<0時(shí),-1,表尾,-2倒數(shù)第二個(gè)元素,依次類推 * @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 時(shí)間(秒) * @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 時(shí)間(秒) * @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個(gè)值為value * * @param key 鍵 * @param count 移除多少個(gè) * @param value 值 * @return 移除的個(gè)數(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腳本,只對(duì)key進(jìn)行操作 * @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腳本,只對(duì)key進(jìn)行操作 * @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腳本,只對(duì)key進(jìn)行操作 * @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); } /** * 時(shí)間窗口限流 * @param key key * @param timeWindow 時(shí)間窗口 * @return */ public Integer rangeByScore(String key, Integer timeWindow) { // 獲取當(dāng)前時(shí)間戳 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ù)存在校驗(yàn) if (ObjectUtil.isNotNull(redisTemplate.opsForZSet().score(key, value))) { redisTemplate.opsForZSet().remove(key, value); } } /** * 通過前綴key值獲取所有key內(nèi)容(hash) * @param keyPrefix 前綴key * @param fieldArray 查詢對(duì)象列信息 */ 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實(shí)現(xiàn)
具體鎖以及解鎖業(yè)務(wù)實(shí)現(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; /** * 簡單分布式鎖 * 可以實(shí)現(xiàn)鎖的重入,鎖自動(dòng)延期 * @program: bank-credit-sy * @author: lixuchun * @create: 2022-04-25 */ @Slf4j @Component public class SimpleDistributeLock { /** * 活躍的鎖集合 */ private volatile static CopyOnWriteArraySet ACTIVE_KEY_SET = new CopyOnWriteArraySet(); /** * 定時(shí)線程池,續(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í)間 * // 如果成功設(shè)置keys,value值,然后設(shè)定過期時(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值相等,則重置過期時(shí)間,直接返回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值相等,則重置過期時(shí)間,直接返回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"; /** * 釋放鎖失敗標(biāo)識(shí) */ private static final long RELEASE_OK_FLAG = 0L; /** * 最大重試時(shí)間間隔,單位毫秒 */ 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) { // 鎖對(duì)應(yīng)值信息 String lockValue = IdUtil.simpleUUID(); // 對(duì)自定義key進(jìn)行加鎖操作,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; } /** * 進(jìn)行加鎖,加鎖失敗,再次進(jìn)行加鎖直到加鎖成功 * @param lockTypeEnum 分布式鎖類型設(shè)定enum * @param customKey 自定義key * @return */ public String tryContinueLock(LockTypeEnum lockTypeEnum, String customKey) { // 鎖對(duì)應(yīng)值信息 String lockValue = IdUtil.simpleUUID(); // 設(shè)置最大重試次數(shù) int maxRetries = 10; // 初始重試間隔,可調(diào)整 int retryIntervalMs = 100; for (int attempt = 1; attempt <= maxRetries; attempt++) { // 對(duì)自定義key進(jìn)行加鎖操作,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; } /** * 如果未能獲取鎖,計(jì)算下一次重試間隔(可使用指數(shù)退避策略), MAX_RETRY_DELAY_MS 為最大重試間隔 * 這行代碼用于計(jì)算下一次重試前的等待間隔(delay)。這里采用了指數(shù)退避策略,這是一種常用的重試間隔設(shè)計(jì)方法,旨在隨著重試次數(shù)的增加逐步增大等待間隔,同時(shí)限制其增長上限。 * 1. (1 << (attempt - 1)):這是一個(gè)二進(jìn)制左移運(yùn)算,相當(dāng)于將 1 左移 attempt - 1 位。對(duì)于整數(shù) attempt,該表達(dá)式的結(jié)果等于 2^(attempt - 1)。隨著 attempt 增加,結(jié)果值按指數(shù)級(jí)增長(1, 2, 4, 8, ...),符合指數(shù)退避策略的要求。 * 2. * retryIntervalMs:將上述結(jié)果乘以基礎(chǔ)重試間隔 retryIntervalMs,得到實(shí)際的等待時(shí)間(單位為毫秒)。 * 3. Math.min(..., MAX_RETRY_DELAY_MS):確保計(jì)算出的 delay 值不超過預(yù)設(shè)的最大重試間隔 MAX_RETRY_DELAY_MS。這樣做可以防止在極端情況下因等待時(shí)間過長而導(dǎo)致系統(tǒng)響應(yīng)緩慢或其他問題。 */ int delay = Math.min((1 << (attempt - 1)) * retryIntervalMs, MAX_RETRY_DELAY_MS); /** * 使用 try-catch 塊包裹線程休眠操作,以處理可能拋出的 InterruptedException 異常。 * 1. Thread.sleep(delay):讓當(dāng)前線程進(jìn)入休眠狀態(tài),暫停執(zhí)行指定的 delay 時(shí)間(之前計(jì)算得出的重試間隔)。在此期間,線程不會(huì)消耗 CPU 資源,有助于減輕系統(tǒng)壓力。 * 2. catch (InterruptedException e):捕獲在休眠過程中被中斷時(shí)拋出的 InterruptedException。線程中斷通常用于請(qǐng)求線程提前結(jié)束其當(dāng)前任務(wù)或進(jìn)入某個(gè)特定狀態(tài)。 * 3. Thread.currentThread().interrupt();:當(dāng)捕獲到 InterruptedException 時(shí),恢復(fù)線程的中斷狀態(tài)。這是因?yàn)樵谔幚碇袛鄷r(shí),Thread.sleep() 方法會(huì)清除中斷狀態(tài)。通過重新設(shè)置中斷狀態(tài),通知后續(xù)代碼(如其他 catch 子句或 finally 子句)或外部代碼當(dāng)前線程已被中斷。 * 4. throw new RuntimeException(e);:將捕獲到的 InterruptedException 包裝成一個(gè)新的 RuntimeException 并拋出。這樣做是為了向上層代碼傳遞中斷信號(hào),并保留原始異常堆棧信息以供調(diào)試。根據(jù)具體應(yīng)用需求,可以選擇拋出自定義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 對(duì)應(yīng) lua腳本返回的0,1 defaultRedisScript.setResultType(Boolean.class); // 設(shè)置重入鎖腳本信息 defaultRedisScript.setScriptText(REENTRANT_LOCK_LUA); // 進(jìn)行重入鎖執(zhí)行 Boolean executeResult = redisUtil.executeBooleanLuaCustom(defaultRedisScript, Collections.singletonList(lockKey), value, lockTypeEnum.getExpireTime().intValue()); if (executeResult) { // 設(shè)置當(dāng)前key為激活狀態(tài) ACTIVE_KEY_SET.add(lockKey); // 設(shè)置定時(shí)任務(wù),進(jìn)行續(xù)期操作 resetLockExpire(lockTypeEnum, customKey, value, lockTypeEnum.getExpireTime()); } return executeResult; } /** * 進(jìn)行續(xù)期操作 * @param lockTypeEnum 鎖定類型 * @param customKey 自定義key * @param value 鎖定值,一般為線程id或者uuid * @param expireTime 過期時(shí)間 單位秒, */ public void resetLockExpire(LockTypeEnum lockTypeEnum, String customKey, String value, long expireTime) { // 續(xù)期的key信息 String resetKey = lockTypeEnum.getLockType().concat(StrUtil.COLON).concat(customKey); // 校驗(yàn)當(dāng)前key是否還在執(zhí)行過程中 if (!ACTIVE_KEY_SET.contains(resetKey)) { return; } // 時(shí)間設(shè)定延遲執(zhí)行時(shí)間delay,默認(rèn)續(xù)期時(shí)間是過期時(shí)間的1/3,在獲取鎖之后每expireTime/3時(shí)間進(jìn)行一次續(xù)期操作 long delay = expireTime <= 3 ? 1 : expireTime / 3; EXECUTOR_SERVICE.schedule(() -> { log.info("自定義key[{}],對(duì)應(yīng)值[{}]開始執(zhí)行續(xù)期操作!", resetKey, value); // 執(zhí)行續(xù)期操作,如果續(xù)期成功則再次添加續(xù)期任務(wù),如果續(xù)期成功,進(jìn)行下一次定時(shí)任務(wù)續(xù)期 DefaultRedisScript<Boolean> defaultRedisScript = new DefaultRedisScript<>(); // Boolean 對(duì)應(yīng) lua腳本返回的0,1 defaultRedisScript.setResultType(Boolean.class); // 設(shè)置重入鎖腳本信息 defaultRedisScript.setScriptText(EXPIRE_LUA); // 進(jìn)行重入鎖執(zhí)行 boolean executeLua = redisUtil.executeBooleanLuaCustom(defaultRedisScript, Collections.singletonList(resetKey), value, lockTypeEnum.getExpireTime().intValue()); if (executeLua) { log.info("執(zhí)行key[{}],value[{}]續(xù)期成功,進(jìn)行下一次續(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) { // 各個(gè)模塊服務(wù)啟動(dòng)時(shí)間差,預(yù)留5秒等待時(shí)間,防止重調(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 鍵值對(duì),若果結(jié)果不為空和0,則驗(yàn)證通過 if (ObjectUtil.isNotNull(result) && result != RELEASE_OK_FLAG) { // 當(dāng)前key釋放成功,從活躍生效keySet中移除 ACTIVE_KEY_SET.remove(releaseKey); return true; } return false; } }
注意,LUA腳本執(zhí)行過程中有時(shí)候會(huì)有執(zhí)行失敗情況,這些情況下異常信息很難捕捉,所以可以在LUA腳本中設(shè)置日志打印,但是需要注意,需要配置redis配置文件,打開日志信息,此處以重入鎖為例子,具體配置以及腳本信息如下:
- 1.redis配置日志級(jí)別,日志存儲(chǔ)位置信息
# 日志級(jí)別,可以設(shè)置為 debug、verbose、notice、warning,默認(rèn)為 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 鎖枚舉類實(shí)現(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, "緩存平臺(tái)初始化緩存信息分布式鎖", 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è)務(wù)冪等性校驗(yàn)"), WORK_FLOW_WEB_SERVICE_LOCK("WORK_FLOW_WEB_SERVICE_LOCK", 15L, "流程webService服務(wù)可用ip地址獲取鎖", 5000L), BASE_PRODUCT_TEST_LOCK("BASE_PRODUCT_TEST_LOCK", 10L, "產(chǎn)品測試分布式鎖", null), ; /** * 鎖類型 */ private String lockType; /** * 即過期時(shí)間,單位為second */ private Long expireTime; /** * 枷鎖成功后,默認(rèn)等待時(shí)間,時(shí)間應(yīng)小于過期時(shí)間,單位毫秒 */ private Long lockedWaitTimeMiles; /** * 描述信息 */ private String lockDesc; /** * 構(gòu)造方法 * @param lockType 類型 * @param lockTime 鎖定時(shí)間 * @param lockDesc 鎖描述 */ LockTypeEnum(String lockType, Long lockTime, String lockDesc) { this.lockDesc = lockDesc; this.expireTime = lockTime; this.lockType = lockType; } /** * 構(gòu)造方法 * @param lockType 類型 * @param lockTime 鎖定時(shí)間 * @param lockDesc 鎖描述 * @param lockedWaitTimeMiles 鎖失效時(shí)間 */ 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個(gè)產(chǎn)品,1000個(gè)線程進(jìn)行爭奪,具體實(shí)現(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,非核心線程空閑等待時(shí)間10秒,隊(duì)列最大1000 ThreadPoolExecutor executor = new ThreadPoolExecutor(100, 100, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000)); // 模擬1000個(gè)請(qǐng)求 CountDownLatch countDownLatch = new CountDownLatch(1000); // 模擬10000個(gè)人搶10個(gè)商品 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è)務(wù)處理 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è)務(wù)處理 try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } // 開始庫存計(jì)算 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個(gè)產(chǎn)品,具體如下圖
3.3 加鎖處理情況
多次實(shí)驗(yàn),沒有發(fā)生產(chǎn)品超發(fā)情況,具體測試結(jié)果如下:
4. 其他實(shí)現(xiàn)
還可以使用Redisson客戶端進(jìn)行分布式鎖實(shí)現(xiàn),這樣更加簡單安全,其有自己的看門狗機(jī)制,續(xù)期加鎖解鎖都更加方便,簡單操作過程實(shí)例代碼如下
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客戶端實(shí)現(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); // 獲取鎖對(duì)象 RLock lock = redisson.getLock("myLock"); try { // 嘗試獲取鎖,最多等待100秒,鎖定之后10秒自動(dòng)釋放 // 鎖定之后會(huì)自動(dòng)續(xù)期10秒 if (lock.tryLock(100, 10, TimeUnit.SECONDS)) { try { // 處理業(yè)務(wù)邏輯 } finally { // 釋放鎖 lock.unlock(); } } } catch (InterruptedException e) { // 處理中斷異常 Thread.currentThread().interrupt(); } finally { // 關(guān)閉Redisson客戶端 redisson.shutdown(); } } }
其中鎖定api是否開啟看門狗,整理如下
// 開始拿鎖,失敗阻塞重試 RLock lock = redissonClient.getLock("guodong"); // 具有Watch Dog 自動(dòng)延期機(jī)制 默認(rèn)續(xù)30s 每隔30/3=10 秒續(xù)到30s lock.lock(); // 嘗試拿鎖10s后停止重試,返回false 具有Watch Dog 自動(dòng)延期機(jī)制 默認(rèn)續(xù)30s boolean res1 = lock.tryLock(10, TimeUnit.SECONDS); // 嘗試拿鎖10s后,沒有Watch Dog lock.lock(10, TimeUnit.SECONDS); // 沒有Watch Dog ,10s后自動(dòng)釋放 lock.lock(10, TimeUnit.SECONDS); // 嘗試拿鎖100s后停止重試,返回false 沒有Watch Dog ,10s后自動(dòng)釋放 boolean res2 = lock.tryLock(100, 10, TimeUnit.SECONDS);
到此這篇關(guān)于redis分布式鎖實(shí)現(xiàn)示例的文章就介紹到這了,更多相關(guān)Redis分布式鎖內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
- Redis分布式鎖的幾種實(shí)現(xiàn)方法
- 使用Redis實(shí)現(xiàn)分布式鎖的代碼演示
- Redis使用SETNX命令實(shí)現(xiàn)分布式鎖
- Redis分布式鎖使用及說明
- Redisson分布式鎖解鎖異常問題
- Redis 實(shí)現(xiàn)分布式鎖時(shí)需要考慮的問題解決方案
- Redis實(shí)現(xiàn)分布式鎖的示例代碼
- Redission實(shí)現(xiàn)分布式鎖lock()和tryLock()方法的區(qū)別小結(jié)
- 從原理到實(shí)踐分析?Redis?分布式鎖的多種實(shí)現(xiàn)方案
- Redis本地鎖和分布式鎖的區(qū)別小結(jié)
相關(guān)文章
Redis list 類型學(xué)習(xí)筆記與總結(jié)
這篇文章主要介紹了Redis list 類型學(xué)習(xí)筆記與總結(jié),本文著重講解了關(guān)于List的一些常用方法,比如lpush 方法、lrange 方法、rpush 方法、linsert 方法、 lset 方法等,需要的朋友可以參考下2015-06-06redis.conf中使用requirepass不生效的原因及解決方法
本文主要介紹了如何啟用requirepass,以及啟用requirepass為什么不會(huì)生效,從代碼層面分析了不生效的原因,以及解決方法,需要的朋友可以參考下2023-07-07K8S部署Redis(單機(jī)、集群)的超詳細(xì)步驟
redis是一款基于BSD協(xié)議,開源的非關(guān)系型數(shù)據(jù)庫(nosql數(shù)據(jù)庫)這篇文章主要給大家介紹了關(guān)于K8S部署Redis(單機(jī)、集群)的超詳細(xì)步驟,文中通過代碼介紹的非常詳細(xì),需要的朋友可以參考下2024-05-05NoSQL和Redis簡介及Redis在Windows下的安裝和使用教程
這篇文章主要介紹了NoSQL和Redis簡介及Redis在Windows下的安裝和使用教程,本文同時(shí)講解了python操作redis,并給出了操作實(shí)例,需要的朋友可以參考下2015-01-01