分佈式鎖:5 個案例,附源碼

hi,大家好,我是老田

今天給大家分享的是分佈式鎖,本文使用五個案例 + 圖 + 源碼分析等來分析。

常見的 synchronized、Lock 等這些鎖都是基於單個JVM的實現的,如果分佈式場景下怎麼辦呢?這時候分佈式鎖就出現了。

關於分佈式的實現方案,在業界流行的有三種:

1、基於數據庫

2、基於Redis

3、基於Zookeeper

另外,還有使用etcdconsul來實現的。

在開發中使用最多的是RedisZookeeper兩種方案,並且兩種方案中最複雜的,最容易出問題的就是Redis的實現方案,所以,我們今天就來把Redis實現方案都聊聊。

本文主要內容

分佈式鎖場景

估計部分朋友還不太清楚分佈式的使用場景,下面我簡單羅列三種:

案例 1

如下代碼模擬了下單減庫存的場景,我們分析下在高併發場景下會存在什麼問題

@RestController
public class IndexController {

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 模擬下單減庫存的場景
     * @return
     */
    @RequestMapping(value = "/duduct_stock")
    public String deductStock(){
        // 從redis 中拿當前庫存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功,剩餘庫存:" + realStock);
        }else{
            System.out.println("扣減失敗,庫存不足");
        }
        return "end";
    }
}

假設在Redis中庫存(stock)初始值是 100。

現在有 5 個客戶端同時請求該接口,可能就會存在同時執行

int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));

這行代碼,獲取到的值都爲 100,緊跟着判斷大於 0 後都進行 - 1 操作,最後設置到 redis 中的值都爲 99。但正常執行完成後 redis 中的值應爲 95。

案例 2 - 使用 synchronized 實現單機鎖

在遇到案例 1 的問題後,大部分人的第一反應都會想到加鎖來控制事務的原子性,如下代碼所示:

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    synchronized (this){
        // 從redis 中拿當前庫存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功,剩餘庫存:" + realStock);
        }else{
            System.out.println("扣減失敗,庫存不足");
        }
    }
    return "end";
}

現在當有多個請求訪問該接口時,同一時刻只有一個請求可進入方法體中進行庫存的扣減,其餘請求等候。

但我們都知道,synchronized 鎖是屬於 JVM 級別的,也就是我們俗稱的 “單機鎖”。但現在基本大部分公司使用的都是集羣部署,現在我們思考下以上代碼在集羣部署的情況下還能保證庫存數據的一致性嗎?

答案是不能,如上圖所示,請求經 Nginx 分發後,可能存在多個服務同時從 Redis 中獲取庫存數據,此時只加 synchronized (單機鎖)是無效的,併發越高,出現問題的幾率就越大。

案例 3 - 使用 SETNX 實現分佈式鎖

setnx:將 key 的值設爲 value,當且僅當 key 不存在。

若給定 key 已經存在,則 setnx 不做任何動作。

使用 setnx 實現簡單的分佈式鎖:

/**
 * 模擬下單減庫存的場景
 * @return
 */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";
    // 使用 setnx 添加分佈式鎖
    // 返回 true 代表之前redis中沒有key爲 lockKey 的值,並已進行成功設置
    // 返回 false 代表之前redis中已經存在 lockKey 這個key了
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
    if(!result){
        // 代表已經加鎖了
        return "error_code";
    }

    // 從redis 中拿當前庫存的值
    int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
    if(stock > 0){
        int realStock = stock - 1;
        stringRedisTemplate.opsForValue().set("stock",realStock + "");
        System.out.println("扣減成功,剩餘庫存:" + realStock);
    }else{
        System.out.println("扣減失敗,庫存不足");
    }

    // 釋放鎖
    stringRedisTemplate.delete(lockKey);
    return "end";
}

我們知道 Redis 是單線程執行,現在再看案例 2 中的流程圖時,哪怕高併發場景下多個請求都執行到了 setnx 的代碼,redis 會根據請求的先後順序進行排列,只有排列在隊頭的請求才能設置成功。其它請求只能返回 “error_code”。

當 setnx 設置成功後,可執行業務代碼對庫存扣減,執行完成後對鎖進行釋放

我們再來思考下以上代碼已經完美實現分佈式鎖了嗎?能夠支撐高併發場景嗎?答案並不是,上面的代碼還是存在很多問題的,離真正的分佈式鎖還差的很遠。

我們分析一下,上面的代碼存在的問題:

死鎖:假如第一個請求在setnx加鎖完成後,執行業務代碼時出現了異常,那釋放鎖的代碼就無法執行,後面所有的請求也都無法進行操作了。

針對死鎖的問題,我們對代碼再次進行優化,添加try-finally,在finally中添加釋放鎖代碼,這樣無論如何都會執行釋放鎖代碼,如下所示:

/**
     * 模擬下單減庫存的場景
     * @return
     */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";

    try{
        // 使用 setnx 添加分佈式鎖
        // 返回 true 代表之前redis中沒有key爲 lockKey 的值,並已進行成功設置
        // 返回 false 代表之前redis中已經存在 lockKey 這個key了
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
        if(!result){
            // 代表已經加鎖了
            return "error_code";
        }
        // 從redis 中拿當前庫存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功,剩餘庫存:" + realStock);
        }else{
            System.out.println("扣減失敗,庫存不足");
        }
    }finally {
        // 釋放鎖
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

經過改進後的代碼是否還存在問題呢?我們思考正常執行的情況下應該是沒有問題,但我們假設請求在執行到業務代碼時服務突然宕機了,或者正巧你的運維同事重新發版,粗暴的 kill -9 掉了呢,那代碼還能執行 finally 嗎?

案例 4 - 加入過期時間

針對想到的問題,對代碼再次進行優化,加入過期時間,這樣即便出現了上述的問題,在時間到期後鎖也會自動釋放掉,不會出現 “死鎖” 的情況。

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";

    try{
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
        if(!result){
            // 代表已經加鎖了
            return "error_code";
        }
        // 從redis 中拿當前庫存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功,剩餘庫存:" + realStock);
        }else{
            System.out.println("扣減失敗,庫存不足");
        }
    }finally {
        // 釋放鎖
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

現在我們再思考一下,給鎖加入過期時間後就可以了嗎?就可以完美運行不出問題了嗎?

超時時間設置的 10s 真的合適嗎?如果不合適設置多少秒合適呢?如下圖所示

假設同一時間有三個請求。

我們現在只是模擬 3 個請求便可看出問題,如果在真正高併發的場景下,可能鎖就會面臨 “一直失效” 或“永久失效”。

那麼具體問題出在哪裏呢?總結爲以下幾點:

針對問題我們思考對應的解決方法:

案例 5-Redisson 分佈式鎖

Spring Boot集成Redisson步驟

引入依賴

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.6.5</version>
</dependency>

初始化客戶端

@Bean
public RedissonClient redisson(){
    // 單機模式
    Config config = new Config();
    config.useSingleServer().setAddress("redis://192.168.3.170:6379").setDatabase(0);
    return Redisson.create(config);
}

Redisson 實現分佈式鎖

@RestController
public class IndexController {

    @Autowired
    private RedissonClient redisson;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 模擬下單減庫存的場景
     * @return
     */
    @RequestMapping(value = "/duduct_stock")
    public String deductStock(){
        String lockKey = "product_001";
        // 1.獲取鎖對象
        RLock redissonLock = redisson.getLock(lockKey);
        try{
            // 2.加鎖
            redissonLock.lock();  // 等價於 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
            // 從redis 中拿當前庫存的值
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock",realStock + "");
                System.out.println("扣減成功,剩餘庫存:" + realStock);
            }else{
                System.out.println("扣減失敗,庫存不足");
            }
        }finally {
            // 3.釋放鎖
            redissonLock.unlock();
        }
        return "end";
    }
}

Redisson 分佈式鎖實現原理圖

Redisson 底層源碼分析

我們點擊lock()方法,查看源碼,最終看到以下代碼

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);

        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}

沒錯,加鎖最終執行的就是這段lua 腳本語言。

if (redis.call('exists', KEYS[1]) == 0) then 
    redis.call('hset', KEYS[1], ARGV[2], 1); 
    redis.call('pexpire', KEYS[1], ARGV[1]); 
    return nil; 
end;

腳本的主要邏輯爲:

這樣來看其實和我們前面案例中的實現方法好像沒什麼區別,但實際上並不是。

這段lua腳本命令在Redis中執行時,會被當成一條命令來執行,能夠保證原子性,故要不都成功,要不都失敗。

我們在源碼中看到Redssion的許多方法實現中很多都用到了lua腳本,這樣能夠極大的保證命令執行的原子性。

下面是Redisson鎖自動 “續命” 源碼:

private void scheduleExpirationRenewal(final long threadId) {
    if (expirationRenewalMap.containsKey(getEntryName())) {
        return;
    }

    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {

            RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                                                                     "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                                                                     "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                                                                     "return 1; " +
                                                                     "end; " +
                                                                     "return 0;",
                                                                     Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

            future.addListener(new FutureListener<Boolean>() {
                @Override
                public void operationComplete(Future<Boolean> future) throws Exception {
                    expirationRenewalMap.remove(getEntryName());
                    if (!future.isSuccess()) {
                        log.error("Can't update lock " + getName() + " expiration", future.cause());
                        return;
                    }

                    if (future.getNow()) {
                        // reschedule itself
                        scheduleExpirationRenewal(threadId);
                    }
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

    if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
        task.cancel();
    }
}

這段代碼是在加鎖後開啓一個守護線程進行監聽Redisson超時時間默認設置 30s,線程每 10s 調用一次判斷鎖還是否存在,如果存在則延長鎖的超時時間。

現在,我們再回過頭來看看案例 5 中的加鎖代碼與原理圖,其實完善到這種程度已經可以滿足很多公司的使用了,並且很多公司也確實是這樣用的。但我們再思考下是否還存在問題呢?例如以下場景:

針對這些問題,我們再次思考解決方案

參考:www.jianshu.com/p/bc4ff4694cf3

本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源https://mp.weixin.qq.com/s/pPxOogzJnH8sa83zrAd7Pg