一、為什么要用分布式鎖

在分布式系統(tǒng)中，為了保證多個(gè)進(jìn)程或線程之間的數(shù)據(jù)一致性和正確性，需要使用鎖來(lái)實(shí)現(xiàn)互斥訪問(wèn)共享資源。然而，使用本地鎖在分布式系統(tǒng)中存在問(wèn)題。

本地鎖的問(wèn)題

• 無(wú)法保證全局唯一性：本地鎖只在本地生效，每個(gè)節(jié)點(diǎn)都有自己的一份數(shù)據(jù)，所以不能保證在整個(gè)集群中全局唯一。
• 無(wú)法協(xié)調(diào)多個(gè)節(jié)點(diǎn)之間的鎖：在分布式系統(tǒng)中，多個(gè)節(jié)點(diǎn)同時(shí)訪問(wèn)同一個(gè)資源時(shí)，需要協(xié)調(diào)各個(gè)節(jié)點(diǎn)之間的鎖，保證資源的互斥訪問(wèn)。而本地鎖只能鎖住當(dāng)前節(jié)點(diǎn)的資源，無(wú)法協(xié)調(diào)各個(gè)節(jié)點(diǎn)之間的鎖。
• 可能會(huì)出現(xiàn)死鎖和鎖競(jìng)爭(zhēng)：由于分布式系統(tǒng)中很難保證各個(gè)節(jié)點(diǎn)的鎖同步，因此容易導(dǎo)致死鎖和鎖競(jìng)爭(zhēng)等問(wèn)題。
• 性能問(wèn)題：在分布式系統(tǒng)中，為了保證多個(gè)節(jié)點(diǎn)之間的鎖同步，通常需要進(jìn)行大量的網(wǎng)絡(luò)通信，這會(huì)影響系統(tǒng)的性能。

        比如商品服務(wù)A和服務(wù)B同時(shí)獲取到庫(kù)存數(shù)量為10的商品信息。商品服務(wù)A和服務(wù)B同時(shí)進(jìn)行扣減庫(kù)存操作，分別將庫(kù)存數(shù)量減少了1。商品服務(wù)A和服務(wù)B均修改了庫(kù)存數(shù)量為9，然后將數(shù)據(jù)寫(xiě)入數(shù)據(jù)庫(kù)中。
        由于使用本地鎖，商品服務(wù)A和服務(wù)B之間沒(méi)有進(jìn)行協(xié)調(diào)，因此就會(huì)出現(xiàn)數(shù)據(jù)不一致的問(wèn)題?？赡艹霈F(xiàn)以下情況：商品服務(wù)A先將庫(kù)存數(shù)量9寫(xiě)入數(shù)據(jù)庫(kù)，然后商品服務(wù)B也將庫(kù)存數(shù)量9寫(xiě)入數(shù)據(jù)庫(kù)，商品服務(wù)B先將庫(kù)存數(shù)量9寫(xiě)入數(shù)據(jù)庫(kù)，然后商品服務(wù)A也將庫(kù)存數(shù)量9寫(xiě)入數(shù)據(jù)庫(kù)，結(jié)果，整個(gè)系統(tǒng)中庫(kù)存數(shù)量實(shí)際只完成了一次扣減，最終庫(kù)存數(shù)量賣(mài)出2份后，還剩下9，出現(xiàn)了數(shù)據(jù)不一致的情況。

        相比之下，分布式鎖可以解決上述問(wèn)題。分布式鎖可以在多個(gè)節(jié)點(diǎn)之間協(xié)調(diào)鎖的使用，確保在分布式系統(tǒng)中多個(gè)進(jìn)程或線程互斥訪問(wèn)共享資源，并保證了全局唯一性，避免了死鎖和鎖競(jìng)爭(zhēng)問(wèn)題，同時(shí)也能夠提高系統(tǒng)的吞吐量和性能。

