Redis分布式鎖及4種常見實(shí)現(xiàn)方法

更新時(shí)間：2024年05月22日 11:24:48 作者：lcx_defender

本文主要介紹了Redis分布式鎖及4種常見實(shí)現(xiàn)方法,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

線程鎖

主要用來給方法、代碼塊加鎖。當(dāng)某個(gè)方法或代碼使用鎖，在同一時(shí)刻僅有一個(gè)線程執(zhí)行該方法或該代碼段。線程鎖只在同一JVM中有效果，因?yàn)榫€程鎖的實(shí)現(xiàn)在根本上是依靠線程之間共享內(nèi)存實(shí)現(xiàn)的，比如Synchronized、Lock等。

進(jìn)程鎖

控制同一操作系統(tǒng)中多個(gè)進(jìn)程訪問某個(gè)共享資源，因?yàn)檫M(jìn)程具有獨(dú)立性，各個(gè)進(jìn)程無法訪問其他進(jìn)程的資源，因此無法通過synchronized等線程鎖實(shí)現(xiàn)進(jìn)程鎖

什么是分布式鎖

分布式鎖：滿足分布式系統(tǒng)或集群模式下多進(jìn)程可見并且互斥的鎖；一個(gè)方法在同一時(shí)間只能被一個(gè)機(jī)器的一個(gè)線程執(zhí)行。

分布式鎖應(yīng)具備的條件

多進(jìn)程可見
互斥
高可用的獲取鎖與釋放鎖；
高性能的獲取鎖與釋放鎖；
具備鎖失效機(jī)制，防止死鎖；
具備可重入特性；
具備非阻塞鎖特性，即沒有獲取到鎖將直接返回獲取鎖失敗；

分布式鎖常見的實(shí)現(xiàn)方式

基于Mysql

在數(shù)據(jù)庫中創(chuàng)建一個(gè)表，表中包含方法名等字段，并在方法名name字段上創(chuàng)建唯一索引，想要執(zhí)行某個(gè)方法，就使用這個(gè)方法名向表中插入一條記錄，成功插入則獲取鎖，刪除對(duì)應(yīng)的行就是鎖釋放。

