作者：whynot_0

編輯：陶家龍

出處：juejin.im/post/6891571079702118407

不過目前互聯(lián)網(wǎng)項目越來越多的項目采用集群部署，也就是分布式情況，這兩種鎖就有些不夠用了。

來兩張圖舉例說明下，本地鎖的情況下：

分布式鎖情況下：

就其思想來說，就是一種“我全都要”的思想，所有服務(wù)都到一個統(tǒng)一的地方來取鎖，只有取到鎖的才能繼續(xù)執(zhí)行下去。

說完思想，下面來說一下具體的實現(xiàn)。

Redis 實現(xiàn)

為實現(xiàn)分布式鎖，在 Redis 中存在 SETNX key value 命令，意為 set if not exists（如果不存在該 key，才去 set 值），就比如說是張三去上廁所，看廁所門鎖著，他就不進去了，廁所門開著他才去。

可以看到，第一次 set 返回了 1，表示成功，但是第二次返回 0，表示 set 失敗，因為已經(jīng)存在這個 key 了。

當(dāng)然只靠 setnx 這個命令可以嗎？當(dāng)然是不行的，試想一種情況，張三在廁所里，但他在里面一直沒有釋放，一直在里面蹲著，那外面人想去廁所全部都去不了，都想錘死他了。

Redis 同理，假設(shè)已經(jīng)進行了加鎖，但是因為宕機或者出現(xiàn)異常未釋放鎖，就造成了所謂的“死鎖”。

聰明的你們肯定早都想到了，為它設(shè)置過期時間不就好了，可以 SETEX key seconds value 命令，為指定 key 設(shè)置過期時間，單位為秒。

但這樣又有另一個問題，我剛加鎖成功，還沒設(shè)置過期時間，Redis 宕機了不就又死鎖了，所以說要保證原子性吖，要么一起成功，要么一起失敗。

當(dāng)然我們能想到的 Redis 肯定早都為你實現(xiàn)好了，在 Redis 2.8 的版本后，Redis 就為我們提供了一條組合命令 SET key value ex seconds nx，加鎖的同時設(shè)置過期時間。

就好比是公司規(guī)定每人最多只能在廁所呆 2 分鐘，不管釋放沒釋放完都得出來，這樣就解決了“死鎖”問題。

但這樣就沒有問題了嗎？怎么可能。

試想又一種情況，廁所門肯定只能從里面開啊，張三上完廁所后張四進去鎖上門，但是外面人以為還是張三在里面，而且已經(jīng)過了 3 分鐘了，就直接把門給撬開了，一看里面卻是張四，這就很尷尬啊。

換成 Redis 就是說比如一個業(yè)務(wù)執(zhí)行時間很長，鎖已經(jīng)自己過期了，別人已經(jīng)設(shè)置了新的鎖，但是當(dāng)業(yè)務(wù)執(zhí)行完之后直接釋放鎖，就有可能是刪除了別人加的鎖，這不是亂套了嗎。

所以在加鎖時候，要設(shè)一個隨機值，在刪除鎖時進行比對，如果是自己的鎖，才刪除。

多說無益，煩人，直接上代碼：

//基于jedis和lua腳本來實現(xiàn) privatestaticfinal String LOCK_SUCCESS = "OK";
privatestaticfinal Long RELEASE_SUCCESS = 1L;
privatestaticfinal String SET_IF_NOT_EXIST = "NX";
privatestaticfinal String SET_WITH_EXPIRE_TIME = "PX"; @Override public String acquire() { try { // 獲取鎖的超時時間，超過這個時間則放棄獲取鎖 long end = System.currentTimeMillis() + acquireTimeout; // 隨機生成一個 value String requireToken = UUID.randomUUID().toString(); while (System.currentTimeMillis() < end) { String result = jedis
                .set(lockKey, requireToken, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime); if (LOCK_SUCCESS.equals(result)) { return requireToken;
            } try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    } catch (Exception e) {
        log.error("acquire lock due to error", e);
    }

    returnnull;
} @Override public boolean release(String identify) { if (identify == null) {
        returnfalse;
    } //通過lua腳本進行比對刪除操作，保證原子性 String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"; Object result = new Object(); try {
        result = jedis.eval(script, Collections.singletonList(lockKey),
            Collections.singletonList(identify)); if (RELEASE_SUCCESS.equals(result)) {
            log.info("release lock success, requestToken:{}", identify);
            returntrue;
        }
    } catch (Exception e) {
        log.error("release lock due to error", e);
    } finally { if (jedis != null) {
            jedis.close();
        }
    }

    log.info("release lock failed, requestToken:{}, result:{}", identify, result);
    returnfalse;
}