二、什么是分布式鎖

        分布式鎖是一種用于在分布式系統(tǒng)中協(xié)調(diào)多個(gè)進(jìn)程或線程之間對(duì)共享資源的互斥訪問(wèn)的機(jī)制。在分布式系統(tǒng)中，由于各個(gè)節(jié)點(diǎn)之間沒(méi)有共享內(nèi)存，因此無(wú)法使用傳統(tǒng)的本地鎖機(jī)制來(lái)實(shí)現(xiàn)進(jìn)程或線程的同步，所以需要使用分布式鎖來(lái)解決這個(gè)問(wèn)題。

        舉一個(gè)生活中的例子，假設(shè)我們?nèi)コ俗哞F，首先要進(jìn)行檢票進(jìn)站，但有很多人都想進(jìn)站。為了避免大家同時(shí)擠進(jìn)去，高鐵站會(huì)設(shè)置檢票閘機(jī)，每次只允許一人檢票通過(guò)，當(dāng)有人檢票進(jìn)入時(shí)，其他人必須等待，直到檢票成功進(jìn)入后，閘機(jī)會(huì)再次反鎖。后面的人再嘗試檢票獲取檢票閘機(jī)的進(jìn)入權(quán)。這里的檢票閘機(jī)就是高鐵站的一把鎖。

 來(lái)看下分布式鎖的基本原理，如下圖所示：

我們來(lái)分析下上圖的分布式鎖：

• 1.前端將 100個(gè) 的高并發(fā)請(qǐng)求轉(zhuǎn)發(fā)兩個(gè)商品微服務(wù)。
• 2.每個(gè)微服務(wù)處理 50個(gè)請(qǐng)求。
• 3.每個(gè)處理請(qǐng)求的線程在執(zhí)行業(yè)務(wù)之前，需要先搶占鎖?？梢岳斫鉃?ldquo;占坑”。
• 4.獲取到鎖的線程在執(zhí)行完業(yè)務(wù)后，釋放鎖?？梢岳斫鉃?ldquo;釋放坑位”。
• 5.未獲取到的線程需要等待鎖釋放。
• 6.釋放鎖后，其他線程搶占鎖。
• 7.重復(fù)執(zhí)行步驟 4、5、6。

大白話解釋?zhuān)核姓?qǐng)求的線程都去同一個(gè)地方“占坑”，如果有坑位，就執(zhí)行業(yè)務(wù)邏輯，沒(méi)有坑位，就需要其他線程釋放“坑位”。這個(gè)坑位是所有線程可見(jiàn)的，可以把這個(gè)坑位放到 Redis 緩存或者數(shù)據(jù)庫(kù)，這篇講的就是如何用 Redis 做“分布式坑位”。

分布式鎖的好處

• 避免重復(fù)操作：如果多個(gè)進(jìn)程或線程同時(shí)嘗試對(duì)同一個(gè)資源進(jìn)行操作，就會(huì)導(dǎo)致重復(fù)操作和數(shù)據(jù)的不一致。使用分布式鎖可以確保只有一個(gè)進(jìn)程或線程能夠獲得鎖，從而避免了重復(fù)操作。
• 防止競(jìng)態(tài)條件：在并發(fā)環(huán)境下，多個(gè)進(jìn)程或線程同時(shí)讀寫(xiě)共享資源時(shí)，容易引發(fā)競(jìng)態(tài)條件（Race Condition）。使用分布式鎖可以保證同一時(shí)間只有一個(gè)進(jìn)程或線程能夠訪問(wèn)共享資源，從而避免了競(jìng)態(tài)條件。
• 提高系統(tǒng)吞吐量：使用分布式鎖可以避免多個(gè)進(jìn)程或線程同時(shí)競(jìng)爭(zhēng)共享資源，從而有效地提高系統(tǒng)的吞吐量和性能。

三、Redis 的 SETNX 

        為了使用分布式鎖，需要我們找到一個(gè)可靠的第三方中間件。Redis剛好可以用來(lái)作為分布式鎖的提供者。

主要原因在于 Redis 具有以下特點(diǎn)：

高性能：Redis 是一種內(nèi)存數(shù)據(jù)庫(kù)，數(shù)據(jù)存儲(chǔ)在內(nèi)存中，讀寫(xiě)速度非?？欤梢钥焖夙憫?yīng)鎖的獲取和釋放請(qǐng)求。

