【180414】分佈式鎖（redis/mysql）

時間 2019-11-16

原文原文鏈接

單臺機器所能承載的量是有限的，用戶的量級上萬，基本上服務都會作分佈式集羣部署。不少時候，會遇到對同一資源的方法。這時候就須要鎖，若是是單機版的，能夠利用java等語言自帶的併發同步處理。若是是多臺機器部署就得要有個中間代理人來作分佈式鎖了。html

經常使用的分佈式鎖的實現有三種方式。java

基於redis實現（利用redis的原子性操做setnx來實現）
基於mysql實現（利用mysql的innodb的行鎖來實現，有兩種方式，悲觀鎖與樂觀鎖）
基於Zookeeper實現（利用zk的臨時順序節點來實現）

目前，我已是用了redis和mysql實現了鎖，而且根據應用場景應用在不一樣的線上環境中。zk實現比較複雜，又無應用場景，有興趣的能夠參考他山之石中的《Zookeeper實現分佈式鎖》。mysql

說說心得和體會。redis

沒有什麼完美的技術、沒有萬能鑰匙、不一樣方式不一樣應用場景 CAP原理：一致性（consistency）、可用性（availability）、分區可容忍性（partition-tolerance）三者取其二。算法

他山之石

Zookeeper實現分佈式鎖：www.jianshu.com/p/5d12a0101…
分佈式鎖的幾種實現方式~：www.hollischuang.com/archives/17…
select for update引起死鎖分析:www.cnblogs.com/micrari/p/8…

基於redis緩存實現分佈式鎖

基於redis的鎖實現比較簡單，因爲redis的執行是單線程執行，自然的具有原子性操做，咱們能夠利用命令setnx和expire來實現，java版代碼參考以下：sql

package com.fenqile.creditcard.appgatewaysale.provider.util;

import com.fenqile.redis.JedisProxy;

import java.util.Date;

/** * User: Rudy Tan * Date: 2017/11/20 * * redis 相關操做 */
public class RedisUtil {

    /** * 獲取分佈式鎖 * * @param key string 緩存key * @param expireTime int 過時時間，單位秒 * @return boolean true-搶到鎖，false-沒有搶到鎖 */
    public static boolean getDistributedLockSetTime(String key, Integer expireTime) {
        try {
            // 移除已經失效的鎖
            String temp = JedisProxy.getMasterInstance().get(key);
            Long currentTime = (new Date()).getTime();
            if (null != temp && Long.valueOf(temp) < currentTime) {
                JedisProxy.getMasterInstance().del(key);
            }

            // 鎖競爭
            Long nextTime = currentTime + Long.valueOf(expireTime) * 1000;
            Long result = JedisProxy.getMasterInstance().setnx(key, String.valueOf(nextTime));
            if (result == 1) {
                JedisProxy.getMasterInstance().expire(key, expireTime);
                return true;
            }
        } catch (Exception ignored) {
        }
        return false;
    }
}
複製代碼

包名和獲取redis操做對象換成本身的就行了。數據庫

基本步驟是緩存

每次進來先檢測一下這個key是否實現。若是失效了移除失效鎖
使用setnx原子命令爭搶鎖。
搶到鎖的設置過時時間。

步驟2爲最核心的東西，爲啥設置步驟3？可能應爲獲取到鎖的線程出現什麼移除請求，而沒法釋放鎖，所以設置一個最長鎖時間，避免死鎖。爲啥設置步驟1？redis可能在設置expire的時候掛掉。設置過時時間不成功，而出現鎖永久生效。併發

線上環境，步驟一、3的問題都出現過。因此要作保底攔截。app

redis集羣部署

一般redis都是以master-slave解決單點問題，多個master-slave組成大集羣，而後經過一致性哈希算法將不一樣的key路由到不一樣master-slave節點上。

redis鎖的優缺點：

優勢：redis自己是內存操做、而且一般是多片部署，所以有這較高的併發控制，能夠抗住大量的請求。缺點：redis自己是緩存，有必定機率出現數據不一致請求。

