在一些分散式環境下、多執行緒併發程式設計中，如果對同一資源進行讀寫操作，避免不了的一個就是資源競爭問題，通過引入分散式鎖這一概念，可以解決資料一致性問題。

作者簡介：五月君，Nodejs Developer，網認證作者，熱愛技術、喜歡分享的 90 後青年，歡迎關注 Nodejs技術棧 和 Github 開源專案 www.nodejs.red

認識執行緒、程式、分散式鎖

執行緒鎖：單執行緒程式設計模式下請求是順序的，一個好處是不需要考慮執行緒安全、資源競爭問題，因此當你進行 Node.js 程式設計時，也不會去考慮執行緒安全問題。那麼多執行緒程式設計模式下，例如 Java 你可能很熟悉一個詞 synchronized，通常也是 Java 中解決併發程式設計最簡單的一種方式，synchronized 可以保證在同一時刻僅有一個執行緒去執行某個方法或某塊程式碼。

程式鎖：一個服務部署於一臺伺服器，同時開啟多個程式，Node.js 程式設計中為了利用作業系統資源，根據 CPU 的核心數可以開啟多程式模式，這個時候如果對一個共享資源操作還是會遇到資源競爭問題，另外每一個程式都是相互獨立的，擁有自己獨立的記憶體空間。關於程式鎖通過 Java 中的 synchronized 也很難去解決，synchronized 僅侷限於在同一個 JVM 中有效。

分散式鎖：一個服務無論是單執行緒還是多程式模式，當多機部署、處於分散式環境下對同一共享資源進行操作還是會面臨同樣的問題。此時就要去引入一個概念分散式鎖。如下圖所示，由於先讀資料在通過業務邏輯修改之後進行 SET 操作，這並不是一個原子操作，當多個客戶端對同一資源進行先讀後寫操作就會引發併發問題，這時就要引入分散式鎖去解決，通常也是一個很廣泛的解決方案。

基於 Redis 的分散式鎖實現思路

實現分散式鎖的方式有很多：資料庫、Redis、Zookeeper。這裡主要介紹的是通過 Redis 來實現一個分散式鎖，至少要保證三個特性：安全性、死鎖、容錯。

安全性：所謂一個蘿蔔一個坑，第一點要做的是上鎖，在任意時刻要保證僅有一個客戶端持有該鎖。

死鎖：造成死鎖可能是由於某種原因，本該釋放的鎖沒有被釋放，因此在上鎖的時候可以同步的設定過期時間，如果由於客戶端自己的原因沒有被釋放，也要保證鎖能夠自動釋放。

容錯：容錯是在多節點的模式下需要考慮的，只要能保證 N/2+1 節點可用，客戶端就可以成功獲取、釋放鎖。

Redis 單例項分散式鎖實現

在 Redis 的單節點例項下實現一個簡單的分散式鎖，這裡會藉助一些簡單的 Lua 指令碼來實現原子性，不瞭解可以參考之前的文章

上鎖

上鎖的第一步就是先通過 setnx 命令佔坑，為了防止死鎖，通常在佔坑之後還會設定一個過期時間 expire，如下所示：

setnx key value
expire key seconds
複製程式碼

以上命令不是一個原子性操作，所謂原子性操作是指命令在執行過程中並不會被其它的執行緒或者請求打斷，以上如果 setnx 執行成功之後，出現網路閃斷 expire 命令便不會得到執行，會導致死鎖出現。

也許你會想到使用事物來解決，但是事物有個特點，要麼成功要麼失敗，都是一口氣執行完成的，在我們上面的例子中，expire 是需要先根據 setnx 的結果來判斷是否需要進行設定，顯然事物在這裡是行不通的，社群也有很多庫來解決這個問題，現在 Redis 官方 2.8 版本之後支援 set 命令傳入 setnx、expire 擴充套件引數，這樣就可以一條命令一口氣執行，避免了上面的問題，如下所示：

• value：建議設定為一個隨機值，在釋放鎖的時候會進一步講解
• EX seconds：設定的過期時間
• PX milliseconds：也是設定過期時間，單位不一樣
• NX|XX：NX 同 setnx 效果是一樣的

set key value [EX seconds] [PX milliseconds] [NX|XX]
複製程式碼

釋放鎖