思考：加鎖和釋放鎖的原子性可以用 lua 腳本來保證，那鎖的自動續(xù)期改如何實現(xiàn)呢？

Redisson 實現(xiàn)

Redisson 顧名思義，Redis 的兒子，本質(zhì)上還是 Redis 加鎖，不過是對 Redis 做了很多封裝，它不僅提供了一系列的分布式的 Java 常用對象，還提供了許多分布式服務(wù)。

在引入 Redisson 的依賴后，就可以直接進行調(diào)用：

<dependency> <groupId>org.redissongroupId> <artifactId>redissonartifactId> <version>3.13.4version> dependency> 

先來一段 Redisson 的加鎖代碼：

private void test() { //分布式鎖名  鎖的粒度越細，性能越好 RLock lock = redissonClient.getLock("test_lock"); lock.lock(); try { //具體業(yè)務(wù)...... } finally { lock.unlock();
    }
}

就是這么簡單，使用方法 jdk 的 ReentrantLock 差不多，并且也支持 ReadWriteLock（讀寫鎖）、Reentrant Lock（可重入鎖）、Fair Lock（公平鎖）、RedLock（紅鎖）等各種鎖，詳細可以參照redisson官方文檔來查看。

那么 Redisson 到底有哪些優(yōu)勢呢？鎖的自動續(xù)期（默認都是 30 秒），如果業(yè)務(wù)超長，運行期間會自動給鎖續(xù)上新的 30s，不用擔(dān)心業(yè)務(wù)執(zhí)行時間超長而鎖被自動刪掉。

加鎖的業(yè)務(wù)只要運行完成，就不會給當(dāng)前續(xù)期，即便不手動解鎖，鎖默認在 30s 后刪除，不會造成死鎖問題。

前面也提到了鎖的自動續(xù)期，我們來看看 Redisson 是如何來實現(xiàn)的。

先說明一下，這里主要講的是 Redisson 中的 RLock，也就是可重入鎖，有兩種實現(xiàn)方法：

// 最常見的使用方法 lock.lock(); // 加鎖以后10秒鐘自動解鎖 // 無需調(diào)用unlock方法手動解鎖 lock.lock(10, TimeUnit.SECONDS);

而只有無參的方法是提供鎖的自動續(xù)期操作的，內(nèi)部使用的是“看門狗”機制，我們來看一看源碼。

不管是空參還是帶參方法，它們都調(diào)用的是同一個 lock 方法，未傳參的話時間傳了一個 -1，而帶參的方法傳過去的就是實際傳入的時間。

繼續(xù)點進 scheduleExpirationRenewal 方法： 點進 renewExpiration 方法：

總結(jié)一下，就是當(dāng)我們指定鎖過期時間，那么鎖到時間就會自動釋放。如果沒有指定鎖過期時間，就使用看門狗的默認時間 30s，只要占鎖成功，就會啟動一個定時任務(wù)，每隔 10s 給鎖設(shè)置新的過期時間，時間為看門狗的默認時間，直到鎖釋放。

小結(jié)：雖然 lock() 有自動續(xù)鎖機制，但是開發(fā)中還是推薦使用 lock(time，timeUnit)，因為它省掉了整個續(xù)期帶來的性能損，可以設(shè)置過期時間長一點，搭配 unlock()。

