Redis分佈式鎖

時間 2021-02-19

標籤 redis 數據庫網絡併發 app dom 異步分佈式高併發性能欄目 Redis 简体版

原文原文鏈接

1、分佈式鎖簡介

1，什麼是分佈式鎖

當在分佈式模型下，數據只有一份（或有限制），此時須要利用鎖的技術控制某一時刻修改數據的進程數。
與單機模式下的鎖不只須要保證進程可見，還須要考慮進程與鎖之間的網絡問題。
分佈式鎖仍是能夠將標記存在內存，只是該內存不是某個進程分配的內存而是公共內存如 Redis、Memcache。至於利用數據庫、文件等作鎖與單機的實現是同樣的，只要保證標記能互斥就行。

2，分佈式鎖具有的條件

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

2、採用Redis實現分佈式鎖

1，常規代碼實現

@RequestMapping("/deduct_stock")
public String deductStock() {
    String lockKey = "product_001";
    try {
       /*Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "aaa"); //jedis.setnx
        stringRedisTemplate.expire(lockKey, 30, TimeUnit.SECONDS); //設置超時*/
        //爲解決原子性問題將設置鎖和設置超時時間合併
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "aaa", 10, TimeUnit.SECONDS);

        //未設置成功，當前key已經存在了，直接返回錯誤
        if (!result) {
            return "error_code";
        }

        //業務邏輯實現，扣減庫存
        ....
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        stringRedisTemplate.delete(lockKey);
    }
    return "end";
}

2，問題分析

　　上述代碼能夠看到，當前鎖的失效時間爲10s，若是當前扣減庫存的業務邏輯執行須要15s時，高併發時會出現問題：redis

線程1，首先執行到10s後，鎖(product_001)失效
線程2，在第10s後一樣進入當前方法，此時加上鎖(product_001)
當執行到15s時，線程1刪除線程2加的鎖(product_001)
線程3，能夠加鎖 .... 如此循環，實際鎖已經沒有意義

a）方案1：當前線程刪除當前線程所加的鎖

@RequestMapping("/deduct_stock")
public String deductStock() {
    String lockKey = "product_001";
    //定義惟一的客戶端ID
    String clientId = UUID.randomUUID().toString();
    try {
        //爲解決原子性問題將設置鎖和設置超時時間合併,將clientID做爲值放入鎖中
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 10, TimeUnit.SECONDS);

        //未設置成功，當前key已經存在了，直接返回錯誤
        if (!result) {
            return "error_code";
        }

        //業務邏輯實現，扣減庫存
        ....
    } catch (Exception e) {
        e.printStackTrace();
    }finally {
        //只有在獲取鎖的值爲當前clientId時纔會進行刪除鎖操做
        if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
            stringRedisTemplate.delete(lockKey);
        }
    }
    return "end";
}

　　這樣能保證每一個線程刪除的鎖爲當前線程添加的鎖，可是仍是會有超賣的問題：由於線程1在尚未執行完成的時候，此時鎖已經到達過時時間，此時線程2則會加鎖成功數據庫

b）方案2：續命鎖

　　定義一個子線程，定時去查看是否存在主線程的持有當前鎖，若是存在則爲其延長過時時間。網絡

c）方案3：Redisson

@Autowired
Redisson redisson;
@RequestMapping("/deduct_stock_redisson")
public String deductStockRedisson() {
    String lockKey = "product_001";
    RLock rlock = redisson.getLock(lockKey);
    try {
        rlock.lock();

        //業務邏輯實現，扣減庫存
        ....
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        rlock.unlock();
    }
    return "end";
}

多個線程去執行lock操做，僅有一個線程可以加鎖成功，其它線程循環阻塞。
加鎖成功，鎖超時時間默認30s，並開啓後臺線程，加鎖的後臺會每隔10秒去檢測線程持有的鎖是否存在，還在的話，就延遲鎖超時時間，從新設置爲30s，即鎖延期。
對於原子性，Redis分佈式鎖底層藉助Lua腳本實現鎖的原子性。鎖延期是經過在底層用Lua進行延時，延時檢測時間是對超時時間timeout /3

3、採用Redisson分佈式鎖的問題分析

1，主從同步問題

　　當主Redis加鎖了，開始執行線程，若還未將鎖經過異步同步的方式同步到從Redis節點，主節點就掛了，此時會把某一臺從節點做爲新的主節點，此時別的線程就能夠加鎖了，這樣就出錯了，怎麼辦？併發

a）採用zookeeper代替Redis

　　因爲zk集羣的特色，其支持的是CP。而Redis集羣支持的則是AP。app

b）採用RedLock

　　假設有3個redis節點，這些節點之間既沒有主從，也沒有集羣關係。客戶端用相同的key和隨機值在3個節點上請求鎖，請求鎖的超時時間應小於鎖自動釋放時間。當在2個（超過半數）redis上請求到鎖的時候，纔算是真正獲取到了鎖。若是沒有獲取到鎖，則把部分已鎖的redis釋放掉。dom

@RequestMapping("/deduct_stock_redlock")
public String deductStockRedlock() {
    String lockKey = "product_001";
    //TODO 這裏須要本身實例化不一樣redis實例的redisson客戶端鏈接，這裏只是僞代碼用一個redisson客戶端簡化了
    RLock rLock1 = redisson.getLock(lockKey);
    RLock rLock2 = redisson.getLock(lockKey);
    RLock rLock3 = redisson.getLock(lockKey);

    // 向3個redis實例嘗試加鎖
    RedissonRedLock redLock = new RedissonRedLock(rLock1, rLock2, rLock3);
    boolean isLock;
    try {
        // 500ms拿不到鎖, 就認爲獲取鎖失敗。10000ms即10s是鎖失效時間。
        isLock = redLock.tryLock(500, 10000, TimeUnit.MILLISECONDS);
        System.out.println("isLock = " + isLock);
        if (isLock) {
            //業務邏輯處理
            ...
        }
    } catch (Exception e) {

    } finally {
        // 不管如何, 最後都要解鎖
        redLock.unlock();
    }
}