釋放鎖的過程就是將原本佔有的坑給刪除掉，但是也並不能僅僅使用 del key 刪除掉就萬事大吉了，這樣很容易刪除掉別人的鎖，為什麼呢？舉一個例子客戶端 A 獲取到一把 key = name1 的鎖（2 秒中），緊接著處理自己的業務邏輯，但是在業務邏輯處理這塊阻塞了耗時超過了鎖的時間，鎖是會自動被釋放的，這期間該資源又被客戶端 B 獲取了 key = name1 的鎖，那麼客戶端 A 在自己的業務處理結束之後直接使用 del key 命令刪除會把客戶端 B 的鎖給釋放掉了，所以釋放鎖的時候要做到僅釋放自己佔有的鎖。

加鎖的過程中建議把 value 設定為一個隨機值，主要是為了更安全的釋放鎖，在 del key 之前先判斷這個 key 存在且 value 等於自己指定的值才執行刪除操作。判斷和刪除不是一個原子性的操作，此處仍需藉助 Lua 指令碼實現。

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end
複製程式碼

Redis 單例項分散式鎖 Node.js 實踐

使用 Node.js 的 Redis 客戶端為 ioredis，npm install ioredis -S 先安裝該包。

初始化自定義 RedisLock

class RedisLock {
    /**
     * 初始化 RedisLock
     * @param {*} client 
     * @param {*} options 
     */
    constructor (client,options={}) {
        if (!client) {
            throw new Error('client 不存在');
        }

        if (client.status !== 'connecting') {
            throw new Error('client 未正常連結');
        }

        this.lockLeaseTime = options.lockLeaseTime || 2; // 預設鎖過期時間 2 秒
        this.lockTimeout = options.lockTimeout || 5; // 預設鎖超時時間 5 秒
        this.expiryMode = options.expiryMode || 'EX';
        this.setMode = options.setMode || 'NX';
        this.client = client;
    }
}
複製程式碼

上鎖

通過 set 命令傳入 setnx、expire 擴充套件引數開始上鎖佔坑，上鎖成功返回，上鎖失敗進行重試，在 lockTimeout 指定時間內仍未獲取到鎖，則獲取鎖失敗。

class RedisLock {
    
    /**
     * 上鎖
     * @param {*} key 
     * @param {*} val 
     * @param {*} expire 
     */
    async lock(key,val,expire) {
        const start = Date.now();
        const self = this;

        return (async function intranetLock() {
            try {
                const result = await self.client.set(key,self.expiryMode,expire || self.lockLeaseTime,self.setMode);
        
                // 上鎖成功
                if (result === 'OK') {
                    console.log(`${key} ${val} 上鎖成功`);
                    return true;
                }

                // 鎖超時
                if (Math.floor((Date.now() - start) / 1000) > self.lockTimeout) {
                    console.log(`${key} ${val} 上鎖重試超時結束`);
                    return false;
                }

                // 迴圈等待重試
                console.log(`${key} ${val} 等待重試`);
                await sleep(3000);
                console.log(`${key} ${val} 開始重試`);

                return intranetLock();
            } catch(err) {
                throw new Error(err);
            }
        })();
    }
}
複製程式碼

釋放鎖

釋放鎖通過 redis.eval(script) 執行我們定義的 redis lua 指令碼。

class RedisLock {
    /**
     * 釋放鎖
     * @param {*} key 
     * @param {*} val 
     */
    async unLock(key,val) {
        const self = this;
        const script = "if redis.call('get',KEYS[1]) == ARGV[1] then" +
        "   return redis.call('del',KEYS[1]) " +
        "else" +
        "   return 0 " +
        "end";

        try {
            const result = await self.client.eval(script,1,key,val);

            if (result === 1) {
                return true;
            }
            
            return false;
        } catch(err) {
            throw new Error(err);
        }
    }
}
複製程式碼

測試

這裡使用了 uuid 來生成唯一 ID，這個隨機數 id 只要保證唯一不管用哪種方式都可。

const Redis = require("ioredis");
const redis = new Redis(6379,"127.0.0.1");
const uuidv1 = require('uuid/v1');
const redisLock = new RedisLock(redis);

function sleep(time) {
    return new Promise((resolve) => {
        setTimeout(function() {
            resolve();
        },time || 1000);
    });
}