CREATE TABLE `method_lock` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '主鍵',
  `method_name` varchar(64) NOT NULL COMMENT '鎖定的方法名',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uidx_method_name` (`method_name`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COMMENT='鎖定中的方法';

這里主要是用method_name字段作為唯一索引來實(shí)現(xiàn)，唯一索引保證了該記錄的唯一性，鎖釋放就直接刪掉該條記錄就行了。

INSERT INTO method_lock (method_name) VALUES ('methodName');

delete from method_lock where method_name ='methodName';

缺點(diǎn)

1、因?yàn)槭腔跀?shù)據(jù)庫實(shí)現(xiàn)的，數(shù)據(jù)庫的可用性和性能將直接影響分布式鎖的可用性及性能。高并發(fā)狀態(tài)下，數(shù)據(jù)庫讀寫效率一般都非常緩慢。所以，數(shù)據(jù)庫需要雙機(jī)部署、數(shù)據(jù)同步、主備切換；

2、不具備可重入的特性，因?yàn)橥粋€(gè)線程在釋放鎖之前，行數(shù)據(jù)一直存在，無法再次成功插入數(shù)據(jù)，所以，需要在表中新增一列，用于記錄當(dāng)前獲取到鎖的機(jī)器和線程信息，在再次獲取鎖的時(shí)候，先查詢表中機(jī)器和線程信息是否和當(dāng)前機(jī)器和線程相同，若相同則直接獲取鎖；

3、沒有鎖失效機(jī)制，因?yàn)橛锌赡艹霈F(xiàn)成功插入數(shù)據(jù)后，服務(wù)器宕機(jī)了，對(duì)應(yīng)的數(shù)據(jù)沒有被刪除，當(dāng)服務(wù)恢復(fù)后一直獲取不到鎖。所以，需要在表中新增一列，用于記錄失效時(shí)間，并且需要有定時(shí)任務(wù)清除這些失效的數(shù)據(jù)；

4、不具備阻塞鎖特性，獲取不到鎖直接返回失敗，所以需要優(yōu)化獲取邏輯，循環(huán)多次去獲取。

5、在實(shí)施的過程中會(huì)遇到各種不同的問題，為了解決這些問題，實(shí)現(xiàn)方式將會(huì)越來越復(fù)雜；依賴數(shù)據(jù)庫需要一定的資源開銷，性能問題需要考慮。

基于Redis分布式鎖

獲取鎖

利用setnx這種互斥命令，利用鎖超時(shí)時(shí)間進(jìn)行到期釋放避免死鎖，且Redis具有高可用高性能等特點(diǎn)及優(yōu)勢。

Redis 的分布式鎖， setnx 命令并設(shè)置過期時(shí)間就行嗎？

setnx [key] [value] 
expire [key] 30

雖然setnx是原子性的，但是setnx + expire就不是了，也就是說setnx和expire是分兩步執(zhí)行的，【加鎖和超時(shí)】兩個(gè)操作是分開的，如果expire執(zhí)行失敗了，那么鎖同樣得不到釋放。

獲取鎖的原子性問題

# 設(shè)置某個(gè) key 的值并設(shè)置多少毫秒或秒 過期
set <key> <value> PX <多少毫秒> NX
或
set <key> <value> EX <多少秒> NX
# 設(shè)置一個(gè)鍵為lock，值為thread，ex表示以秒為單位，px以微秒為單位，nx表示不存在該key的時(shí)候才能設(shè)置
set lock thread1 nx ex 10

當(dāng)且僅當(dāng)key值lock不存在時(shí)，set一個(gè)key為lock，val為thread1的字符串，返回1；若key存在，則什么都不做，返回0。

Java的實(shí)現(xiàn)

	public boolean tryLock(long timeoutSec) {
        // 獲取線程標(biāo)識(shí),ID_PREFIX為
        String threadId = ID_PREFIX +  Thread.currentThread().getId();
        // 獲取鎖,name為自定義的業(yè)務(wù)名稱
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(success);
    }

釋放鎖

手動(dòng)釋放

# 將對(duì)應(yīng)的鍵刪除即可
del [key]

超時(shí)釋放

釋放錯(cuò)誤的鎖

假設(shè)如下三個(gè)線程是同一個(gè)用戶的業(yè)務(wù)線程，即假設(shè)線程1、線程2、線程3申請(qǐng)的分布式鎖key一樣：

線程1獲取成功了一個(gè)分布式鎖，由于一些問題，線程1執(zhí)行超時(shí)了，分布式鎖被超期釋放。
在鎖釋放后，有一個(gè)線程2又來獲取鎖，并且成功。
在線程2執(zhí)行過程中，線程1運(yùn)行結(jié)束，由于不知道自己鎖已經(jīng)被超期釋放，所以它直接手動(dòng)釋放鎖，錯(cuò)誤的釋放了線程2的鎖。
這時(shí)如果又有一個(gè)線程3前來獲取鎖，就能獲取成功；而線程2此時(shí)也持有鎖。

所以，設(shè)置鎖的過期時(shí)間時(shí)，還需要設(shè)置唯一編號(hào)。在編程實(shí)現(xiàn)釋放鎖的時(shí)候，需要判斷當(dāng)前釋放的鎖的值是否與之前的一致;若一致，則刪除；不一致，則不操作。

代碼示例：

	public void unlock() {
        // 獲取線程標(biāo)識(shí)
        String threadId = ID_PREFIX +  Thread.currentThread().getId();
        // 獲取鎖中的標(biāo)識(shí)
        String id = stringRedisTemplate.opsForValue().get(KEY_PREFIX + name);
        // 判斷標(biāo)識(shí)是否一致
        if (threadId.equals(id)) {
            // 釋放鎖
            stringRedisTemplate.delete(KEY_PREFIX + name);
        }
    }

刪除鎖的原子性問題

線程1在時(shí)限內(nèi)完成了業(yè)務(wù)，它開始執(zhí)行刪除鎖的操作。
線程1判斷完當(dāng)前鎖的標(biāo)識(shí)(對(duì)應(yīng)的value)一樣后，由于一些問題，線程1被阻塞，該key超期被刪除了
線程2過來申請(qǐng)分布式鎖，并且成功
此時(shí)，線程1才正式對(duì)分布式鎖執(zhí)行刪除，由于可能是同一個(gè)用戶的業(yè)務(wù)線程，線程1與線程2的申請(qǐng)的分布式鎖key一樣，所以線程1調(diào)用的刪除鎖操作將線程2的鎖刪掉了（悲！故技重施！）…

所以，我們還得確保獲取和刪除操作之間的原子性?？梢越柚鶯ua腳本保證原子性，釋放鎖的核心邏輯【GET、判斷、DEL】，寫成 Lua 腳本，讓Redis調(diào)用。

使用Lua腳本改進(jìn)Redis釋放分布式鎖

Lua中Redis的調(diào)用函數(shù)

redis.call('命令名稱','key','其他參數(shù)',...)

比如我們執(zhí)行set name jack命令，可以使用:

redis.call('set','name','jack')

使用Redis調(diào)用Lua腳本

調(diào)用方法

# script腳本語句；numkeys腳本需要的key類型的參數(shù)個(gè)數(shù)
eval script numkeys key [key ...] arg [arg ...]

例如，執(zhí)行redis.call('set', 'name', 'Jack')腳本設(shè)置redis鍵值，語法如下：

eval "return redis.call('set', 'name', 'Jack')" 0

如果key和value不想寫死，可以使用如下格式

eval "return redis.call('set', KEYS[1], ARGV[1])" 1 name Jack

Lua中，數(shù)組下標(biāo)從1開始，此處1標(biāo)識(shí)只有一個(gè)key類型的參數(shù)，其他參數(shù)都會(huì)放入ARGV數(shù)組。在腳本中，可以通過KEYS數(shù)組和ARGV數(shù)組獲取參數(shù)

Lua腳本（unlock.lua）

if(redis.call('get', KEYS[1]) == ARGV[1]) then
    return redis.call('del', KEYS[1])
end
return 0

參考完整代碼

import cn.hutool.core.lang.UUID;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

public class SimpleRedisLock {
    private String name;
    private StringRedisTemplate stringRedisTemplate;
    private static final String KEY_PREFIX = "lock:";
    private static final String ID_PREFIX = UUID.randomUUID().toString(true) + "-";
    // 加載腳本
    private static final DefaultRedisScript<Long> UNLOCK_SCRIPT;

    static {
        UNLOCK_SCRIPT = new DefaultRedisScript<>();
        // 加載工程resourcecs下的unlock.lua文件
        UNLOCK_SCRIPT.setLocation(new ClassPathResource("unlock.lua"));
        UNLOCK_SCRIPT.setResultType(Long.class);
    }
    public SimpleRedisLock(String name, StringRedisTemplate stringRedisTemplate) {
        this.name = name;
        this.stringRedisTemplate = stringRedisTemplate;
    }
    
    public boolean tryLock(long timeoutSec) {
        // 獲取線程標(biāo)識(shí)
        String threadId = ID_PREFIX +  Thread.currentThread().getId();
        // 獲取鎖
        Boolean success = stringRedisTemplate.opsForValue().setIfAbsent(KEY_PREFIX + name, threadId, timeoutSec, TimeUnit.SECONDS);
        return Boolean.TRUE.equals(success);
    }

    public void unlock() {
        // 調(diào)用Lua腳本
        stringRedisTemplate.execute(UNLOCK_SCRIPT,
                Collections.singletonList(KEY_PREFIX + name),
                ID_PREFIX +  Thread.currentThread().getId());
    }
}

業(yè)務(wù)調(diào)用使用方法

	/**
	* 業(yè)務(wù)service導(dǎo)入
	*/
	@Resource
    private RedissonClient redissonClient;

	/** 
	* 業(yè)務(wù)方法內(nèi)
	*/
	SimpleRedisLock lock = new SimpleRedisLock("order:" + userId, stringRedisTemplate);
	// 創(chuàng)建鎖對(duì)象
    RLock lock = redissonClient.getLock("lock:order:" + userId);
    // 嘗試獲取鎖
    boolean isLock = lock.tryLock();
    // 判斷是否成功
    if (!isLock) {
    	// 獲取失敗
        return Result.fail("不允許重復(fù)下單");
    }
	try {
    	// 需要鎖執(zhí)行的業(yè)務(wù)代碼部分
    } finally {
        // 釋放鎖
        lock.unlock();
    }

當(dāng)前還存在的問題

不可重入：同一個(gè)線程無法多次獲取同一把鎖(線程1在執(zhí)行方法1的時(shí)候調(diào)用了方法2，而方法2也需要鎖)
不可重試：獲取鎖只嘗試一次就返回false，沒有重試機(jī)制；當(dāng)前直接返回結(jié)果。
超時(shí)釋放：鎖超時(shí)釋放雖然可以避免死鎖，但是如果業(yè)務(wù)執(zhí)行耗時(shí)較長，也會(huì)導(dǎo)致鎖釋放，存在安全隱患。
主從一致性：如果redis提供了主從集群，主存同步存在延遲。當(dāng)主結(jié)點(diǎn)宕機(jī)時(shí)，從節(jié)點(diǎn)尚未同步主結(jié)點(diǎn)鎖數(shù)據(jù)，則會(huì)造成鎖失效。

Redisson框架中就實(shí)現(xiàn)了WatchDog（看門狗），加鎖時(shí)沒有指定加鎖時(shí)間時(shí)會(huì)啟用 watchdog 機(jī)制，默認(rèn)加鎖 30 秒，每 10 秒鐘檢查一次，如果存在就重新設(shè)置過期時(shí)間。

RedLock應(yīng)對(duì)主從一致性問題

Redis 的作者提出一種解決方案 Redlock ，基于多個(gè) Redis 節(jié)點(diǎn)，不再需要部署從庫和哨兵實(shí)例，只部署主庫。但主庫要部署多個(gè)，官方推薦至少 5 個(gè)實(shí)例。

流程：

Client先獲取「當(dāng)前時(shí)間戳T1」
Client依次向這 5 個(gè) Redis 實(shí)例發(fā)起加鎖請(qǐng)求（用前面講到的 SET 命令），且每個(gè)請(qǐng)求會(huì)設(shè)置超時(shí)時(shí)間（毫秒級(jí)，要遠(yuǎn)小于鎖的有效時(shí)間），如果某一個(gè)實(shí)例加鎖失敗（包括網(wǎng)絡(luò)超時(shí)、鎖被其它人持有等各種異常情況），就立即向下一個(gè) Redis 實(shí)例申請(qǐng)加鎖
如果Client從 >=3 個(gè)（大多數(shù)）以上 Redis 實(shí)例加鎖成功，則再次獲取「當(dāng)前時(shí)間戳T2」，如果 T2 - T1 < 鎖的過期時(shí)間，此時(shí)，認(rèn)為客戶端加鎖成功，否則認(rèn)為加鎖失敗
加鎖成功，去操作共享資源（例如修改 MySQL 某一行，或發(fā)起一個(gè) API 請(qǐng)求）
加鎖失敗，Client向「全部節(jié)點(diǎn)」發(fā)起釋放鎖請(qǐng)求（前面講到的 Lua 腳本釋放鎖）

為什么要向多個(gè)Redis申請(qǐng)鎖？

向多臺(tái)Redis申請(qǐng)鎖，即使部分服務(wù)器異常宕機(jī)，剩余的Redis加鎖成功，整個(gè)鎖服務(wù)依舊可用。

為什么步驟 3 加鎖成功后，還要計(jì)算加鎖的累計(jì)耗時(shí)？

加鎖操作的針對(duì)的是分布式中的多個(gè)節(jié)點(diǎn)，所以耗時(shí)肯定是比單個(gè)實(shí)例耗時(shí)更，還要考慮網(wǎng)絡(luò)延遲、丟包、超時(shí)等情況發(fā)生，網(wǎng)絡(luò)請(qǐng)求次數(shù)越多，異常的概率越大。
所以即使 N/2+1 個(gè)節(jié)點(diǎn)加鎖成功，但如果加鎖的累計(jì)耗時(shí)已經(jīng)超過了鎖的過期時(shí)間，那么此時(shí)的鎖已經(jīng)沒有意義了

釋放鎖操作為什么要針對(duì)所有結(jié)點(diǎn)？

為了清除干凈所有的鎖。在之前申請(qǐng)鎖的操作過程中，鎖雖然已經(jīng)加在Redis上，但是在獲取結(jié)果的時(shí)候，出現(xiàn)網(wǎng)絡(luò)等方面的問題，導(dǎo)致顯示失敗。所以在釋放鎖的時(shí)候，不管以前有沒有加鎖成功，都要釋放所有節(jié)點(diǎn)相關(guān)鎖。

Zookeeper

ZooKeeper 的數(shù)據(jù)存儲(chǔ)結(jié)構(gòu)就像一棵樹，這棵樹由節(jié)點(diǎn)組成，這種節(jié)點(diǎn)叫做 Znode。

Client嘗試創(chuàng)建一個(gè) znode 節(jié)點(diǎn)，比如/lock，比如Client1先到達(dá)就創(chuàng)建成功了，相當(dāng)于拿到了鎖
其它的客戶端會(huì)創(chuàng)建失?。▃node 已存在），獲取鎖失敗。
Client2可以進(jìn)入一種等待狀態(tài)，等待當(dāng)/lock 節(jié)點(diǎn)被刪除的時(shí)候，ZooKeeper 通過 watch 機(jī)制通知它
持有鎖的Client1訪問共享資源完成后，將 znode 刪掉，鎖釋放掉了
Client2繼續(xù)完成獲取鎖操作，直到獲取到鎖為止

ZooKeeper不需要考慮過期時(shí)間，而是用【臨時(shí)節(jié)點(diǎn)】，Client拿到鎖之后，只要連接不斷，就會(huì)一直持有鎖。即使Client崩潰，相應(yīng)臨時(shí)節(jié)點(diǎn)Znode也會(huì)自動(dòng)刪除，保證了鎖釋放。

Zookeeper 是檢測客戶端是否崩潰

每個(gè)客戶端都與 ZooKeeper 維護(hù)著一個(gè) Session，這個(gè) Session 依賴定期的心跳(heartbeat)來維持。

如果 Zookeeper 長時(shí)間收不到客戶端的心跳，就認(rèn)為這個(gè) Session 過期了，也會(huì)把這個(gè)臨時(shí)節(jié)點(diǎn)刪除。

當(dāng)然這也并不是完美的解決方案

以下場景中Client1和Client2在窗口時(shí)間內(nèi)可能同時(shí)獲得鎖：

Client 1 創(chuàng)建了 znode 節(jié)點(diǎn)/lock，獲得了鎖。
Client 1 進(jìn)入了長時(shí)間的 GC pause。（或者網(wǎng)絡(luò)出現(xiàn)問題、或者 zk 服務(wù)檢測心跳線程出現(xiàn)問題等等）
Client 1 連接到 ZooKeeper 的 Session 過期了。znode 節(jié)點(diǎn)/lock 被自動(dòng)刪除。
Client 2 創(chuàng)建了 znode 節(jié)點(diǎn)/lock，從而獲得了鎖。
Client 1 從 GC pause 中恢復(fù)過來，它仍然認(rèn)為自己持有鎖。

Zookeeper 的優(yōu)點(diǎn)

不需要考慮鎖的過期時(shí)間，使用起來比較方便
watch 機(jī)制，加鎖失敗，可以 watch 等待鎖釋放，實(shí)現(xiàn)樂觀鎖

Zookeeper 的缺點(diǎn)

性能不如 Redis
部署和運(yùn)維成本高
客戶端與 Zookeeper 的長時(shí)間失聯(lián)，鎖被釋放問題

Etcd

Etcd是一個(gè)Go語言實(shí)現(xiàn)的非?？煽康膋v存儲(chǔ)系統(tǒng)，常在分布式系統(tǒng)中存儲(chǔ)著關(guān)鍵的數(shù)據(jù)，通常應(yīng)用在配置中心、服務(wù)發(fā)現(xiàn)與注冊(cè)、分布式鎖等場景。

Etcd特性

Lease機(jī)制：即租約機(jī)制(TTL,Time To Live)，etcd可以為存儲(chǔ)的kv對(duì)設(shè)置租約，當(dāng)租約到期，kv將失效刪除；同時(shí)也支持續(xù)約，keepalive
Revision機(jī)制：每個(gè)key帶有一個(gè)Revision屬性值，etcd每進(jìn)行一次事務(wù)對(duì)應(yīng)的全局Revision值都會(huì)+1，因此每個(gè)key對(duì)應(yīng)的Revision屬性值都是全局唯一的。通過比較Revision的大小就可以知道進(jìn)行寫操作的順序
在實(shí)現(xiàn)分布式鎖時(shí)，多個(gè)程序同時(shí)搶鎖，根據(jù)Revision值大小依次獲得鎖，避免“驚群效應(yīng)”，實(shí)現(xiàn)公平鎖
Prefix機(jī)制：也稱為目錄機(jī)制，可以根據(jù)前綴獲得該目錄下所有的key及其對(duì)應(yīng)的屬性值
Watch機(jī)制：watch支持watch某個(gè)固定的key或者一個(gè)前綴目錄，當(dāng)watch的key發(fā)生變化，客戶端將收到通知

滿足分布式鎖的特性：

租約機(jī)制(Lease)：用于支撐異常情況下的鎖自動(dòng)釋放能力
前綴和 Revision 機(jī)制：用于支撐公平獲取鎖和排隊(duì)等待的能力
監(jiān)聽機(jī)制(Watch)：用于支撐搶鎖能力
集群模式：用于支撐鎖服務(wù)的高可用

func main() {
    config := clientv3.Config{
        Endpoints:   []string{"xxx.xxx.xxx.xxx:2379"},
        DialTimeout: 5 * time.Second,
    }
 
    // 獲取客戶端連接
    client, err := clientv3.New(config)
    if err != nil {
        fmt.Println(err)
        return
    }
 
    // 1. 上鎖（創(chuàng)建租約，自動(dòng)續(xù)租，拿著租約去搶占一個(gè)key ）
    // 用于申請(qǐng)租約
    lease := clientv3.NewLease(client)
 
    // 申請(qǐng)一個(gè)10s的租約
    leaseGrantResp, err := lease.Grant(context.TODO(), 10) //10s
    if err != nil {
        fmt.Println(err)
        return
    }
 
    // 拿到租約的id
    leaseID := leaseGrantResp.ID
 
    // 準(zhǔn)備一個(gè)用于取消續(xù)租的context
    ctx, cancelFunc := context.WithCancel(context.TODO())
 
    // 確保函數(shù)退出后，自動(dòng)續(xù)租會(huì)停止
    defer cancelFunc()
        // 確保函數(shù)退出后，租約會(huì)失效
    defer lease.Revoke(context.TODO(), leaseID)
 
    // 自動(dòng)續(xù)租
    keepRespChan, err := lease.KeepAlive(ctx, leaseID)
    if err != nil {
        fmt.Println(err)
        return
    }
 
    // 處理續(xù)租應(yīng)答的協(xié)程
    go func() {
        select {
        case keepResp := <-keepRespChan:
            if keepRespChan == nil {
                fmt.Println("lease has expired")
                goto END
            } else {
                // 每秒會(huì)續(xù)租一次
                fmt.Println("收到自動(dòng)續(xù)租應(yīng)答", keepResp.ID)
            }
        }
    END:
    }()
 
    // if key 不存在，then設(shè)置它，else搶鎖失敗
    kv := clientv3.NewKV(client)
    // 創(chuàng)建事務(wù)
    txn := kv.Txn(context.TODO())
    // 如果key不存在
    txn.If(clientv3.Compare(clientv3.CreateRevision("/cron/lock/job7"), "=", 0)).
        Then(clientv3.OpPut("/cron/jobs/job7", "", clientv3.WithLease(leaseID))).
        Else(clientv3.OpGet("/cron/jobs/job7")) //如果key存在
 
    // 提交事務(wù)
    txnResp, err := txn.Commit()
    if err != nil {
        fmt.Println(err)
        return
    }
 
    // 判斷是否搶到了鎖
    if !txnResp.Succeeded {
        fmt.Println("鎖被占用了：", string(txnResp.Responses[0].GetResponseRange().Kvs[0].Value))
        return
    }
 
    // 2. 處理業(yè)務(wù)（鎖內(nèi)，很安全）
 
    fmt.Println("處理任務(wù)")
    time.Sleep(5 * time.Second)
 
    // 3. 釋放鎖（取消自動(dòng)續(xù)租，釋放租約）
    // defer會(huì)取消續(xù)租，釋放鎖
}

clientv3提供的concurrency包也實(shí)現(xiàn)了分布式鎖