若業(yè)務(wù)執(zhí)行完成，會手動釋放鎖，若是業(yè)務(wù)執(zhí)行超時，那一般我們服務(wù)也都會設(shè)置業(yè)務(wù)超時時間，就直接報錯了，報錯后就會通過設(shè)置的過期時間來釋放鎖。

public void test() {
    RLock lock = redissonClient.getLock("test_lock"); lock.lock(30, TimeUnit.SECONDS); try { //.......具體業(yè)務(wù) } finally { //手動釋放鎖 lock.unlock();
    }
}

基于 Zookeeper 來實現(xiàn)分布式鎖

很多小伙伴都知道在分布式系統(tǒng)中，可以用 ZK 來做注冊中心，但其實在除了做祖冊中心以外，用 ZK 來做分布式鎖也是很常見的一種方案。

先來看一下 ZK 中是如何創(chuàng)建一個節(jié)點的？ZK 中存在 create [-s] [-e]  path [data] 命令，-s 為創(chuàng)建有序節(jié)點，-e 創(chuàng)建臨時節(jié)點。

這樣就創(chuàng)建了一個父節(jié)點并為父節(jié)點創(chuàng)建了一個子節(jié)點，組合命令意為創(chuàng)建一個臨時的有序節(jié)點。

而 ZK 中分布式鎖主要就是靠創(chuàng)建臨時的順序節(jié)點來實現(xiàn)的。至于為什么要用順序節(jié)點和為什么用臨時節(jié)點不用持久節(jié)點？先考慮一下，下文將作出說明。

同時還有 ZK 中如何查看節(jié)點？ZK 中 ls [-w] path 為查看節(jié)點命令，-w 為添加一個 watch（監(jiān)視器），/ 為查看根節(jié)點所有節(jié)點，可以看到我們剛才所創(chuàng)建的節(jié)點，同時如果是跟著指定節(jié)點名字的話為查看指定節(jié)點下的子節(jié)點。 后面的 00000000 為 ZK 為順序節(jié)點增加的順序。注冊監(jiān)聽器也是 ZK 實現(xiàn)分布式鎖中比較重要的一個東西。

下面來看一下 ZK 實現(xiàn)分布式鎖的主要流程：

• 當(dāng)?shù)谝粋€線程進來時會去父節(jié)點上創(chuàng)建一個臨時的順序節(jié)點。

• 第二個線程進來發(fā)現(xiàn)鎖已經(jīng)被持有了，就會為當(dāng)前持有鎖的節(jié)點注冊一個 watcher 監(jiān)聽器。

• 第三個線程進來發(fā)現(xiàn)鎖已經(jīng)被持有了，因為是順序節(jié)點的緣故，就會為上一個節(jié)點去創(chuàng)建一個 watcher 監(jiān)聽器。

• 當(dāng)?shù)谝粋€線程釋放鎖后，刪除節(jié)點，由它的下一個節(jié)點去占有鎖。

看到這里，聰明的小伙伴們都已經(jīng)看出來順序節(jié)點的好處了。非順序節(jié)點的話，每進來一個線程進來都會去持有鎖的節(jié)點上注冊一個監(jiān)聽器，容易引發(fā)“羊群效應(yīng)”。

這么大一群羊一起向你飛奔而來，不管你頂不頂?shù)米?，反?ZK 服務(wù)器是會增大宕機的風(fēng)險。

而順序節(jié)點的話就不會，順序節(jié)點當(dāng)發(fā)現(xiàn)已經(jīng)有線程持有鎖后，會向它的上一個節(jié)點注冊一個監(jiān)聽器，這樣當(dāng)持有鎖的節(jié)點釋放后，也只有持有鎖的下一個節(jié)點可以搶到鎖，相當(dāng)于是排好隊來執(zhí)行的，降低服務(wù)器宕機風(fēng)險。

至于為什么使用臨時節(jié)點，和 Redis 的過期時間一個道理，就算 ZK 服務(wù)器宕機，臨時節(jié)點會隨著服務(wù)器的宕機而消失，避免了死鎖的情況。

下面來上一段代碼的實現(xiàn)：