在線上，以前，利用redis作庫存計數器，獎品發放理論上只發放10個的，最後發放了14個。出現了數據的一致性問題。

所以在這以後，引入了mysql數據庫分佈式鎖。

基於mysql實現的分佈式鎖。

實現初版

在此以前，在網上搜索了大量的文章，基本上都是插入、刪除發的方式或是直接經過"select for update"這種形式獲取鎖、計數器。具體能夠參考他山之石中的《分佈式鎖的幾種實現方式~》關於數據庫鎖章節。

一開始，個人實現方式僞代碼以下：

public boolean getLock(String key){
     select for update if (記錄存在){
           update
     }else {
           insert 
   }
}
複製代碼

這樣實現出現了很嚴重的死鎖問題，具體緣由能夠能夠參考他山之石中的《select for update引起死鎖分析》這個版本中存在以下幾個比較嚴重的問題：

1.一般線上數據是不容許作物理刪除的 2.經過惟一鍵重複報錯，處理錯誤形式是不太合理的。 3.若是appclient在處理中還沒釋放鎖以前就掛掉了，會出現鎖一直存在，出現死鎖。 4.若是以這種方式，實現redis中的計數器（incr decr）,當記錄不存在的時候，會出現大量死鎖的狀況。

所以考慮引入，記錄狀態字段、中央鎖概念。

實現第二版

在第二版中完善了數據庫表設計，參考以下：

-- 鎖表，單庫單表
CREATE TABLE IF NOT EXISTS credit_card_user_tag_db.t_tag_lock (

    -- 記錄index
    Findex INT NOT NULL AUTO_INCREMENT COMMENT '自增索引id',

    -- 鎖信息（key、計數器、過時時間、記錄描述）
    Flock_name VARCHAR(128) DEFAULT '' NOT NULL COMMENT '鎖名key值',
    Fcount INT NOT NULL DEFAULT 0 COMMENT '計數器',
    Fdeadline DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT '鎖過時時間',
    Fdesc VARCHAR(255) DEFAULT '' NOT NULL COMMENT '值/描述',
    
    -- 記錄狀態及相關事件
    Fcreate_time DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT '建立時間',
    Fmodify_time DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT '修改時間',
    Fstatus TINYINT NOT NULL DEFAULT 1 COMMENT '記錄狀態，0：無效，1：有效',

    -- 主鍵（PS：總索引數不能超過5）
    PRIMARY KEY (Findex),
    -- 惟一約束
    UNIQUE KEY uniq_Flock_name(Flock_name),
    -- 普通索引
    KEY idx_Fmodify_time(Fmodify_time)

)ENGINE=INNODB DEFAULT CHARSET=UTF8 COMMENT '信用卡|鎖與計數器表|rudytan|20180412';
複製代碼

在這個版本中，考慮到再條鎖併發插入存在死鎖（間隙鎖爭搶）狀況，引入中央鎖概念。

基本方式是：

根據sql建立好數據庫
建立一條記錄Flock_name="center_lock"的記錄。
在對其餘鎖(如Flock_name="sale_invite_lock")進行操做的時候，先對"center_lock"記錄select for update
"sale_invite_lock"記錄本身的增刪改查。

考慮到不一樣公司引入的數據庫操做包不一樣，所以提供僞代碼，以便於理解僞代碼

// 開啓事務
@Transactional
public boolean getLock(String key){
      // 獲取中央鎖
      select * from tbl where Flock_name="center_lock"    
    
     // 查詢key相關記錄
     select for update if (記錄存在){
           update
     }else {
           insert 
   }
}
複製代碼

