分佈式鎖的實現

時間 2019-11-08

原文原文鏈接

本文摘自網上，僅做學習使用！javascript

用一web應用集羣，負載均衡部署實現：java

在上圖能夠看到，變量A在JVM一、JVM二、JVM3三個JVM內存中（這個變量A主要體現是在一個類中的一個成員變量，是一個有狀態的對象），若是咱們不加任何控制的話，變量A同進都會在JVM分配一塊內存，三個請求發過來同時對這個變量進行操做，顯然結果不是咱們想要的。web

若是咱們業務中存在這樣的場景的話，就須要找到一種方法來解決。redis

爲了保證一個方法或屬性在高併發的狀況下同一時間只能被同一個線程執行，在傳統單機部署的狀況下，可使用Java併發處理相關的API（如ReentrantLock或Synchronized）進行互斥控制。可是，隨之業務發展的須要，原單機部署的系統演化成分佈式集羣系統後，因爲分佈式系統多線程、多進程而且分佈在不一樣的機器上，這將原來的單機部署狀況下的併發控制鎖策略失效，單純的Java API並不能提供分佈式鎖的能力。
爲了解決這個問題，就須要一種跨JVM的互斥機制來控制共享資源的訪問，這就是分佈式鎖要解決的問題！數據庫

分佈式鎖應該具有哪些條件

在分佈式系統環境下，一個方法在同一時間只能被一個機器的一個線程執行；
高可用、高性能的獲取鎖與釋放鎖；
具有可重入特性；
具有鎖失效機制、防止死鎖；
具有非阻塞鎖特性，即沒有獲取到鎖直接返回獲取鎖失敗；

分佈式鎖的實現方式

目前幾乎全部大型網站及應用都是分佈式部署，分佈式場景中的數據一致性問題一直是一個比較重要的話題，分佈式的CAP理論告訴咱們任何一個分佈式系統都沒法同時知足一致性（Consistency）、可用性（Availability）和分區容錯性（Partition tolerance）,最多隻能同時知足兩項。
通常狀況下，都須要犧牲強一致性來換取系統的高可用性，系統每每只須要保證最終一致性，只要這個最終時間是在用戶能夠接受的範圍內便可。
在不少時候，爲了保證數據的最終一致性，須要不少的技術方案來支持，好比分佈式事務、分佈式鎖等。有的時候，咱們須要保證一信方法在同一時間內只能被同一個線程執行。
而分佈式鎖的具體實現方案有以下三種：緩存

基於數據庫實現；
基於緩存（Redis等）實現；
基於Zookeeper實現；服務器

以上儘管有三種方案，可是咱們須要根據不一樣的業務進行選型。多線程

數據庫鎖

基於數據庫表

要實現分佈式鎖，最簡單的方式可能就是直接建立一張鎖表，而後經過操做該表中的數據來實現了。併發

當咱們要鎖住某個方法或資源時，咱們就在該表中增長一條記錄，想要釋放鎖的時候就刪除這條記錄。負載均衡

