這篇文章主要介紹了redis怎么實(shí)現(xiàn)分布式鎖和等待序列,具有一定借鑒價(jià)值,感興趣的朋友可以參考下,希望大家閱讀完這篇文章之后大有收獲,下面讓小編帶著大家一起了解一下。
站在用戶的角度思考問(wèn)題,與客戶深入溝通,找到尉氏網(wǎng)站設(shè)計(jì)與尉氏網(wǎng)站推廣的解決方案,憑借多年的經(jīng)驗(yàn),讓設(shè)計(jì)與互聯(lián)網(wǎng)技術(shù)結(jié)合,創(chuàng)造個(gè)性化、用戶體驗(yàn)好的作品,建站類型包括:成都做網(wǎng)站、成都網(wǎng)站建設(shè)、企業(yè)官網(wǎng)、英文網(wǎng)站、手機(jī)端網(wǎng)站、網(wǎng)站推廣、域名注冊(cè)、網(wǎng)頁(yè)空間、企業(yè)郵箱。業(yè)務(wù)覆蓋尉氏地區(qū)。
在集群下,經(jīng)常會(huì)因?yàn)橥瑫r(shí)處理發(fā)生資源爭(zhēng)搶和并發(fā)問(wèn)題,但是我們都知道同步鎖 synchronized 、 cas 、 ReentrankLock 這些鎖的作用范圍都是 JVM ,說(shuō)白了在集群下沒(méi)啥用。這時(shí)我們就需要能在多臺(tái) JVM 之間決定執(zhí)行順序的鎖了,現(xiàn)在分布式鎖主要有 redis 、 Zookeeper 實(shí)現(xiàn)的,還有數(shù)據(jù)庫(kù)的方式,不過(guò)性能太差,也就是需要一個(gè)第三方的監(jiān)管。
背景
最近在做一個(gè)消費(fèi) Kafka 消息的時(shí)候發(fā)現(xiàn),由于線上的消費(fèi)者過(guò)多,經(jīng)常會(huì)遇到,多個(gè)機(jī)器同時(shí)處理一個(gè)主鍵類型的數(shù)據(jù)的情況發(fā)生,如果最后是執(zhí)行更新操作的話,也就是一個(gè)更新順序的問(wèn)題,但是如果恰好都需要插入數(shù)據(jù)的時(shí)候,會(huì)出現(xiàn)主鍵重復(fù)的問(wèn)題。這是生產(chǎn)上不被允許的(因?yàn)楣居挟惓1O(jiān)管的機(jī)制,扣分啥的),這是就需要個(gè)分布式鎖了,斟酌后用了 Redis 的實(shí)現(xiàn)方式(因?yàn)榫W(wǎng)上例子多)
分析
redis 實(shí)現(xiàn)的分布式鎖,實(shí)現(xiàn)原理是 set 方法,因?yàn)槎鄠€(gè)線程同時(shí)請(qǐng)求的時(shí)候,只有一個(gè)線程可以成功并返回結(jié)果,還可以設(shè)置有效期,來(lái)避免死鎖的發(fā)生,一切都是這么的完美,不過(guò)有個(gè)問(wèn)題,在 set 的時(shí)候,會(huì)直接返回結(jié)果,成功或者失敗,不具有阻塞效果,需要我們自己對(duì)失敗的線程進(jìn)程處理,有兩種方式
丟棄
等待重試 由于我們的系統(tǒng)需要這些數(shù)據(jù),那么只能重新嘗試獲取。這里使用 redis 的 List 類型實(shí)現(xiàn)等待序列的作用
代碼
直接上代碼 其實(shí)直接redis的工具類就可以解決了
package com.test import redis.clients.jedis.Jedis; import java.util.Collections; import java.util.List; /** * @desc redis隊(duì)列實(shí)現(xiàn)方式 * @anthor * @date **/ public class RedisUcUitl { private static final String LOCK_SUCCESS = "OK"; private static final String SET_IF_NOT_EXIST = "NX"; private static final String SET_WITH_EXPIRE_TIME = "PX"; private static final Long RELEASE_SUCCESS = 1L; private RedisUcUitl() { } /** * logger **/ /** * 存儲(chǔ)redis隊(duì)列順序存儲(chǔ) 在隊(duì)列首部存入 * * @param key 字節(jié)類型 * @param value 字節(jié)類型 */ public static Long lpush(Jedis jedis, final byte[] key, final byte[] value) { return jedis.lpush(key, value); } /** * 移除列表中最后一個(gè)元素 并將改元素添加入另一個(gè)列表中 ,當(dāng)列表為空時(shí) 將阻塞連接 直到等待超時(shí) * * @param srckey * @param dstkey * @param timeout 0 表示永不超時(shí) * @return */ public static byte[] brpoplpush(Jedis jedis,final byte[] srckey, final byte[] dstkey, final int timeout) { return jedis.brpoplpush(srckey, dstkey, timeout); } /** * 返回制定的key,起始位置的redis數(shù)據(jù) * @param redisKey * @param start * @param end -1 表示到最后 * @return */ public static List<byte[]> lrange(Jedis jedis,final byte[] redisKey, final long start, final long end) { return jedis.lrange(redisKey, start, end); } /** * 刪除key * @param redisKey */ public static void delete(Jedis jedis, final byte[] redisKey) { return jedis.del(redisKey); } /** * 嘗試加鎖 * @param lockKey key名稱 * @param requestId 身份標(biāo)識(shí) * @param expireTime 過(guò)期時(shí)間 * @return */ public static boolean tryGetDistributedLock(Jedis jedis,final String lockKey, final String requestId, final int expireTime) { String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime); return LOCK_SUCCESS.equals(result); } /** * 釋放鎖 * @param lockKey key名稱 * @param requestId 身份標(biāo)識(shí) * @return */ public static boolean releaseDistributedLock(Jedis jedis,final String lockKey, final String requestId) { final String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId)); return RELEASE_SUCCESS.equals(result); } }
業(yè)務(wù)邏輯主要代碼如下
1.先消耗隊(duì)列中的
while(true){ // 消費(fèi)隊(duì)列 try{ // 被放入redis隊(duì)列的數(shù)據(jù) 序列化后的 byte[] bytes = RedisUcUitl.brpoplpush(keyStr.getBytes(UTF_8), dstKeyStr.getBytes(UTF_8), 1); if(bytes == null || bytes.isEmpty()){ // 隊(duì)列中沒(méi)數(shù)據(jù)時(shí)退出 break; } // 反序列化對(duì)象 Map<String, Object> singleMap = (Map<String, Object>) ObjectSerialUtil.bytesToObject(bytes); // 塞入唯一的值 防止被其他線程誤解鎖 String requestId = UUID.randomUUID().toString(); boolean lockGetFlag = RedisUcUitl.tryGetDistributedLock(keyStr,requestId, 100); if(lockGetFlag){ // 成功獲取鎖 進(jìn)行業(yè)務(wù)處理 //TODO // 處理完畢釋放鎖 boolean freeLock = RedisUcUitl.releaseDistributedLock(keyStr, requestId); }else{ // 未能獲得鎖放入等待隊(duì)列 RedisUcUitl.lpush(keyStr.getBytes(UTF_8), ObjectSerialUtil.objectToBytes(param)); } }catch(Exception e){ break; } }
2.處理最新接到的數(shù)據(jù)
同樣是走嘗試獲取鎖,獲取不到放入隊(duì)列的流程
一般序列化用 fastJson 之列的就可以了,這里用的是 JDK 自帶的,工具類如下
public class ObjectSerialUtil { private ObjectSerialUtil() { // 工具類 } /** * 將Object對(duì)象序列化為byte[] * * @param obj 對(duì)象 * @return byte數(shù)組 * @throws Exception */ public static byte[] objectToBytes(Object obj) throws IOException { ByteArrayOutputStream bos = new ByteArrayOutputStream(); ObjectOutputStream oos = new ObjectOutputStream(bos); oos.writeObject(obj); byte[] bytes = bos.toByteArray(); bos.close(); oos.close(); return bytes; } /** * 將bytes數(shù)組還原為對(duì)象 * * @param bytes * @return * @throws Exception */ public static Object bytesToObject(byte[] bytes) { try { ByteArrayInputStream bin = new ByteArrayInputStream(bytes); ObjectInputStream ois = new ObjectInputStream(bin); return ois.readObject(); } catch (Exception e) { throw new BaseException("反序列化出錯(cuò)!", e); } } }
感謝你能夠認(rèn)真閱讀完這篇文章,希望小編分享的“Redis怎么實(shí)現(xiàn)分布式鎖和等待序列”這篇文章對(duì)大家有幫助,同時(shí)也希望大家多多支持創(chuàng)新互聯(lián),關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道,更多相關(guān)知識(shí)等著你來(lái)學(xué)習(xí)!
文章名稱:Redis怎么實(shí)現(xiàn)分布式鎖和等待序列
URL網(wǎng)址:http://bm7419.com/article20/geigjo.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供小程序開發(fā)、定制開發(fā)、軟件開發(fā)、企業(yè)網(wǎng)站制作、商城網(wǎng)站、全網(wǎng)營(yíng)銷推廣
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來(lái)源: 創(chuàng)新互聯(lián)