|
為什么用分布式鎖？在討論這個問題之前，我們先來看一個業(yè)務(wù)場景。 圖片來自 Pexels 為什么用分布式鎖？ 系統(tǒng) A 是一個電商系統(tǒng)，目前是一臺機(jī)器部署，系統(tǒng)中有一個用戶下訂單的接口，但是用戶下訂單之前一定要去檢查一下庫存，確保庫存足夠了才會給用戶下單。 由于系統(tǒng)有一定的并發(fā)，所以會預(yù)先將商品的庫存保存在 Redis 中，用戶下單的時候會更新 Redis 的庫存。 此時系統(tǒng)架構(gòu)如下： 但是這樣一來會產(chǎn)生一個問題：假如某個時刻，Redis 里面的某個商品庫存為 1。 此時兩個請求同時到來，其中一個請求執(zhí)行到上圖的第 3 步，更新數(shù)據(jù)庫的庫存為 0，但是第 4 步還沒有執(zhí)行。 而另外一個請求執(zhí)行到了第 2 步，發(fā)現(xiàn)庫存還是 1，就繼續(xù)執(zhí)行第 3 步。這樣的結(jié)果，是導(dǎo)致賣出了 2 個商品，然而其實庫存只有 1 個。 很明顯不對?。∵@就是典型的庫存超賣問題。此時，我們很容易想到解決方案：用鎖把 2、3、4 步鎖住，讓他們執(zhí)行完之后，另一個線程才能進(jìn)來執(zhí)行第 2 步。 按照上面的圖，在執(zhí)行第 2 步時，使用 Java 提供的 Synchronized 或者 ReentrantLock 來鎖住，然后在第 4 步執(zhí)行完之后才釋放鎖。 這樣一來，2、3、4 這 3 個步驟就被“鎖”住了，多個線程之間只能串行化執(zhí)行。 但是好景不長，整個系統(tǒng)的并發(fā)飆升，一臺機(jī)器扛不住了?，F(xiàn)在要增加一臺機(jī)器，如下圖： 增加機(jī)器之后，系統(tǒng)變成上圖所示，我的天！假設(shè)此時兩個用戶的請求同時到來，但是落在了不同的機(jī)器上，那么這兩個請求是可以同時執(zhí)行了，還是會出現(xiàn)庫存超賣的問題。 為什么呢？因為上圖中的兩個 A 系統(tǒng)，運(yùn)行在兩個不同的 JVM 里面，他們加的鎖只對屬于自己 JVM 里面的線程有效，對于其他 JVM 的線程是無效的。 因此，這里的問題是：Java 提供的原生鎖機(jī)制在多機(jī)部署場景下失效了，這是因為兩臺機(jī)器加的鎖不是同一個鎖（兩個鎖在不同的 JVM 里面）。 那么，我們只要保證兩臺機(jī)器加的鎖是同一個鎖，問題不就解決了嗎？此時，就該分布式鎖隆重登場了。 分布式鎖的思路是：在整個系統(tǒng)提供一個全局、唯一的獲取鎖的“東西”，然后每個系統(tǒng)在需要加鎖時，都去問這個“東西”拿到一把鎖，這樣不同的系統(tǒng)拿到的就可以認(rèn)為是同一把鎖。 至于這個“東西”，可以是 Redis、Zookeeper，也可以是數(shù)據(jù)庫。文字描述不太直觀，我們來看下圖： 通過上面的分析，我們知道了庫存超賣場景在分布式部署系統(tǒng)的情況下使用 Java 原生的鎖機(jī)制無法保證線程安全，所以我們需要用到分布式鎖的方案。 那么，如何實現(xiàn)分布式鎖呢？接著往下看！ 基于 Redis 實現(xiàn)分布式鎖 上面分析為啥要使用分布式鎖了，這里我們來具體看看分布式鎖落地的時候應(yīng)該怎么樣處理。 ①常見的一種方案就是使用 Redis 做分布式鎖 使用 Redis 做分布式鎖的思路大概是這樣的：在 Redis 中設(shè)置一個值表示加了鎖，然后釋放鎖的時候就把這個 Key 刪除。 具體代碼是這樣的： // 獲取鎖 // NX是指如果key不存在就成功，key存在返回false，PX可以指定過期時間 SET anyLock unique_value NX PX 30000 // 釋放鎖：通過執(zhí)行一段lua腳本 // 釋放鎖涉及到兩條指令，這兩條指令不是原子性的 // 需要用到redis的lua腳本支持特性，redis執(zhí)行l(wèi)ua腳本是原子性的 if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end 這種方式有幾大要點(diǎn)：
這時避免了一種情況：假設(shè) A 獲取了鎖，過期時間 30s，此時 35s 之后，鎖已經(jīng)自動釋放了，A 去釋放鎖，但是此時可能 B 獲取了鎖。A 客戶端就不能刪除 B 的鎖了。 除了要考慮客戶端要怎么實現(xiàn)分布式鎖之外，還需要考慮 Redis 的部署問題。 Redis 有 3 種部署方式：
使用 Redis 做分布式鎖的缺點(diǎn)在于：如果采用單機(jī)部署模式，會存在單點(diǎn)問題，只要 Redis 故障了。加鎖就不行了。 采用 Master-Slave 模式，加鎖的時候只對一個節(jié)點(diǎn)加鎖，即便通過 Sentinel 做了高可用，但是如果 Master 節(jié)點(diǎn)故障了，發(fā)生主從切換，此時就會有可能出現(xiàn)鎖丟失的問題。 基于以上的考慮，Redis 的作者也考慮到這個問題，他提出了一個 RedLock 的算法。 這個算法的意思大概是這樣的：假設(shè) Redis 的部署模式是 Redis Cluster，總共有 5 個 Master 節(jié)點(diǎn)。 通過以下步驟獲取一把鎖：
但是這樣的這種算法還是頗具爭議的，可能還會存在不少的問題，無法保證加鎖的過程一定正確。 ②另一種方式：Redisson 此外，實現(xiàn) Redis 的分布式鎖，除了自己基于 Redis Client 原生 API 來實現(xiàn)之外，還可以使用開源框架：Redission。 Redisson 是一個企業(yè)級的開源 Redis Client，也提供了分布式鎖的支持。我也非常推薦大家使用，為什么呢？ 回想一下上面說的，如果自己寫代碼來通過 Redis 設(shè)置一個值，是通過下面這個命令設(shè)置的： SET anyLock unique_value NX PX 30000 這里設(shè)置的超時時間是 30s，假如我超過 30s 都還沒有完成業(yè)務(wù)邏輯的情況下，Key 會過期，其他線程有可能會獲取到鎖。 這樣一來的話，第一個線程還沒執(zhí)行完業(yè)務(wù)邏輯，第二個線程進(jìn)來了也會出現(xiàn)線程安全問題。 所以我們還需要額外的去維護(hù)這個過期時間，太麻煩了~我們來看看 Redisson 是怎么實現(xiàn)的？ 先感受一下使用 Redission 的爽： Config config = new Config(); config.useClusterServers() .addNodeAddress('redis://192.168.31.101:7001') .addNodeAddress('redis://192.168.31.101:7002') .addNodeAddress('redis://192.168.31.101:7003') .addNodeAddress('redis://192.168.31.102:7001') .addNodeAddress('redis://192.168.31.102:7002') .addNodeAddress('redis://192.168.31.102:7003'); RedissonClient redisson = Redisson.create(config); RLock lock = redisson.getLock('anyLock'); lock.lock(); lock.unlock(); 就是這么簡單，我們只需要通過它的 API 中的 Lock 和 Unlock 即可完成分布式鎖，他幫我們考慮了很多細(xì)節(jié)：
這樣的話，就算一直持有鎖也不會出現(xiàn) Key 過期了，其他線程獲取到鎖的問題了。
這里稍微貼出來其實現(xiàn)代碼： // 加鎖邏輯 private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { if (leaseTime != -1) { return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } // 調(diào)用一段lua腳本，設(shè)置一些key、過期時間 RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); ttlRemainingFuture.addListener(new FutureListener<Long>() { @Override public void operationComplete(Future<Long> future) throws Exception { if (!future.isSuccess()) { return; } Long ttlRemaining = future.getNow(); // lock acquired if (ttlRemaining == null) { // 看門狗邏輯 scheduleExpirationRenewal(threadId); } } }); return ttlRemainingFuture; } <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { internalLockLeaseTime = unit.toMillis(leaseTime); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, 'if (redis.call('exists', KEYS[1]) == 0) then ' + 'redis.call('hset', KEYS[1], ARGV[2], 1); ' + 'redis.call('pexpire', KEYS[1], ARGV[1]); ' + 'return nil; ' + 'end; ' + 'if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then ' + 'redis.call('hincrby', KEYS[1], ARGV[2], 1); ' + 'redis.call('pexpire', KEYS[1], ARGV[1]); ' + 'return nil; ' + 'end; ' + 'return redis.call('pttl', KEYS[1]);', Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); } // 看門狗最終會調(diào)用了這里 private void scheduleExpirationRenewal(final long threadId) { if (expirationRenewalMap.containsKey(getEntryName())) { return; } // 這個任務(wù)會延遲10s執(zhí)行 Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { // 這個操作會將key的過期時間重新設(shè)置為30s RFuture<Boolean> future = renewExpirationAsync(threadId); future.addListener(new FutureListener<Boolean>() { @Override public void operationComplete(Future<Boolean> future) throws Exception { expirationRenewalMap.remove(getEntryName()); if (!future.isSuccess()) { log.error('Can't update lock ' + getName() + ' expiration', future.cause()); return; } if (future.getNow()) { // reschedule itself // 通過遞歸調(diào)用本方法，無限循環(huán)延長過期時間 scheduleExpirationRenewal(threadId); } } }); } }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) { task.cancel(); } } 另外，Redisson 還提供了對 Redlock 算法的支持，它的用法也很簡單： RedissonClient redisson = Redisson.create(config); RLock lock1 = redisson.getFairLock('lock1'); RLock lock2 = redisson.getFairLock('lock2'); RLock lock3 = redisson.getFairLock('lock3'); RedissonRedLock multiLock = new RedissonRedLock(lock1, lock2, lock3); multiLock.lock(); multiLock.unlock(); 小結(jié)：本節(jié)分析了使用 Redis 作為分布式鎖的具體落地方案以及其一些局限性，然后介紹了一個 Redis 的客戶端框架 Redisson，這也是我推薦大家使用的，比自己寫代碼實現(xiàn)會少 Care 很多細(xì)節(jié)。 基于 Zookeeper 實現(xiàn)分布式鎖 常見的分布式鎖實現(xiàn)方案里面，除了使用 Redis 來實現(xiàn)之外，使用 Zookeeper 也可以實現(xiàn)分布式鎖。 在介紹 Zookeeper(下文用 ZK 代替)實現(xiàn)分布式鎖的機(jī)制之前，先粗略介紹一下 ZK 是什么東西：ZK 是一種提供配置管理、分布式協(xié)同以及命名的中心化服務(wù)。 ZK 的模型是這樣的：ZK 包含一系列的節(jié)點(diǎn)，叫做 Znode，就好像文件系統(tǒng)一樣，每個 Znode 表示一個目錄。 然后 Znode 有一些特性：
例如我們可以創(chuàng)建子節(jié)點(diǎn)“/lock/node-”并且指明有序，那么 ZK 在生成子節(jié)點(diǎn)時會根據(jù)當(dāng)前的子節(jié)點(diǎn)數(shù)量自動添加整數(shù)序號。 也就是說，如果是第一個創(chuàng)建的子節(jié)點(diǎn)，那么生成的子節(jié)點(diǎn)為 /lock/node-0000000000，下一個節(jié)點(diǎn)則為 /lock/node-0000000001，依次類推。
當(dāng)前 ZK 有如下四種事件：
基于以上的一些 ZK 的特性，我們很容易得出使用 ZK 實現(xiàn)分布式鎖的落地方案：
比如當(dāng)前線程獲取到的節(jié)點(diǎn)序號為 /lock/003，然后所有的節(jié)點(diǎn)列表為[/lock/001，/lock/002，/lock/003]，則對 /lock/002 這個節(jié)點(diǎn)添加一個事件監(jiān)聽器。 如果鎖釋放了，會喚醒下一個序號的節(jié)點(diǎn)，然后重新執(zhí)行第 3 步，判斷是否自己的節(jié)點(diǎn)序號是最小。 比如 /lock/001 釋放了，/lock/002 監(jiān)聽到時間，此時節(jié)點(diǎn)集合為[/lock/002，/lock/003]，則 /lock/002 為最小序號節(jié)點(diǎn)，獲取到鎖。 整個過程如下： 具體的實現(xiàn)思路就是這樣，至于代碼怎么寫，這里比較復(fù)雜就不貼出來了。 Curator 介紹 Curator 是一個 ZK 的開源客戶端，也提供了分布式鎖的實現(xiàn)。它的使用方式也比較簡單： InterProcessMutex interProcessMutex = new InterProcessMutex(client,'/anyLock'); interProcessMutex.acquire(); interProcessMutex.release(); 其實現(xiàn)分布式鎖的核心源碼如下： private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception { boolean haveTheLock = false; boolean doDelete = false; try { if ( revocable.get() != null ) { client.getData().usingWatcher(revocableWatcher).forPath(ourPath); } while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock ) { // 獲取當(dāng)前所有節(jié)點(diǎn)排序后的集合 List<String> children = getSortedChildren(); // 獲取當(dāng)前節(jié)點(diǎn)的名稱 String sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash // 判斷當(dāng)前節(jié)點(diǎn)是否是最小的節(jié)點(diǎn) PredicateResults predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases); if ( predicateResults.getsTheLock() ) { // 獲取到鎖 haveTheLock = true; } else { // 沒獲取到鎖，對當(dāng)前節(jié)點(diǎn)的上一個節(jié)點(diǎn)注冊一個監(jiān)聽器 String previousSequencePath = basePath + '/' + predicateResults.getPathToWatch(); synchronized(this){ Stat stat = client.checkExists().usingWatcher(watcher).forPath(previousSequencePath); if ( stat != null ){ if ( millisToWait != null ){ millisToWait -= (System.currentTimeMillis() - startMillis); startMillis = System.currentTimeMillis(); if ( millisToWait <= 0 ){ doDelete = true; // timed out - delete our node break; } wait(millisToWait); }else{ wait(); } } } // else it may have been deleted (i.e. lock released). Try to acquire again } } } catch ( Exception e ) { doDelete = true; throw e; } finally{ if ( doDelete ){ deleteOurPath(ourPath); } } return haveTheLock; } 其實 Curator 實現(xiàn)分布式鎖的底層原理和上面分析的是差不多的。這里我們用一張圖詳細(xì)描述其原理： 小結(jié)：本節(jié)介紹了 ZK 實現(xiàn)分布式鎖的方案以及 ZK 的開源客戶端的基本使用，簡要的介紹了其實現(xiàn)原理。 兩種方案的優(yōu)缺點(diǎn)比較 學(xué)完了兩種分布式鎖的實現(xiàn)方案之后，本節(jié)需要討論的是 Redis 和 ZK 的實現(xiàn)方案中各自的優(yōu)缺點(diǎn)。 對于 Redis 的分布式鎖而言，它有以下缺點(diǎn)：
但是另一方面使用 Redis 實現(xiàn)分布式鎖在很多企業(yè)中非常常見，而且大部分情況下都不會遇到所謂的“極端復(fù)雜場景”。 所以使用 Redis 作為分布式鎖也不失為一種好的方案，最重要的一點(diǎn)是 Redis 的性能很高，可以支撐高并發(fā)的獲取、釋放鎖操作。 對于 ZK 分布式鎖而言:
但是 ZK 也有其缺點(diǎn)：如果有較多的客戶端頻繁的申請加鎖、釋放鎖，對于 ZK 集群的壓力會比較大。 小結(jié)：綜上所述，Redis 和 ZK 都有其優(yōu)缺點(diǎn)。我們在做技術(shù)選型的時候可以根據(jù)這些問題作為參考因素。 一些建議 通過前面的分析，實現(xiàn)分布式鎖的兩種常見方案：Redis 和 ZK，他們各有千秋。應(yīng)該如何選型呢？ 就個人而言的話，我比較推崇 ZK 實現(xiàn)的鎖：因為 Redis 是有可能存在隱患的，可能會導(dǎo)致數(shù)據(jù)不對的情況。但是，怎么選用要看具體在公司的場景了。 如果公司里面有 ZK 集群條件，優(yōu)先選用 ZK 實現(xiàn)，但是如果說公司里面只有 Redis 集群，沒有條件搭建 ZK 集群。 那么其實用 Redis 來實現(xiàn)也可以，另外還可能是系統(tǒng)設(shè)計者考慮到了系統(tǒng)已經(jīng)有 Redis，但是又不希望再次引入一些外部依賴的情況下，可以選用 Redis。這個是要系統(tǒng)設(shè)計者基于架構(gòu)來考慮了。 |
|
|