CREATE TABLE `methodLock` ( `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主鍵', `method_name` varchar(64) NOT NULL DEFAULT '' COMMENT '鎖定的方法名', `desc` varchar(1024) NOT NULL DEFAULT '備註信息', `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '保存數據時間，自動生成', PRIMARY KEY (`id`), UNIQUE KEY `uidx_method_name` (`method_name `) USING BTREE ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='鎖定中的方法';

當咱們想要鎖住某個方法時，執行如下SQL：

insert into methodLock(method_name,desc) values (‘method_name’,‘desc’)

由於咱們對method_name作了惟一性約束，這裏若是有多個請求同時提交到數據庫的話，數據庫會保證只有一個操做能夠成功，那麼咱們就能夠認爲

操做成功的那個線程得到了該方法的鎖，能夠執行方法體內容。

當方法執行完畢以後，想要釋放鎖的話，須要執行如下Sql:

delete from methodLock where method_name ='method_name'

上面這種簡單的實現有如下幾個問題：

一、這把鎖強依賴數據庫的可用性，數據庫是一個單點，一旦數據庫掛掉，會致使業務系統不可用。

二、這把鎖沒有失效時間，一旦解鎖操做失敗，就會致使鎖記錄一直在數據庫中，其餘線程沒法再得到到鎖。

三、這把鎖只能是非阻塞的，由於數據的insert操做，一旦插入失敗就會直接報錯。沒有得到鎖的線程並不會進入排隊隊列，要想再次得到鎖就要再次觸發得到鎖操做。

四、這把鎖是非重入的，同一個線程在沒有釋放鎖以前沒法再次得到該鎖。由於數據中數據已經存在了。

固然，咱們也能夠有其餘方式解決上面的問題。

數據庫是單點？搞兩個數據庫，數據以前雙向同步。一旦掛掉快速切換到備庫上。
沒有失效時間？只要作一個定時任務，每隔必定時間把數據庫中的超時數據清理一遍。
非阻塞的？搞一個while循環，直到insert成功再返回成功。
非重入的？在數據庫表中加個字段，記錄當前得到鎖的機器的主機信息和線程信息，那麼下次再獲取鎖的時候先查詢數據庫，若是當前機器的主機信息和線程信息在數據庫能夠查到的話，直接把鎖分配給他就能夠了。

基於數據庫的排它鎖

除了能夠經過增刪操做數據表中的記錄之外，其實還能夠藉助數據庫中自帶的鎖來實現分佈式的鎖。

咱們還用剛剛建立的那張數據庫表。能夠經過數據庫的排他鎖來實現分佈式鎖。

在查詢語句後面增長for update，數據庫會在查詢過程當中給數據庫表增長排他鎖。當某條記錄被加上排他鎖以後，其餘線程沒法再在該行記錄上增長排他鎖。

咱們能夠認爲得到排它鎖的線程便可得到分佈式鎖，當獲取到鎖以後，能夠執行方法的業務邏輯，執行完方法以後，再經過如下方法解鎖：

public void unlock(){ connection.commit(); }

經過connection.commit()操做來釋放鎖。

這種方法能夠有效的解決上面提到的沒法釋放鎖和阻塞鎖的問題。

阻塞鎖？ for update語句會在執行成功後當即返回，在執行失敗時一直處於阻塞狀態，直到成功。
鎖定以後服務宕機，沒法釋放？使用這種方式，服務宕機以後數據庫會本身把鎖釋放掉。

可是仍是沒法直接解決數據庫單點和可重入問題。

總結：

總結一下使用數據庫來實現分佈式鎖的方式，這兩種方式都是依賴數據庫的一張表，一種是經過表中的記錄的存在狀況肯定當前是否有鎖存在，另一種是經過數據庫的排他鎖來實現分佈式鎖。

**數據庫實現分佈式鎖的優勢: **直接藉助數據庫，容易理解。

**數據庫實現分佈式鎖的缺點: **會有各類各樣的問題，在解決問題的過程當中會使整個方案變得愈來愈複雜。

操做數據庫須要必定的開銷，性能問題須要考慮。

樂觀鎖

樂觀鎖假設認爲數據通常狀況下不會形成衝突，只有在進行數據的提交更新時，纔會檢測數據的衝突狀況，若是發現衝突了，則返回錯誤信息

實現方式：

時間戳（timestamp）記錄機制實現：給數據庫表增長一個時間戳字段類型的字段，當讀取數據時，將timestamp字段的值一同讀出，數據每更新一次，timestamp也同步更新。當對數據作提交更新操做時，檢查當前數據庫中數據的時間戳和本身更新前取到的時間戳進行對比，若相等，則更新，不然認爲是失效數據。

若出現更新衝突，則須要上層邏輯修改，啓動重試機制

一樣也可使用version的方式。

性能對比

（1）悲觀鎖實現方式是獨佔數據，其它線程須要等待，不會出現修改的衝突，可以保證數據的一致性，可是依賴數據庫的實現，且在線程較多時出現等待形成效率下降的問題。通常狀況下，對於數據很敏感且讀取頻率較低的場景，能夠採用悲觀鎖的方式

（2）樂觀鎖能夠多線程同時讀取數據，若出現衝突，也能夠依賴上層邏輯修改，可以保證高併發下的讀取，適用於讀取頻率很高而修改頻率較少的場景

（3）因爲庫存回寫數據屬於敏感數據且讀取頻率適中，因此建議使用悲觀鎖優化

基於redis的分佈式鎖

相比較於基於數據庫實現分佈式鎖的方案來講，基於緩存來實如今性能方面會表現的更好一點。並且不少緩存是能夠集羣部署的，能夠解決單點問題。

首先，爲了確保分佈式鎖可用，咱們至少要確保鎖的實現同時知足如下四個條件：

互斥性。在任意時刻，只有一個客戶端能持有鎖。
不會發生死鎖。即便有一個客戶端在持有鎖的期間崩潰而沒有主動解鎖，也能保證後續其餘客戶端能加鎖。
具備容錯性。只要大部分的Redis節點正常運行，客戶端就能夠加鎖和解鎖。
解鈴還須繫鈴人。加鎖和解鎖必須是同一個客戶端，客戶端本身不能把別人加的鎖給解了。

基於緩存（Redis）的實現方式

使用Redis實現分佈式鎖的理由：

Redis具備很高的性能；
Redis的命令對此支持較好，實現起來很方便；

Redis命令介紹：

SETNX

// 當且僅當key不存在時，set一個key爲val的字符串，返回1；
// 若key存在，則什麼都不作，返回0。
SETNX key val;

expire

// 爲key設置一個超時時間，單位爲second，超過這個時間鎖會自動釋放，避免死鎖。
expire key timeout;

delete

// 刪除key
delete key;

咱們經過Redis實現分佈式鎖時，主要經過上面的這三個命令。

經過Redis實現分佈式的核心思想爲：

獲取鎖的時候，使用setnx加鎖，並使用expire命令爲鎖添加一個超時時間，超過該時間自動釋放鎖，鎖的value值爲一個隨機生成的UUID，經過這個value值，在釋放鎖的時候進行判斷。

獲取鎖的時候還設置一個獲取的超時時間，若超過這個時間則放棄獲取鎖。
3.釋放鎖的時候，經過UUID判斷是否是當前持有的鎖，若時該鎖，則執行delete進行鎖釋放。

具體實現代碼以下：

public class DistributedLock {

    private final JedisPool jedisPool;
    private final static String KEY_PREF = "lock:"; // 鎖的前綴

    public DistributedLock(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
    }

    /**
     * 加鎖
     *
     * @param lockName       String 鎖的名稱(key)
     * @param acquireTimeout long 獲取超時時間
     * @param timeout        long 鎖的超時時間
     * @return 鎖標識
     */
    public String lockWithTimeout(String lockName, long acquireTimeout, long timeout) {
        Jedis conn = null;

        try {
            // 獲取鏈接
            conn = jedisPool.getResource();
            // 隨機生成一個value
            String identifier = UUID.randomUUID().toString();
            // 鎖名,即 key值
            String lockKey = KEY_PREF + lockName;
            // 超時時間, 上鎖後超過此時間則自動釋放鎖
            int lockExpire = (int) (timeout / 1000);

            // 獲取鎖的超時時間,超過這個時間則放棄獲取鎖
            long end = System.currentTimeMillis() + acquireTimeout;
            while (System.currentTimeMillis() < end) {
                if (conn.setnx(lockKey, identifier) == 1) {
                    conn.expire(lockKey, lockExpire);
                    // 返回value值,用於釋放鎖時間確認
                    return identifier;
                }

                // 返回-1表明key沒有設置超時時間,爲key設置一個超時時間
                if (conn.ttl(lockKey) == -1) {
                    conn.expire(lockKey, lockExpire);
                }

                try {
                    Thread.sleep(10);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return null;
    }

    /**
     * 釋放鎖
     *
     * @param lockName   String 鎖key
     * @param identifier String 釋放鎖的標識
     * @return boolean
     */
    public boolean releaseLock(String lockName, String identifier) {
        Jedis conn = null;
        String lockKey = KEY_PREF + lockName;
        boolean retFlag = false;
        try {
            conn = jedisPool.getResource();
            while (true) {
                // 監視lock, 準備開始事務
                conn.watch(lockKey);
                // 經過前面返回的value值判斷是否是該鎖,若時該鎖,則刪除釋放鎖
                if (identifier.equals(conn.get(lockKey))) {
                    Transaction transaction = conn.multi();
                    transaction.del(lockKey);
                    List<Object> results = transaction.exec();
                    if (results == null) continue;

                    retFlag = true;
                }

                conn.unwatch();
                break;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (conn != null) {
                conn.close();
            }
        }
        return retFlag;
    }
}