原子操作：Redis 支持原子操作，例如 SETNX（SET if Not eXists）命令可以實(shí)現(xiàn)“只有在鍵不存在時(shí)設(shè)置鍵值”的操作，可以保證同時(shí)只會(huì)有一個(gè)客戶端成功獲取到鎖，并且避免了因?yàn)閳?zhí)行多個(gè)操作而導(dǎo)致的競(jìng)態(tài)條件問(wèn)題。

可靠性高：Redis 可以進(jìn)行主從復(fù)制和持久化備份等操作，可以確保即使出現(xiàn)網(wǎng)絡(luò)中斷或 Redis 實(shí)例宕機(jī)的情況，也可以保證分布式鎖的正確性和一致性。

        基于以上特點(diǎn)，我們可以使用 Redis 來(lái)實(shí)現(xiàn)分布式鎖的機(jī)制。具體做法是通過(guò) SETNX 命令在 Redis 中創(chuàng)建一個(gè)鍵值對(duì)作為鎖，當(dāng)有其他客戶端嘗試獲取鎖時(shí)，如果該鍵值對(duì)已經(jīng)存在，則表示鎖已經(jīng)被其他客戶端持有；反之，則表示當(dāng)前客戶端獲取鎖成功。

        Redis 中的 SETNX 命令用于設(shè)置指定鍵的值，但是只有在該鍵不存在時(shí)才進(jìn)行設(shè)置。如果該鍵已經(jīng)存在，則 SETNX 命令不會(huì)對(duì)其進(jìn)行任何操作。

SETNX 的語(yǔ)法如下：

SETNX key value

SETNX 的源碼實(shí)現(xiàn)比較簡(jiǎn)單，其實(shí)現(xiàn)過(guò)程如下：

• 檢查給定鍵是否在 Redis 中已經(jīng)存在，如果存在則返回 0，不對(duì) key 的值進(jìn)行修改。
• 如果 key 不存在，則將 key 的值設(shè)置為 value，并返回 1。

SETNX 命令的 C 語(yǔ)言實(shí)現(xiàn)如下：

