一、為什么要用分布式鎖

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

本地鎖的問題

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

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

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

二、什么是分布式鎖

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

        舉一個生活中的例子，假設我們去乘坐高鐵，首先要進行檢票進站，但有很多人都想進站。為了避免大家同時擠進去，高鐵站會設置檢票閘機，每次只允許一人檢票通過，當有人檢票進入時，其他人必須等待，直到檢票成功進入后，閘機會再次反鎖。后面的人再嘗試檢票獲取檢票閘機的進入權。這里的檢票閘機就是高鐵站的一把鎖。

 來看下分布式鎖的基本原理，如下圖所示：

我們來分析下上圖的分布式鎖：

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

大白話解釋：所有請求的線程都去同一個地方“占坑”，如果有坑位，就執(zhí)行業(yè)務邏輯，沒有坑位，就需要其他線程釋放“坑位”。這個坑位是所有線程可見的，可以把這個坑位放到 Redis 緩存或者數(shù)據(jù)庫，這篇講的就是如何用 Redis 做“分布式坑位”。

分布式鎖的好處

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

三、Redis 的 SETNX 

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

主要原因在于 Redis 具有以下特點：

高性能：Redis 是一種內存數(shù)據(jù)庫，數(shù)據(jù)存儲在內存中，讀寫速度非?？?，可以快速響應鎖的獲取和釋放請求。

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

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

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

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

SETNX 的語法如下：

SETNX key value

SETNX 的源碼實現(xiàn)比較簡單，其實現(xiàn)過程如下：

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

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

void setnxCommand(client *c) {
    robj *o;
    int nx = c->argc == 3; /* 如果參數(shù)個數(shù)為 3，說明設置 NX（key 不存在才設置） */
    long long expire = 0; /* 默認不設置過期時間 */
    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ù)字轉換為 long long 型數(shù)字 */
    if (getTimeoutFromObjectOrReply(c,c->argv[3],&expire,UNIT_SECONDS)
        != C_OK) return;
    /* 值為空則返回錯誤 */
    if (checkStringLength(c,c->argv[2]->ptr,sdslen(c->argv[2]->ptr)) != C_OK)
        return;
    /* 嘗試將鍵值對插入到數(shù)據(jù)庫中 */
    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);
        /* 設置過期時間 */
        if (expire) setExpire(c->db,c->argv[1],mstime()+expire);
        server.dirty++;
        addReply(c, shared.cone);
    } else {
        decrRefCount(o);
        addReply(c, shared.czero);
    }
}

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

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

SETNX 方案流程圖

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

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

代碼示例

@Service
public class ProductService {
    private final RedisTemplate<String, String> redisTemplate;
    @Autowired
    public ProductService(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }
    /**
     * 扣減庫存
     *
     * @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)) {
                        // 庫存不存在或已過期
                        return false;
                    }
                    int stock = Integer.parseInt(stockStr);
                    if (stock < quantity) {
                        // 庫存不足
                        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() 方法中，首先定義了一個 lockKey，用于對商品庫存進行加鎖。進入 while 循環(huán)后，使用 Redis 的 setIfAbsent() 方法嘗試獲取鎖，如果返回值為 true，則表示成功獲取鎖。在成功獲取鎖后，再從 Redis 中獲取商品庫存，判斷庫存是否充足，如果充足則扣減庫存并返回 true；否則直接返回 false。最后，在 finally 塊中刪除加鎖的 key。

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

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

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

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

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

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

SETNXEX key seconds value

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

源碼分析：

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

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

Redis 在底層使用 SETNX 和 SETEX 命令實現(xiàn) SETNXEX 命令，它的 C 語言實現(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);
}

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

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

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

SETNXEX 方案流程圖

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

• 客戶端向 Redis 服務器發(fā)送申請鎖的請求，請求內容包括鎖的名稱和過期時間；
• Redis 服務器接收到請求后進行處理，使用 SETNXEX 命令將鎖鍵和值寫入到 Redis 的鍵值對數(shù)據(jù)庫中，并設置過期時間；
• 如果 SETNXEX 返回值是 1，則客戶端成功獲取到鎖，執(zhí)行業(yè)務邏輯并在完成后釋放鎖；
• 如果 SETNXEX 返回值是 0，則客戶端未獲取到鎖，等待一段時間后重試獲取鎖；
• 客戶端在釋放鎖時，先確認自己是否持有該鎖，如果持有則使用 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;
    /**
     * 扣減庫存
     * @param productId 商品ID
     * @param num 扣減數(shù)量
     */
    public boolean reduceStock(Long productId, int num) {
        // 構造鎖的key
        String lockKey = LOCK_KEY_PREFIX + productId;
        // 構造鎖的value，這里使用當前線程的ID
        String lockValue = String.valueOf(Thread.currentThread().getId());
        try {
            // 嘗試獲取鎖，設置過期時間為10秒
            Boolean locked = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 10L, TimeUnit.SECONDS);
            if (!locked) {
                // 獲取鎖失敗，等待10秒后重新嘗試獲取鎖
                Thread.sleep(10000);
                return reduceStock(productId, num);
            }
            // 獲取鎖成功，執(zhí)行扣減庫存代碼
            // TODO ... 扣減庫存代碼
            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);
            }
        }
    }
}

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

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

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

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

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

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

如上圖所示，當有兩個線程同時請求獲取鎖時，執(zhí)行流程如下：

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

 代碼示例

@Component
public class StockService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    /**
     * 扣減庫存
     *
     * @param stockId 庫存 ID
     * @param num 扣減數(shù)量
     * @return 是否扣減成功
     */
    public boolean reduceStock(String stockId, int num) throws InterruptedException {
        // 構造鎖的名稱
        String lockKey = "stock_lock_" + stockId;
        // 獲取當前線程 ID
        String threadId = String.valueOf(Thread.currentThread().getId());
        try {
            // 使用 SETNXEX 命令申請鎖
            Boolean lockResult = redisTemplate.opsForValue().setIfAbsent(lockKey, threadId, 30, TimeUnit.SECONDS);
            if (!lockResult) {
                // 如果獲取鎖失敗，則等待一段時間后重試
                Thread.sleep(10000);
                return reduceStock(stockId, num);
            }
            // 設置鎖的過期時間為 30 秒，并啟動一個定時任務續(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è)務邏輯，例如扣減庫存
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 釋放鎖
            if (threadId.equals(redisTemplate.opsForValue().get(lockKey))) {
                redisTemplate.delete(lockKey);
            }
        }
        return false;
    }
}

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

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

方案的缺陷

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

針對上述問題的解決方案 ，下篇見。

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

最新評論