public class ZooKeeperDistributedLock implements Watcher { private ZooKeeper zk; private String locksRoot = "/locks"; private String productId; private String waitNode; private String lockNode; private CountDownLatch latch; private CountDownLatch connectedLatch = new CountDownLatch(1); private int sessionTimeout = 30000; public ZooKeeperDistributedLock(String productId) { this.productId = productId; try {
            String address = "192.168.189.131:2181,192.168.189.132:2181";
            zk = new ZooKeeper(address, sessionTimeout, this);
            connectedLatch.await();
        } catch (IOException e) { throw new LockException(e);
        } catch (KeeperException e) { throw new LockException(e);
        } catch (InterruptedException e) { throw new LockException(e);
        }
    } public void process(WatchedEvent event) { if (event.getState() == KeeperState.SyncConnected) {
            connectedLatch.countDown(); return;
        } if (this.latch != null) { this.latch.countDown();
        }
    } public void acquireDistributedLock() { try { if (this.tryLock()) { return;
            } else {
                waitForLock(waitNode, sessionTimeout);
            }
        } catch (KeeperException e) { throw new LockException(e);
        } catch (InterruptedException e) { throw new LockException(e);
        }
    } //獲取鎖 public boolean tryLock() { try { // 傳入進去的locksRoot + “/” + productId // 假設(shè)productId代表了一個商品id，比如說1 // locksRoot = locks // /locks/10000000000，/locks/10000000001，/locks/10000000002 lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 看看剛創(chuàng)建的節(jié)點是不是最小的節(jié)點 // locks：10000000000，10000000001，10000000002 Listlocks = zk.getChildren(locksRoot, false);
        Collections.sort(locks); if(lockNode.equals(locksRoot+"/"+ locks.get(0))){ //如果是最小的節(jié)點,則表示取得鎖 return true;
        } //如果不是最小的節(jié)點，找到比自己小1的節(jié)點 int previousLockIndex = -1; for(int i = 0; i < locks.size(); i++) { if(lockNode.equals(locksRoot + “/” + locks.get(i))) {
                    previousLockIndex = i - 1; break;
        }
       } this.waitNode = locks.get(previousLockIndex);
        } catch (KeeperException e) { throw new LockException(e);
        } catch (InterruptedException e) { throw new LockException(e);
        } return false;
    } private boolean waitForLock(String waitNode, long waitTime) throws InterruptedException, KeeperException {
        Stat stat = zk.exists(locksRoot + "/" + waitNode, true); if (stat != null) { this.latch = new CountDownLatch(1); this.latch.await(waitTime, TimeUnit.MILLISECONDS); this.latch = null;
        } return true;
    } //釋放鎖 public void unlock() { try {
            System.out.println("unlock " + lockNode);
            zk.delete(lockNode, -1);
            lockNode = null;
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    } //異常 public class LockException extends RuntimeException { private static final long serialVersionUID = 1L; public LockException(String e) {
            super(e);
        } public LockException(Exception e) {
            super(e);
        }
    }
}

總結(jié)

既然明白了 Redis 和 ZK 分別對分布式鎖的實現(xiàn)，那么總該有所不同的吧。沒錯，我都幫大家整理好了：

• 實現(xiàn)方式的不同，Redis 實現(xiàn)為去插入一條占位數(shù)據(jù)，而 ZK 實現(xiàn)為去注冊一個臨時節(jié)點。

• 遇到宕機情況時，Redis 需要等到過期時間到了后自動釋放鎖，而 ZK 因為是臨時節(jié)點，在宕機時候已經(jīng)是刪除了節(jié)點去釋放鎖。

• Redis 在沒搶占到鎖的情況下一般會去自旋獲取鎖，比較浪費性能，而 ZK 是通過注冊監(jiān)聽器的方式獲取鎖，性能而言優(yōu)于 Redis。

不過具體要采用哪種實現(xiàn)方式，還是需要具體情況具體分析，結(jié)合項目引用的技術(shù)棧來落地實現(xiàn)。