void setnxCommand(client *c) {
    robj *o;
    int nx = c->argc == 3; /* 如果參數(shù)個(gè)數(shù)為 3，說(shuō)明設(shè)置 NX（key 不存在才設(shè)置） */
    long long expire = 0; /* 默認(rèn)不設(shè)置過(guò)期時(shí)間 */
    int retval;
    if (nx) {
        /* NX 模式下檢查 key 是否已經(jīng)存在 */
        if (lookupKeyWrite(c->db,c->argv[1]) != NULL) {
            addReply(c,shared.czero);
            return;
        }
    } else {
        /* XX 模式下檢查 key 是否不存在 */
        if (lookupKeyWrite(c->db,c->argv[1]) == NULL) {
            addReply(c,shared.czero);
            return;
        }
    }
    /* 嘗試將字符串型或整型數(shù)字轉(zhuǎn)換為 long long 型數(shù)字 */
    if (getTimeoutFromObjectOrReply(c,c->argv[3],&expire,UNIT_SECONDS)
        != C_OK) return;
    /* 值為空則返回錯(cuò)誤 */
    if (checkStringLength(c,c->argv[2]->ptr,sdslen(c->argv[2]->ptr)) != C_OK)
        return;
    /* 嘗試將鍵值對(duì)插入到數(shù)據(jù)庫(kù)中 */
    o = createStringObject(c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
    retval = dictAdd(c->db->dict,c->argv[1],o);
    if (retval == DICT_OK) {
        incrRefCount(o);
        /* 設(shè)置過(guò)期時(shí)間 */
        if (expire) setExpire(c->db,c->argv[1],mstime()+expire);
        server.dirty++;
        addReply(c, shared.cone);
    } else {
        decrRefCount(o);
        addReply(c, shared.czero);
    }
}

從源碼實(shí)現(xiàn)可以看出，SETNX 命令的執(zhí)行過(guò)程非?？焖?，由于 Redis 存儲(chǔ)數(shù)據(jù)是采用字典結(jié)構(gòu)，在判斷 key 是否存在時(shí)可以達(dá)到 O(1) 的時(shí)間復(fù)雜度，因此 SETNX 命令的性能很高。

四、使用Redis SETNX 實(shí)現(xiàn)分布式鎖的方案

SETNX 方案流程圖

如上圖所示，使用 Redis 的 SETNX 命令來(lái)實(shí)現(xiàn)分布式鎖的過(guò)程如下：

• 客戶端嘗試獲取鎖，以鎖的名稱為鍵名，將客戶端唯一標(biāo)識(shí)（如 UUID）作為鍵值，調(diào)用 Redis 的 SETNX 命令。
• 如果 Redis 中不存在該鍵，即返回的結(jié)果是 1，則表示鎖獲取成功，客戶端可以進(jìn)入臨界區(qū)進(jìn)行操作。
• 如果 Redis 中已經(jīng)存在該鍵，即返回的結(jié)果是 0，則表示鎖已經(jīng)被其他客戶端持有，當(dāng)前客戶端沒(méi)有獲取到鎖，需要等待或重試。
• 當(dāng)客戶端完成操作后，調(diào)用 Redis 的 DEL 命令來(lái)釋放鎖，刪除鍵。

代碼示例

@Service
public class ProductService {
    private final RedisTemplate<String, String> redisTemplate;
    @Autowired
    public ProductService(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    /**
     * 扣減庫(kù)存
     *
     * @param productId 商品ID
     * @param quantity  數(shù)量
     * @return true 扣減成功，false 扣減失敗
     */
    public boolean decreaseStock(String productId, int quantity) {
        String lockKey = "stock_" + productId;
        while (true) {
            Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, "", 10, TimeUnit.SECONDS);
            if (lockResult != null && lockResult) {
                try {
                    String stockKey = "product_" + productId;
                    String stockStr = redisTemplate.opsForValue().get(stockKey);
                    if (StringUtils.isEmpty(stockStr)) {
                        // 庫(kù)存不存在或已過(guò)期
                        return false;
                    }
                    int stock = Integer.parseInt(stockStr);
                    if (stock < quantity) {
                        // 庫(kù)存不足
                        return false;
                    }
                    int newStock = stock - quantity;
                    redisTemplate.opsForValue().set(stockKey, String.valueOf(newStock));
                    return true;
                } finally {
                    redisTemplate.delete(lockKey);
                }
            }
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

         在 decreaseStock() 方法中，首先定義了一個(gè) lockKey，用于對(duì)商品庫(kù)存進(jìn)行加鎖。進(jìn)入 while 循環(huán)后，使用 Redis 的 setIfAbsent() 方法嘗試獲取鎖，如果返回值為 true，則表示成功獲取鎖。在成功獲取鎖后，再?gòu)?Redis 中獲取商品庫(kù)存，判斷庫(kù)存是否充足，如果充足則扣減庫(kù)存并返回 true；否則直接返回 false。最后，在 finally 塊中刪除加鎖的 key。

        如果獲取鎖失敗，則等待 10 秒后再次嘗試獲取鎖，直到獲取成功為止。

SETNX 實(shí)現(xiàn)分布式鎖的缺陷

使用 Redis SETNX 實(shí)現(xiàn)分布式鎖可能存在以下缺陷：

• 競(jìng)爭(zhēng)激烈時(shí)容易出現(xiàn)死鎖情況。這種情況可以通過(guò)在加鎖時(shí)設(shè)置一個(gè)唯一標(biāo)識(shí)符（例如 UUID），釋放鎖時(shí)檢查標(biāo)識(shí)符是否匹配來(lái)避免。
• 鎖的釋放不及時(shí)?？梢酝ㄟ^(guò)在加鎖時(shí)設(shè)置一個(gè)過(guò)期時(shí)間，確保即使客戶端意外宕機(jī)，鎖也會(huì)在一定時(shí)間后自動(dòng)釋放。
• 客戶端誤刪其他客戶端的鎖。這種情況可以通過(guò)為每個(gè)客戶端生成一個(gè)唯一標(biāo)識(shí)符，加鎖時(shí)將標(biāo)識(shí)符寫(xiě)入 Redis，釋放鎖時(shí)檢查標(biāo)識(shí)符是否匹配來(lái)避免。

五、Redis SETNX優(yōu)化方案 SETNXEX

針對(duì)使用 Redis SETNX 實(shí)現(xiàn)分布式鎖可能出現(xiàn)死鎖的情況,，可以使用SETNXEX進(jìn)行優(yōu)化，Redis SETNXEX 命令是 Redis 提供的一個(gè)原子操作指令，用于設(shè)置一個(gè)有過(guò)期時(shí)間的字符串類(lèi)型鍵值對(duì)，當(dāng)且僅當(dāng)該鍵不存在時(shí)設(shè)置成功，返回 1，否則返回 0。SETNXEX 命令的語(yǔ)法如下：

SETNXEX key seconds value

其中，key 是鍵名；seconds 為整數(shù)，表示鍵值對(duì)的過(guò)期時(shí)間（單位為秒）；value 是鍵值。

源碼分析：

實(shí)現(xiàn) SETNXEX 命令的關(guān)鍵在于如何保證該操作的原子性和一致性。其實(shí)現(xiàn)過(guò)程如下：

• 如果鍵 key 已經(jīng)存在，則返回 0。如果鍵 key 不存在，則將鍵 key 的值設(shè)置為 value，并設(shè)置過(guò)期時(shí)間為 seconds 秒。
• 如果設(shè)置成功，則返回 1；否則，返回 0。

Redis 在底層使用 SETNX 和 SETEX 命令實(shí)現(xiàn) SETNXEX 命令，它的 C 語(yǔ)言實(shí)現(xiàn)代碼如下：

void setnxexCommand(client *c) {
    robj *key = c->argv[1], *val = c->argv[3];
    long long expire = strtoll(c->argv[2]->ptr,NULL,10);
    expire *= 1000;
    if (getExpire(c,key) != -1) {
        addReply(c, shared.czero);
        return;
    }
    setKey(c,c->db,key,val,LOOKUP_NOTOUCH|LOOKUP_EX|LOOKUP_NX,0,0,NULL);
    if (c->flags & CLIENT_MULTI) {
        addReply(c, shared.cone);
        return;
    }
    server.dirty++;
    if (expire) setExpire(c,c->db,key,mstime()+expire);
    addReply(c, shared.cone);
    notifyKeyspaceEvent(NOTIFY_STRING,"set",key,c->db->id);
    notifyKeyspaceEvent(NOTIFY_GENERIC,"expire",key,c->db->id);
}

        在這個(gè)代碼中，首先從客戶端傳來(lái)的參數(shù)中獲取 key、value 和 expire 值，并通過(guò) getExpire 函數(shù)檢查鍵是否已經(jīng)存在。如果已經(jīng)存在，則返回 0；否則，調(diào)用 setKey 函數(shù)將鍵值對(duì)設(shè)置為 value，并加上過(guò)期時(shí)間 expire。當(dāng)然，這里的過(guò)期時(shí)間是以毫秒為單位的，需要轉(zhuǎn)換成 Redis 的標(biāo)準(zhǔn)格式。最后，通過(guò) addReply 函數(shù)向客戶端發(fā)送成功的響應(yīng)消息，并通過(guò) notifyKeyspaceEvent 函數(shù)發(fā)送鍵空間通知。 

        需要注意的是，雖然 SETNXEX 被稱為“原子操作”，但實(shí)際上在高并發(fā)場(chǎng)景下，SETNX 和 SETEX 操作之間可能會(huì)發(fā)生競(jìng)爭(zhēng)問(wèn)題，導(dǎo)致 SETNX 和 SETEX 操作不具備原子性。如果在分布式場(chǎng)景下需要保證 SETNXEX 的原子性，還需要使用分布式鎖等機(jī)制來(lái)避免競(jìng)爭(zhēng)問(wèn)題。因此，在使用 SETNXEX 命令時(shí)，需要根據(jù)具體情況，評(píng)估其安全性和可靠性，采用合適的解決方案。

六、使用Redis SETNXEX 實(shí)現(xiàn)分布式鎖的方案

SETNXEX 方案流程圖

如上圖所示，使用 Redis 的 SETNXEX 命令來(lái)實(shí)現(xiàn)分布式鎖的過(guò)程如下

• 客戶端向 Redis 服務(wù)器發(fā)送申請(qǐng)鎖的請(qǐng)求，請(qǐng)求內(nèi)容包括鎖的名稱和過(guò)期時(shí)間；
• Redis 服務(wù)器接收到請(qǐng)求后進(jìn)行處理，使用 SETNXEX 命令將鎖鍵和值寫(xiě)入到 Redis 的鍵值對(duì)數(shù)據(jù)庫(kù)中，并設(shè)置過(guò)期時(shí)間；
• 如果 SETNXEX 返回值是 1，則客戶端成功獲取到鎖，執(zhí)行業(yè)務(wù)邏輯并在完成后釋放鎖；
• 如果 SETNXEX 返回值是 0，則客戶端未獲取到鎖，等待一段時(shí)間后重試獲取鎖；
• 客戶端在釋放鎖時(shí)，先確認(rèn)自己是否持有該鎖，如果持有則使用 DEL 命令刪除鎖。

 代碼示例

@Component
public class StockService {
    private final Logger logger = LoggerFactory.getLogger(StockService.class);
    private final String LOCK_KEY_PREFIX = "stock:lock:";
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 扣減庫(kù)存
     * @param productId 商品ID
     * @param num 扣減數(shù)量
     */
    public boolean reduceStock(Long productId, int num) {
        // 構(gòu)造鎖的key
        String lockKey = LOCK_KEY_PREFIX + productId;
        // 構(gòu)造鎖的value，這里使用當(dāng)前線程的ID
        String lockValue = String.valueOf(Thread.currentThread().getId());
        try {
            // 嘗試獲取鎖，設(shè)置過(guò)期時(shí)間為10秒
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 10L, TimeUnit.SECONDS);
            if (!locked) {
                // 獲取鎖失敗，等待10秒后重新嘗試獲取鎖
                Thread.sleep(10000);
                return reduceStock(productId, num);
            }
            // 獲取鎖成功，執(zhí)行扣減庫(kù)存代碼
            // TODO ... 扣減庫(kù)存代碼
            return true;
        } catch (InterruptedException e) {
            logger.error("Failed to acquire stock lock", e);
            return false;
        } finally {
            // 釋放鎖
            if (lockValue.equals(redisTemplate.opsForValue().get(lockKey))) {
                redisTemplate.delete(lockKey);
            }
        }
    }
}

        上述代碼中，首先構(gòu)造了鎖的key和value，然后使用 RedisTemplate 的 setIfAbsent 方法嘗試獲取鎖。如果獲取鎖失敗，則線程會(huì)等待10秒后重新嘗試獲取鎖，直到獲取鎖成功為止。如果獲取鎖成功，則執(zhí)行扣減庫(kù)存的業(yè)務(wù)邏輯，待操作完成后釋放鎖。

SETNXEX 實(shí)現(xiàn)分布式鎖的缺陷 

• 非阻塞式獲取鎖：使用 SETNXEX 命令獲取鎖時(shí)，如果鎖已經(jīng)被其他客戶端持有，則 SETNXEX 操作會(huì)失敗，并返回 0。在這種情況下，當(dāng)前客戶端可以繼續(xù)執(zhí)行其他操作，而無(wú)需等待鎖的釋放。這種非阻塞式獲取鎖的策略可能會(huì)導(dǎo)致死鎖和數(shù)據(jù)競(jìng)爭(zhēng)問(wèn)題，對(duì)系統(tǒng)的可靠性和正確性產(chǎn)生負(fù)面影響。
• 鎖過(guò)期機(jī)制：使用 SETNXEX 命令設(shè)置鎖時(shí)需要指定過(guò)期時(shí)間，如果鎖的持有者在過(guò)期時(shí)間內(nèi)沒(méi)有完成操作，鎖會(huì)自動(dòng)釋放，從而導(dǎo)致其他客戶端可以獲取該鎖。但是，如果在鎖過(guò)期前持有鎖的客戶端還未完成操作，那么其他客戶端就有可能獲取到該鎖，從而導(dǎo)致多個(gè)客戶端同時(shí)修改同一個(gè)資源，引發(fā)數(shù)據(jù)競(jìng)爭(zhēng)問(wèn)題。
• 非可重入鎖：使用 SETNXEX 命令獲取鎖時(shí)，不能重復(fù)獲取已經(jīng)持有的鎖，否則會(huì)導(dǎo)致死鎖問(wèn)題。因此，SETNXEX 命令實(shí)現(xiàn)的分布式鎖是一種非可重入鎖，不能滿足某些場(chǎng)景下的需求。
• 非原子性操作：在分布式環(huán)境中，如果在比較鎖的值和刪除鎖之間，有其他客戶端獲取了鎖并修改了數(shù)據(jù)，那么該鎖的值可能已經(jīng)被改變，導(dǎo)致誤刪鎖或刪除其他客戶端持有的鎖，引發(fā)數(shù)據(jù)競(jìng)爭(zhēng)問(wèn)題。

七、Redis SETNXEX 實(shí)現(xiàn)分布式鎖缺陷的優(yōu)化方案

        針對(duì)SETNXEX鎖過(guò)期問(wèn)題的優(yōu)化方案：在執(zhí)行業(yè)務(wù)邏輯前，我們?cè)O(shè)置鎖的過(guò)期時(shí)間為 30 秒，并啟動(dòng)一個(gè)定時(shí)任務(wù)續(xù)租鎖，以防止鎖因長(zhǎng)時(shí)間持有而超時(shí)失效。
在 finally 塊中釋放鎖，首先判斷當(dāng)前線程是否持有該鎖，如果是則刪除該鎖。

SETNXEX 優(yōu)化方案流程圖

如上圖所示，當(dāng)有兩個(gè)線程同時(shí)請(qǐng)求獲取鎖時(shí)，執(zhí)行流程如下：

• 線程 A 和線程 B 同時(shí)想要獲取名為 lock 的鎖。
• 線程 A 先到達(dá) Redis 中，執(zhí)行 SETNXEX lock 30 命令嘗試獲取鎖。如果返回值為 1，則說(shuō)明線程 A 成功獲取到鎖，進(jìn)入業(yè)務(wù)邏輯執(zhí)行階段。
• 線程 B 到達(dá) Redis 中，執(zhí)行 SETNXEX lock 30 命令嘗試獲取鎖。由于線程 A 已經(jīng)獲取了鎖且正在執(zhí)行業(yè)務(wù)邏輯，因此線程 B 獲取鎖失敗，需要等待一段時(shí)間后重新嘗試獲取。
• 在獲取鎖失敗后，線程 B 進(jìn)入等待狀態(tài)，等待一段時(shí)間后再次嘗試獲取鎖。
• 在線程 A 執(zhí)行業(yè)務(wù)邏輯前，將鎖的過(guò)期時(shí)間設(shè)置為 30 秒，并開(kāi)啟一個(gè)定時(shí)任務(wù)每隔 10 秒續(xù)租一次鎖，以保證在業(yè)務(wù)邏輯執(zhí)行期間鎖不會(huì)超時(shí)失效。
• 在 finally 塊中釋放鎖，首先判斷當(dāng)前線程是否持有該鎖，如果是則刪除該鎖。如果線程 A 的業(yè)務(wù)邏輯執(zhí)行完畢，則釋放鎖；如果線程 B 成功獲取到鎖，并在后面的某個(gè)時(shí)間釋放了鎖，之后的請(qǐng)求會(huì)有機(jī)會(huì)獲取到鎖。

 代碼示例

@Component
public class StockService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    /**
     * 扣減庫(kù)存
     *
     * @param stockId 庫(kù)存 ID
     * @param num 扣減數(shù)量
     * @return 是否扣減成功
     */
    public boolean reduceStock(String stockId, int num) throws InterruptedException {
        // 構(gòu)造鎖的名稱
        String lockKey = "stock_lock_" + stockId;
        // 獲取當(dāng)前線程 ID
        String threadId = String.valueOf(Thread.currentThread().getId());
        try {
            // 使用 SETNXEX 命令申請(qǐng)鎖
            Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, threadId, 30, TimeUnit.SECONDS);
            if (!lockResult) {
                // 如果獲取鎖失敗，則等待一段時(shí)間后重試
                Thread.sleep(10000);
                return reduceStock(stockId, num);
            }
            // 設(shè)置鎖的過(guò)期時(shí)間為 30 秒，并啟動(dòng)一個(gè)定時(shí)任務(wù)續(xù)租鎖
            ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                Long expireResult = redisTemplate.getExpire(lockKey);
                if (expireResult < 10) {
                    redisTemplate.expire(lockKey, expireResult + 10, TimeUnit.SECONDS);
                }
            }, 10, 10, TimeUnit.SECONDS);
            // TODO：執(zhí)行業(yè)務(wù)邏輯，例如扣減庫(kù)存
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 釋放鎖
            if (threadId.equals(redisTemplate.opsForValue().get(lockKey))) {
                redisTemplate.delete(lockKey);
            }
        }
        return false;
    }
}

        在上面的代碼示例中，我們使用了 redisTemplate 的 opsForValue().setIfAbsent() 方法來(lái)申請(qǐng)鎖。如果獲取鎖失敗，則等待 10 秒后重新嘗試獲取鎖。在獲取鎖成功后，我們?cè)O(shè)置鎖的過(guò)期時(shí)間為 30 秒，并啟動(dòng)一個(gè)定時(shí)任務(wù)續(xù)租鎖，以防止鎖因長(zhǎng)時(shí)間持有而超時(shí)失效。在執(zhí)行完業(yè)務(wù)邏輯后，返回 true 表示扣減成功。

        在釋放鎖時(shí)，我們首先通過(guò) redisTemplate.opsForValue().get(lockKey) 方法獲取當(dāng)前持有鎖的線程 ID，然后判斷當(dāng)前線程是否持有該鎖，如果是則刪除該鎖。這里使用了 redisTemplate.delete() 方法來(lái)刪除鎖。

