代碼對應github地址

為什么需要分布式鎖

鎖的作用就是多個執行緒或者行程對同一份資源進行修改時，保證資源是被正確地修改，例如多個執行緒同時對一個數字加一，由于讀取、修改、賦值，不一定是一個原子操作，需要鎖來保證這一個程序的原子性

分布式鎖就是保證多個應用或行程對同一份資源進行操作時，結果的正確性

分布式鎖常見解決方案

分布式鎖需要具備的幾個條件:

• 互斥（必須）：同一時刻，分布式部署的應用中，同一個方法/資源只能被一臺機器上的一個執行緒占用
• 鎖失效保護（必須）：出現客戶端斷電等例外情況，鎖仍然能被其它客戶端獲取，防止死鎖
• 可重入（可選）：同一個執行緒在沒有釋放鎖之前，如果想再次操作，可以直接獲得鎖
• 阻塞/非阻塞（可選）：若沒有獲取到鎖，回傳獲取失敗
• 高可用、高性能（可選）：獲取釋放鎖最好是原子操作，獲取釋放鎖的性能要好

資料庫實作分布式鎖

CREATE TABLE `testlock` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主鍵',
  `method_name` varchar(100) NOT NULL DEFAULT '' COMMENT '鎖定的方法名',
  `value` varchar(1024) NOT NULL DEFAULT '鎖資訊',
  `expire` timestamp(6) NULL DEFAULT NULL COMMENT '超時機制',
  `request_info` varchar(255) DEFAULT NULL COMMENT '持有資訊',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uidx_method_name` (`method_name`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='鎖定中的方法';

加鎖

insert into testlock(method_name, value, expire, request_info) values ('m1', '1', timeout, machine+thread);

釋放鎖

delete from myLock where method_name ='m1';

鎖失效

應用起一個執行緒，輪詢洗掉表中超時的資料

鎖重入

若果要實作鎖重入，則將value初始化為0，用悲觀鎖或者樂觀鎖鎖住，然后獲取鎖時，判斷value是否為0/當前持有資訊是否為當前執行緒，是的話則獲取鎖成功，value = value+1，釋放鎖時同樣value = value - 1，當value = 0 時，置空request_info；

BEGIN;
select * from testlock where method='';
if(value==0||request_info==current_thread){
    //success
    update testlock set value = value+1,request_info=current_thread;
}
COMMIT;

爭奪鎖
代碼層面實作重試

單點問題
配置主從資料庫

方案
優點：跟redis相比，好像優點不大
缺點：性能問題

Redis實作分布式鎖

加鎖

set key value PX milliseconds NX

key、value：鍵值對

PX milliseconds：設定鍵的過期時間為 milliseconds 毫秒

NX：只在鍵不存在時，才對鍵進行設定操作，SET key value NX 效果等同于 SETNX key value

PX、expireTime 引數則是用于解決沒有解鎖導致的死鎖問題，因為如果沒有過期時間，萬一程式員寫的代碼有 bug 導致沒有解鎖操作，則就出現了死鎖，因此該引數起到了一個“兜底”的作用，

NX 引數用于保證在多個執行緒并發 set 下，只會有1個執行緒成功，起到了鎖的“唯一”性，

解鎖

1. 查詢當前“鎖”是否還是我們持有，因為存在過期時間，所以可能等你想解鎖的時候，“鎖”已經到期，然后被其他執行緒獲取了，所以我們在解鎖前需要先判斷自己是否還持有“鎖”
2. 如果“鎖”還是我們持有，則執行解鎖操作，也就是洗掉該鍵值對，并回傳成功；否則，直接回傳失敗，

由于當前 Redis 還沒有原子命令直接支持這兩步操作，所以當前通常是使用 Lua 腳本來執行解鎖操作，Redis 會保證腳本里的內容執行是一個原子操作

腳本代碼如下，邏輯比較簡單：

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

兩個引數的意義如下：

KEYS[1]：我們要解鎖的 key

ARGV[1]：我們加鎖時的 value，用于判斷當“鎖”是否還是我們持有，如果被其他執行緒持有了，value 就會發生變化

代碼實作

首先模擬應用場景

@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class RedisLockTest {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    int count = 0;
    String methodName = "addOne";

    @Test
    public void test() throws InterruptedException {
        RedisLock redisLock = new RedisLock(redisTemplate);
        ThreadPoolExecutor threadPoolExecutor =
                new ThreadPoolExecutor(100, 200, 20, TimeUnit.SECONDS, new LinkedBlockingDeque<>());
        for (int i = 0; i < 500; i++) {
            CompletableFuture.runAsync(() -> addOne(), threadPoolExecutor);
            CompletableFuture.runAsync(() -> addOne(), threadPoolExecutor);
            CompletableFuture.runAsync(() -> addOne(), threadPoolExecutor);
        }
        // 等待50s，等剩余任務跑完
        Thread.sleep(50000);
        System.out.println(count);
    }

    public void addOne() {
        count++;
    }
}

output：1982
以上這段代碼理論上count最后應該為1500，但是運行多次都少于1500，這就是我們常見的應用場景，資料不一致的問題

接下來，我們撰寫分布式鎖

public class RedisLock {
    /* redis操作類 */
    private RedisTemplate<String, String> redisTemplate;
    // 獲取分布式鎖
    public boolean lock(String methodName, String lockId, long timeout) {
        if (timeout <= 1) throw new RuntimeException("過期時間應大于1毫秒");
        while (true) {
            // 沒有獲取到鎖就自旋，應該還有掛起通知的方式，暫且不研究
            boolean lock = tryLock(methodName, lockId, timeout);
            if (lock) {
                break;
            }
            try {
                // 執行緒暫停50毫秒，避免請求redis太頻繁
                Thread.sleep(50);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return true;
}
    // 實際獲取鎖的程序
    public boolean tryLock(String methodName, String lockId, long timeout) {
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(methodName, lockId, timeout, TimeUnit.MILLISECONDS);
        if (flag == null)
            return false;
        return flag;
    }
    // 解除鎖的腳本
    public static final String UP_LOCK_SCRIPT = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
            "then\n" +
            "    return redis.call(\"del\",KEYS[1])\n" +
            "else\n" +
            "    return 0\n" +
            "end\n";
    
    // 解除分布式鎖
    public boolean unlock(String methodName, String lockId) {
        DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(UN_LOCK_SCRIPT, Boolean.class);
        Boolean result = redisTemplate.execute(redisScript, Collections.singletonList(methodName), lockId);
        if (result == null) {
            return false;
        }
        return result;
}
}

我們修改一下addOne方法

public void addOne(RedisLock redisLock) {
    long id = Thread.currentThread().getId();
    redisLock.lock(methodName, String.valueOf(id), 2000);
    count++;
    redisLock.unlock(methodName, String.valueOf(id));
}

在使用redis分布式鎖的情況下，最終結果總是等于1500，注意，主執行緒一定得等其它執行緒跑完才停止，時間設定的長一點

過期未處理完怎么解決

為了防止死鎖，我們會給分布式鎖加一個過期時間，但是萬一這個時間到了，我們業務邏輯還沒處理完，怎么辦？

首先，我們在設定過期時間時要結合業務場景去考慮，盡量設定一個比較合理的值，就是理論上正常處理的話，在這個過期時間內是一定能處理完畢的，

之后，我們再來考慮對這個問題進行兜底設計，

關于這個問題，目前常見的解決方法有兩種：

1. 守護執行緒“續命”：額外起一個執行緒，定期檢查執行緒是否還持有鎖，如果有則延長過期時間，Redisson
   里面就實作了這個方案，使用“看門狗”定期檢查（每1/3的鎖時間檢查1次），如果執行緒還持有鎖，則重繪過期時間，

2. 超時回滾：當我們解鎖時發現鎖已經被其他執行緒獲取了，說明此時我們執行的操作已經是“不安全”的了，此時需要進行回滾，并回傳失敗

同時，需要進行告警，人為介入驗證資料的正確性，然后找出超時原因，是否需要對超時時間進行優化等等

我們來實作一個自動續期，每 1/2個過期時間檢查一次，重新將過期時間設定為一個原過期時間單位的看門狗

我們對redislock增加三個屬性，用延時佇列來操作延時任務，更多的延時任務方法請看這片文章https://blog.csdn.net/echizao1839/article/details/105533214

/* 最大延遲次數 需要配置可自己修改代碼 */
public final static long MAX_TIMES = 5;
/* 看門狗守護執行緒 */
private Thread daemonTread;
/* 延時佇列 */
DelayQueue<DelayTask> queue = new DelayQueue<>();

public RedisLock(RedisTemplate<String, String> redisTemplate) {
    this.redisTemplate = redisTemplate;
    daemonTread = new Thread(() -> watchDog(queue), "delay-watchdog");
    daemonTread.setDaemon(true);
    daemonTread.start();
}

撰寫一個延時任務類

@Getter
public static class DelayTask implements Delayed {
    // 方法名稱
    private String methodName;
    // 執行緒名稱
    private String lockId;
    // 單位為毫秒
    private long expire;
    // 單位為毫秒
    private long exeTime;
    // 當前重試次數
    private long times;

    public DelayTask(String methodName, String lockId, long expire, long times) {
        this(methodName, lockId, expire);
        this.times = times;
    }

    public DelayTask(String methodName, String lockId, long expire) {
        this.methodName = methodName;
        this.lockId = lockId;
        this.expire = expire;
        this.exeTime = System.currentTimeMillis() + expire;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return exeTime - System.currentTimeMillis();
    }

    @Override
    public int compareTo(Delayed o) {
        DelayTask t = (DelayTask) o;
        if (this.exeTime - t.exeTime <= 0) {
            return -1;
        } else {
            return 1;
        }
    }
}

要想延長redis key的時間，撰寫操作redis的腳本

// 延長鎖時間lua腳本
public static final String EXTEND_EXPIRE_TIME_SCRIPT = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
        "then\n" +
        "    return redis.call(\"pexpire\",KEYS[1],ARGV[2])\n" +
        "else\n" +
        "    return 0\n" +
        "end\n";

// 延長鎖的時間
public boolean extendExpire(String methodName, String lockId, long expire) {
    DefaultRedisScript<Boolean> redisScript = new DefaultRedisScript<>(EXTEND_EXPIRE_TIME_SCRIPT, Boolean.class);
    Boolean result = this.redisTemplate.execute(redisScript, Collections.singletonList(methodName), lockId, String.valueOf(expire));
    if (result == null) {
        return false;
    }
    return result;
}

實作看門狗的邏輯

public void watchDog(DelayQueue<DelayTask> queue) {
    while (true) {
        try {
            DelayTask delayTask = queue.take();
            String methodName = delayTask.getMethodName();
            String threadId = delayTask.getLockId();
            long times = delayTask.getTimes();
            long expire = delayTask.getExpire();
            if (times < MAX_TIMES) {
                boolean success = extendExpire(methodName, threadId, expire);
                if (success) {
                    // 延時成功，重新加入延時佇列
                    queue.add(new DelayTask(methodName, threadId, (expire + 1) / 2, times + 1));
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

守護執行緒續命的方案有什么漏洞

Redisson 使用看門狗（守護執行緒）“續命”的方案在大多數場景下是挺不錯的，也被廣泛應用于生產環境，但是在極端情況下還是會存在問題

問題例子如下：

1. 執行緒1首先獲取鎖成功，將鍵值對寫入 redis 的 master 節點
2. 在 redis 將該鍵值對同步到 slave 節點之前，master 發生了故障
3. redis 觸發故障轉移，其中一個 slave 升級為新的 master
4. 此時新的 master 并不包含執行緒1寫入的鍵值對，因此執行緒2嘗試獲取鎖也可以成功拿到鎖
5. 此時相當于有兩個執行緒獲取到了鎖，可能會導致各種預期之外的情況發生，例如最常見的臟資料

解決方法：上述問題的根本原因主要是由于 redis 異步復制帶來的資料不一致問題導致的，因此解決的方向就是保證資料的一致

當前比較主流的解法和思路有兩種：
1）Redis 作者提出的 RedLock；2）Zookeeper 實作的分布式鎖

RedLock的優缺點

首先，該方案也是基于文章開頭的那個方案（set加鎖、lua腳本解鎖）進行改良的，所以 antirez 只描述了差異的地方，大致方案如下，

假設我們有 N 個 Redis 主節點，例如 N = 5，這些節點是完全獨立的，我們不使用復制或任何其他隱式協調系統，為了取到鎖，客戶端應該執行以下操作:

1. 獲取當前時間，以毫秒為單位
2. 依次嘗試從5個實體，使用相同的 key 和隨機值（例如UUID）獲取鎖，當向Redis 請求獲取鎖時，客戶端應該設定一個超時時間，這個超時時間應該小于鎖的失效時間，例如你的鎖自動失效時間為10秒，則超時時間應該在 5-50 毫秒之間，這樣可以防止客戶端在試圖與一個宕機的 Redis 節點對話時長時間處于阻塞狀態，如果一個實體不可用，客戶端應該盡快嘗試去另外一個Redis實體請求獲取鎖
3. 客戶端通過當前時間減去步驟1記錄的時間來計算獲取鎖使用的時間，當且僅當從大多數（N/2+1，這里是3個節點）的Redis節點都取到鎖，并且獲取鎖使用的時間小于鎖失效時間時，鎖才算獲取成功
4. 如果取到了鎖，其有效時間等于有效時間減去獲取鎖所使用的時間（步驟3計算的結果）
5. 如果由于某些原因未能獲得鎖（無法在至少N/2+1個Redis實體獲取鎖、或獲取鎖的時間超過了有效時間），客戶端應該在所有的Redis實體上進行解鎖（即便某些Redis實體根本就沒有加鎖成功，防止某些節點獲取到鎖但是客戶端沒有得到回應而導致接下來的一段時間不能被重新獲取鎖）

可以看出，該方案為了解決資料不一致的問題，直接舍棄了異步復制，只使用 master 節點，同時由于舍棄了 slave，為了保證可用性，引入了 N 個節點，官方建議是 5，

該方案看著挺美好的，但是實際上我所了解到的在實際生產上應用的不多，主要有兩個原因：
1）該方案的成本似乎有點高，需要使用5個實體；
2）該方案一樣存在問題，

該方案主要存以下問題：

1）嚴重依賴系統時鐘，如果執行緒1從3個實體獲取到了鎖，但是這3個實體中的某個實體的系統時間走的稍微快一點，則它持有的鎖會提前過期被釋放，當他釋放后，此時又有3個實體是空閑的，則執行緒2也可以獲取到鎖，則可能出現兩個執行緒同時持有鎖了，

2）如果執行緒1從3個實體獲取到了鎖，但是萬一其中有1臺重啟了，則此時又有3個實體是空閑的，則執行緒2也可以獲取到鎖，此時又出現兩個執行緒同時持有鎖了

針對以上問題其實后續也有人給出一些相應的解法，但是整體上來看還是不夠完美，所以目前實際應用得不是那么多

Zookeeper實作分布式鎖

Zookeeper 的分布式鎖實作方案如下：

1. 創建一個鎖目錄 /locks，該節點為持久節點
2. 想要獲取鎖的執行緒都在鎖目錄下創建一個臨時順序節點
3. 獲取鎖目錄下所有子節點，對子節點按節點自增序號從小到大排序
4. 判斷本節點是不是第一個子節點，如果是，則成功獲取鎖，開始執行業務邏輯操作；如果不是，則監聽自己的上一個節點的洗掉事件
5. 持有鎖的執行緒釋放鎖，只需洗掉當前節點即可
6. 當自己監聽的節點被洗掉時，監聽事件觸發，則回到第3步重新進行判斷，直到獲取到鎖

由于 Zookeeper 保證了資料的強一致性，因此不會存在之前 Redis 方案中的問題，整體上來看還是比較不錯的

Zookeeper 方案的主要問題在于性能不如 Redis 那么好，當申請鎖和釋放鎖的頻率較高時，會對集群造成壓力，此時集群的穩定性可用性能可能又會遭受挑戰

接下來我們來撰寫zookeeper分布式鎖
首先，自己用docker搭建zookeeper環境，然后了解zookeeper的基本結構和指令
實體化一個zookeeper客戶端，我們采用單例模式來實體化

private static volatile ZooKeeper zkCli = null;
/* zookeeper的地址和埠 */
private static final String connectString = "127.0.0.1:2181";
/* 會話超時時間 */
private static final int sessionTimeout = 30000;

/**
 * 初始化
 * @return
 */
public static ZooKeeper getClient() {
    if (zkCli == null) {
        synchronized (ZooKeeper.class) {
            if (zkCli == null) {
                CountDownLatch countDownLatch = new CountDownLatch(1);
                try {
                    zkCli = new ZooKeeper(connectString, sessionTimeout, event -> {
                        if (zkCli.getState().isConnected()) {
                            // 監聽連接zookeeper狀態
                            countDownLatch.countDown();
                        }
                    });
                    // 因為連接需要時間，代碼不能直接往下走
                    countDownLatch.await();
                } catch (Exception e) {
                    System.out.println("create zookeeper client fail！！！！！！！！！！！！！！！！！！！");
                } finally {
                    countDownLatch.countDown();
                }
            }
        }
    }
    return zkCli;
}

實作zookeeper加鎖邏輯

@SneakyThrows
public static String lock(String methodName, String lockId) {

    String path = getPath(methodName);
    // 注冊一個節點
    String zNode = ZookeeperUtil.getClient().create(path + "/node", lockId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    while (true) {
        // 獲取所有節點
        List<String> children = ZookeeperUtil.getClient().getChildren(path, true);
        int index = children.indexOf(zNode.substring(zNode.lastIndexOf("/") + 1));
        // 判斷是否第一個
        if (index != 0) {
            //否，注冊監聽到前一個節點，然后掛起
            CountDownLatch countDownLatch = new CountDownLatch(1);
            Stat exists = ZookeeperUtil.getClient().exists(path + "/" + children.get(index - 1), event -> {
                countDownLatch.countDown();
            });
            if (exists != null) {
                countDownLatch.await();
            }
        } else {
            return zNode;
        }
    }
}

解鎖的邏輯

@SneakyThrows
public static void unLock(String zNode) {
    // 因為洗掉的介面有個版本號，所以先獲取節點的資訊，再洗掉
    Stat stat = ZookeeperUtil.getClient().exists(zNode, true);
    ZookeeperUtil.getClient().delete(zNode, stat.getVersion());
}

加入快取來減少一步zookeeper查詢

/* 將方法名和zookeeper節點路徑快取起來，提高性能，可能存在的問題就是快取下來了，有其他人偷偷刪了，典型快取不一致問題，可以根據實際情況思考 */
private static final ConcurrentHashMap<String, String> concurrentHashMap = new ConcurrentHashMap<>();

@SneakyThrows
public static String getPath(String methodName) {

    String path = concurrentHashMap.get(methodName);
    if (path == null) {
        // 通過方法名，判斷zookeeper目錄節點是否存在，不存在則創建一個持久性節點
        Stat stat = ZookeeperUtil.getClient().exists("/locks/" + methodName, true);
        if (stat == null) {
            ZookeeperUtil.getClient().create("/locks/" + methodName, "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        concurrentHashMap.put(methodName, "/locks/" + methodName);
    }
    return concurrentHashMap.get(methodName);
}

測驗代碼，具體環境和結果可以參照前面redis的，基本一樣

public void addOneWithZookeeperLock() {
    long id = Thread.currentThread().getId();
    String lock = ZookeeperLock.lock(methodName, String.valueOf(id));
    count++;
    System.out.println(count);
    ZookeeperLock.unLock(lock);
}

分布式鎖的選擇

上面我們對三種分布式鎖的實作方案進行了闡述
在日常作業中如何選擇合適的方案來使用

• 如果我們的作業環境中已經引入了redis和zookeeper中間件，那么可以優先考慮這兩種再根據實際的業務情況，比如業務的性能和穩定性需求來綜合考慮
• 如果沒有使用上面這兩種中間件，并且性能要求不高，那么可以直接用資料庫來解決
• 如果性能不夠，那么可以用redis來解決
• 如果對穩定性和安全性要求極高，可以考慮用zookeeper

參考文章