/** * 初始化記錄,若是有記錄update,若是沒有記錄insert */
    private LockRecord initLockRecord(String key){
        // 查詢記錄是否存在
        LockRecord lockRecord = lockMapper.queryRecord(key);
        if (null == lockRecord) {
            // 記錄不存在，建立
            lockRecord = new LockRecord();
            lockRecord.setLockName(key);
            lockRecord.setCount(0);
            lockRecord.setDesc("");
            lockRecord.setDeadline(new Date(0));
            lockRecord.setStatus(1);
            lockMapper.insertRecord(lockRecord);
        }
        return lockRecord;
    }

   /** * 獲取鎖，代碼片斷 */
    @Override
    @Transactional
    public GetLockResponse getLock(GetLockRequest request) {
        // 檢測參數
        if(StringUtils.isEmpty(request.lockName)) {
            ResultUtil.throwBusinessException(CreditCardErrorCode.PARAM_INVALID);
        }

        // 兼容參數初始化
        request.expireTime = null==request.expireTime? 31536000: request.expireTime;
        request.desc = Strings.isNullOrEmpty(request.desc)?"":request.desc;
        Long nowTime = new Date().getTime();

        GetLockResponse response = new GetLockResponse();
        response.lock = 0;

        // 獲取中央鎖，初始化記錄
        lockMapper.queryRecordForUpdate("center_lock");
        LockRecord lockRecord = initLockRecord(request.lockName);

        // 未釋放鎖或未過時，獲取失敗
        if (lockRecord.getStatus() == 1
                && lockRecord.getDeadline().getTime() > nowTime){
            return response;
        }

        // 獲取鎖
        Date deadline = new Date(nowTime + request.expireTime*1000);
        int num = lockMapper.updateRecord(request.lockName, deadline, 0, request.desc, 1);
        response.lock = 1;
        return response;
    }
複製代碼

到此，該方案，可以知足個人分佈式鎖的需求。

可是該方案，有一個比較致命的問題，就是全部記錄共享一個鎖，併發並不高。

通過測試，開啓50*100個線程併發修改，5次耗時平均爲8秒。

實現第三版

因爲方案二，存在共享同一把中央鎖，併發不高的請求。參考concurrentHashMap實現原理，引入分段鎖概念，下降鎖粒度。

基本方式是：

根據sql建立好數據庫
建立100條記錄Flock_name="center_lock_xx"的記錄(xx爲00-99)。
在對其餘鎖(如Flock_name="sale_invite_lock")進行操做的時候，根據crc32算法找到對應的center_lock_02，先對"center_lock_02"記錄select for update
"sale_invite_lock"記錄本身的增刪改查。

僞代碼以下：

// 開啓事務
@Transactional
public boolean getLock(String key){
      // 獲取中央鎖
      select * from tbl where Flock_name="center_lock"    
    
     // 查詢key相關記錄
     select for update if (記錄存在){
           update
     }else {
           insert 
   }
}
複製代碼

/** * 獲取中央鎖Key */
    private boolean getCenterLock(String key){
        String prefix = "center_lock_";
        Long hash = SecurityUtil.crc32(key);
        if (null == hash){
            return false;
        }
        //取crc32中的最後兩位值
        Integer len = hash.toString().length();
        String slot = hash.toString().substring(len-2);

        String centerLockKey = prefix + slot;
        lockMapper.queryRecordForUpdate(centerLockKey);
        return true;
    }

      /** * 獲取鎖 */
    @Override
    @Transactional
    public GetLockResponse getLock(GetLockRequest request) {
        // 檢測參數
        if(StringUtils.isEmpty(request.lockName)) {
            ResultUtil.throwBusinessException(CreditCardErrorCode.PARAM_INVALID);
        }

        // 兼容參數初始化
        request.expireTime = null==request.expireTime? 31536000: request.expireTime;
        request.desc = Strings.isNullOrEmpty(request.desc)?"":request.desc;
        Long nowTime = new Date().getTime();

        GetLockResponse response = new GetLockResponse();
        response.lock = 0;

        // 獲取中央鎖，初始化記錄
        getCenterLock(request.lockName);
        LockRecord lockRecord = initLockRecord(request.lockName);

        // 未釋放鎖或未過時，獲取失敗
        if (lockRecord.getStatus() == 1
                && lockRecord.getDeadline().getTime() > nowTime){
            return response;
        }

        // 獲取鎖
        Date deadline = new Date(nowTime + request.expireTime*1000);
        int num = lockMapper.updateRecord(request.lockName, deadline, 0, request.desc, 1);
        response.lock = 1;
        return response;
    }
複製代碼