方案的缺陷

• 可重入性問(wèn)題：如果一個(gè)線程已經(jīng)獲取了鎖，再次嘗試獲取鎖時(shí)會(huì)失敗，此時(shí)線程會(huì)進(jìn)入等待狀態(tài)。但是如果在等待期間，持有鎖的線程又嘗試獲取鎖，則會(huì)導(dǎo)致可重入性問(wèn)題。
• 死鎖問(wèn)題：如果持有鎖的線程異常退出或者業(yè)務(wù)執(zhí)行過(guò)長(zhǎng)時(shí)間不釋放鎖，那么其他線程就會(huì)一直等待該鎖，從而導(dǎo)致死鎖問(wèn)題。
• 定時(shí)任務(wù)續(xù)租問(wèn)題：雖然定時(shí)任務(wù)可以續(xù)租鎖，但是無(wú)法保證定時(shí)任務(wù)一定能夠執(zhí)行成功。如果定時(shí)任務(wù)執(zhí)行失敗，那么就會(huì)出現(xiàn)鎖過(guò)期但沒(méi)有自動(dòng)釋放的情況。
• 解鎖問(wèn)題：當(dāng)線程在 finally 塊中釋放鎖時(shí)，首先需要判斷當(dāng)前線程是否持有該鎖。但是如果線程在業(yè)務(wù)執(zhí)行期間被重新創(chuàng)建并獲取了同一把鎖，那么該判斷就會(huì)失效，從而導(dǎo)致無(wú)法正確釋放鎖的問(wèn)題 。

針對(duì)上述問(wèn)題的解決方案 ，下篇見(jiàn)。

到此這篇關(guān)于從原理到實(shí)踐分析 Redis 分布式鎖的多種實(shí)現(xiàn)方案的文章就介紹到這了,更多相關(guān) Redis 分布式鎖內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論