async function test(key) {
    try {
        const id = uuidv1();
        await redisLock.lock(key,id,20);
        await sleep(3000);
        
        const unLock = await redisLock.unLock(key,id);
        console.log('unLock: ',unLock);
    } catch (err) {
        console.log('上鎖失敗',err);
    }  
}

test('name1');
test('name1');
複製程式碼

同時呼叫了兩次 test 方法進行上鎖，只有第一個是成功的，第二個 name1 26e02970-0532-11ea-b978-2160dffafa30 上鎖的時候發現 key = name1 已被佔坑，開始重試，由於以上測試中設定了 3 秒鐘之後自動釋放鎖，name1 26e02970-0532-11ea-b978-2160dffafa30 在經過兩次重試之後上鎖成功。

name1 26e00260-0532-11ea-b978-2160dffafa30 上鎖成功
name1 26e02970-0532-11ea-b978-2160dffafa30 等待重試
name1 26e02970-0532-11ea-b978-2160dffafa30 開始重試
name1 26e02970-0532-11ea-b978-2160dffafa30 等待重試
unLock:  name1 26e00260-0532-11ea-b978-2160dffafa30 true
name1 26e02970-0532-11ea-b978-2160dffafa30 開始重試
name1 26e02970-0532-11ea-b978-2160dffafa30 上鎖成功
unLock:  name1 26e02970-0532-11ea-b978-2160dffafa30 true
複製程式碼

原始碼地址

https://github.com/Q-Angelo/project-training/tree/master/redis/lock/redislock.js
複製程式碼

Redlock 演演算法

以上是使用 Node.js 對 Redis 分散式鎖的一個簡單實現，在單例項中是可用的，當我們對 Redis 節點做一個擴充套件，在 Sentinel、Redis Cluster 下會怎麼樣呢？

以下是一個 Redis Sentinel 的故障自動轉移示例圖，假設我們客戶端 A 在主節點 192.168.6.128 獲取到鎖之後，主節點還未來得及同步資訊到從節點就掛掉了，這時候 Sentinel 會選舉另外一個從節點做為主節點，那麼客戶端 B 此時也來申請相同的鎖，就會出現同樣一把鎖被多個客戶端持有，對資料的最終一致性有很高的要求還是不行的。

Redlock 介紹

鑑於這些問題，Redis 官網 redis.io/topics/dist… 提供了一個使用 Redis 實現分散式鎖的規範演演算法 Redlock，中文翻譯版參考 redis.cn/topics/dist…

Redlock 在上述檔案也有描述，這裡簡單做個總結：Redlock 在 Redis 單例項或多例項中提供了強有力的保障，本身具備容錯能力，它會從 N 個例項使用相同的 key、隨機值嘗試 set key value [EX seconds] [PX milliseconds] [NX|XX] 命令去獲取鎖，在有效時間內至少 N/2+1 個 Redis 例項取到鎖，此時就認為取鎖成功，否則取鎖失敗，失敗情況下客戶端應該在所有的 Redis 例項上進行解鎖。

Node.js 中應用 Redlock

github.com/mike-marcac… 是 Node.js 版的 Redlock 實現，使用起來也很簡單，開始之前先安裝 ioredis、redlock 包。

npm i ioredis -S
npm i redlock -S
複製程式碼

編碼

const Redis = require("ioredis");
const client1 = new Redis(6379,"127.0.0.1");
const Redlock = require('redlock');
const redlock = new Redlock([client1],{
    retryDelay: 200,// time in ms
    retryCount: 5,});

// 多個 Redis 例項
// const redlock = new Redlock(
//     [new Redis(6379,"127.0.0.1"),new Redis(6379,"127.0.0.2"),"127.0.0.3")],
// )

async function test(key,ttl,client) {
    try {
        const lock = await redlock.lock(key,ttl);

        console.log(client,lock.value);
        // do something ...

        // return lock.unlock();
    } catch(err) {
        console.error(client,err);
    }
}

test('name1',10000,'client1');
test('name1','client2');
複製程式碼

測試

對同一個 key name1 兩次上鎖，由於 client1 先取到了鎖，client2 無法獲取鎖，重試 5 次之後報錯 LockError: Exceeded 5 attempts to lock the resource "name1".