關(guān)于分布式鎖(Redisson)的原理分析

更新時(shí)間：2022年08月26日 10:06:15 作者：頭未禿

這篇文章主要介紹了關(guān)于分布式鎖(Redisson)的原理，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

1、分布式鎖場(chǎng)景

互聯(lián)網(wǎng)秒殺
搶優(yōu)惠卷
接口冪等性校驗(yàn)

1.1 案例1

如下代碼模擬了下單減庫(kù)存的場(chǎng)景，我們分析下在高并發(fā)場(chǎng)景下會(huì)存在什么問(wèn)題

package com.wangcp.redisson;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class IndexController {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 模擬下單減庫(kù)存的場(chǎng)景
     * @return
     */
    @RequestMapping(value = "/duduct_stock")
    public String deductStock(){
        // 從redis 中拿當(dāng)前庫(kù)存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功，剩余庫(kù)存：" + realStock);
        }else{
            System.out.println("扣減失敗，庫(kù)存不足");
        }
        return "end";
    }
}

假設(shè)在redis中庫(kù)存（stock）初始值是100。

現(xiàn)在有5個(gè)客戶端同時(shí)請(qǐng)求該接口，可能就會(huì)存在同時(shí)執(zhí)行

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

這行代碼，獲取到的值都為100，緊跟著判斷大于0后都進(jìn)行-1操作，最后設(shè)置到redis 中的值都為99。但正常執(zhí)行完成后redis中的值應(yīng)為 95。

1.2 案例2-使用synchronized 實(shí)現(xiàn)單機(jī)鎖

在遇到案例1的問(wèn)題后，大部分人的第一反應(yīng)都會(huì)想到加鎖來(lái)控制事務(wù)的原子性，如下代碼所示：

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

現(xiàn)在當(dāng)有多個(gè)請(qǐng)求訪問(wèn)該接口時(shí)，同一時(shí)刻只有一個(gè)請(qǐng)求可進(jìn)入方法體中進(jìn)行庫(kù)存的扣減，其余請(qǐng)求等候。

但我們都知道，synchronized 鎖是屬于JVM級(jí)別的，也就是我們俗稱的“單機(jī)鎖”。但現(xiàn)在基本大部分公司使用的都是集群部署，現(xiàn)在我們思考下以上代碼在集群部署的情況下還能保證庫(kù)存數(shù)據(jù)的一致性嗎？

答案是不能，如上圖所示，請(qǐng)求經(jīng)Nginx分發(fā)后，可能存在多個(gè)服務(wù)同時(shí)從Redis中獲取庫(kù)存數(shù)據(jù)，此時(shí)只加synchronized （單機(jī)鎖）是無(wú)效的，并發(fā)越高，出現(xiàn)問(wèn)題的幾率就越大。

1.3 案例3-使用redis的SETNX實(shí)現(xiàn)分布式鎖

setnx：將 key 的值設(shè)為 value，當(dāng)且僅當(dāng) key 不存在。

若給定 key 已經(jīng)存在，則 setnx 不做任何動(dòng)作。

使用setnx實(shí)現(xiàn)簡(jiǎn)單的分布式鎖：

/**
 * 模擬下單減庫(kù)存的場(chǎng)景
 * @return
 */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";
    // 使用 setnx 添加分布式鎖
    // 返回 true 代表之前redis中沒(méi)有key為 lockKey 的值，并已進(jìn)行成功設(shè)置
    // 返回 false 代表之前redis中已經(jīng)存在 lockKey 這個(gè)key了
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
    if(!result){
        // 代表已經(jīng)加鎖了
        return "error_code";
    }
    // 從redis 中拿當(dāng)前庫(kù)存的值
    int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
    if(stock > 0){
        int realStock = stock - 1;
        stringRedisTemplate.opsForValue().set("stock",realStock + "");
        System.out.println("扣減成功，剩余庫(kù)存：" + realStock);
    }else{
        System.out.println("扣減失敗，庫(kù)存不足");
    }
    // 釋放鎖
    stringRedisTemplate.delete(lockKey);
    return "end";
}

我們知道 Redis 是單線程執(zhí)行，現(xiàn)在再看案例2中的流程圖時(shí)，哪怕高并發(fā)場(chǎng)景下多個(gè)請(qǐng)求都執(zhí)行到了setnx的代碼，redis會(huì)根據(jù)請(qǐng)求的先后順序進(jìn)行排列，只有排列在隊(duì)頭的請(qǐng)求才能設(shè)置成功。其它請(qǐng)求只能返回“error_code”。

當(dāng)setnx設(shè)置成功后，可執(zhí)行業(yè)務(wù)代碼對(duì)庫(kù)存扣減，執(zhí)行完成后對(duì)鎖進(jìn)行釋放。

我們?cè)賮?lái)思考下以上代碼已經(jīng)完美實(shí)現(xiàn)分布式鎖了嗎？能夠支撐高并發(fā)場(chǎng)景嗎？答案并不是，上面的代碼還是存在很多問(wèn)題的，離真正的分布式鎖還差的很遠(yuǎn)。我們分析下以上代碼存在的問(wèn)題：

死鎖：假如第一個(gè)請(qǐng)求在setnx加鎖完成后，執(zhí)行業(yè)務(wù)代碼時(shí)出現(xiàn)了異常，那釋放鎖的代碼就無(wú)法執(zhí)行，后面所有的請(qǐng)求也都無(wú)法進(jìn)行操作了。

針對(duì)死鎖的問(wèn)題，我們對(duì)代碼再次進(jìn)行優(yōu)化，添加try-finally，在finally中添加釋放鎖代碼，這樣無(wú)論如何都會(huì)執(zhí)行釋放鎖代碼，如下所示：

/**
     * 模擬下單減庫(kù)存的場(chǎng)景
     * @return
     */
@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";
    try{
        // 使用 setnx 添加分布式鎖
        // 返回 true 代表之前redis中沒(méi)有key為 lockKey 的值，并已進(jìn)行成功設(shè)置
        // 返回 false 代表之前redis中已經(jīng)存在 lockKey 這個(gè)key了
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "wangcp");
        if(!result){
            // 代表已經(jīng)加鎖了
            return "error_code";
        }
        // 從redis 中拿當(dāng)前庫(kù)存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功，剩余庫(kù)存：" + realStock);
        }else{
            System.out.println("扣減失敗，庫(kù)存不足");
        }
    }finally {
        // 釋放鎖
        stringRedisTemplate.delete(lockKey);
    }
    return "end";
}

經(jīng)過(guò)改進(jìn)后的代碼是否還存在問(wèn)題呢？我們思考正常執(zhí)行的情況下應(yīng)該是沒(méi)有問(wèn)題，但我們假設(shè)請(qǐng)求在執(zhí)行到業(yè)務(wù)代碼時(shí)服務(wù)突然宕機(jī)了，或者正巧你的運(yùn)維同事重新發(fā)版，粗暴的 kill -9 掉了呢，那代碼還能執(zhí)行 finally 嗎？

1.4 案例4-加入過(guò)期時(shí)間

針對(duì)想到的問(wèn)題，對(duì)代碼再次進(jìn)行優(yōu)化，加入過(guò)期時(shí)間，這樣即便出現(xiàn)了上述的問(wèn)題，在時(shí)間到期后鎖也會(huì)自動(dòng)釋放掉，不會(huì)出現(xiàn)“死鎖”的情況。

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";
    try{
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
        if(!result){
            // 代表已經(jīng)加鎖了
            return "error_code";
        }
        // 從redis 中拿當(dāng)前庫(kù)存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功，剩余庫(kù)存：" + realStock);
        }else{
            System.out.println("扣減失敗，庫(kù)存不足");
        }
    }finally {
        // 釋放鎖
        stringRedisTemplate.delete(lockKey);
    }
    return "end";
}

現(xiàn)在我們?cè)偎伎家幌?，給鎖加入過(guò)期時(shí)間后就可以了嗎？就可以完美運(yùn)行不出問(wèn)題了嗎？

超時(shí)時(shí)間設(shè)置的10s真的合適嗎？如果不合適設(shè)置多少秒合適呢？如下圖所示

假設(shè)同一時(shí)間有三個(gè)請(qǐng)求。

請(qǐng)求1首先加鎖后需執(zhí)行15秒，但在執(zhí)行到10秒時(shí)鎖失效釋放。
請(qǐng)求2進(jìn)入后加鎖執(zhí)行，在請(qǐng)求2執(zhí)行到5秒時(shí)，請(qǐng)求1執(zhí)行完成進(jìn)行鎖釋放，但此時(shí)釋放掉的是請(qǐng)求2的鎖。
請(qǐng)求3在請(qǐng)求2執(zhí)行5秒時(shí)開(kāi)始執(zhí)行，但在執(zhí)行到3秒時(shí)請(qǐng)求2執(zhí)行完成將請(qǐng)求3的鎖進(jìn)行釋放。

我們現(xiàn)在只是模擬3個(gè)請(qǐng)求便可看出問(wèn)題，如果在真正高并發(fā)的場(chǎng)景下，可能鎖就會(huì)面臨“一直失效”或“永久失效”。

那么具體問(wèn)題出在哪里呢？總結(jié)為以下幾點(diǎn)：

1.存在請(qǐng)求釋放鎖時(shí)釋放掉的并不是自己的鎖
2.超時(shí)時(shí)間過(guò)短，存在代碼未執(zhí)行完便自動(dòng)釋放

針對(duì)問(wèn)題我們思考對(duì)應(yīng)的解決方法：

針對(duì)問(wèn)題1，我們想到在請(qǐng)求進(jìn)入時(shí)生成一個(gè)唯一id，使用該唯一id作為鎖的value值，釋放時(shí)先進(jìn)行獲取比對(duì)，比對(duì)相同時(shí)再進(jìn)行釋放，這樣就可以解決釋放掉其它請(qǐng)求鎖的問(wèn)題。
針對(duì)問(wèn)題2，我們可使用延長(zhǎng)過(guò)期時(shí)間。

1.5 案例5-使用唯一id作為鎖的value值

針對(duì)想到的問(wèn)題，對(duì)代碼再次進(jìn)行優(yōu)化，使用唯一id作為鎖的value值，這樣便不存在請(qǐng)求釋放鎖時(shí)釋放掉的并不是自己的鎖。

@RequestMapping(value = "/duduct_stock")
public String deductStock(){
    String lockKey = "product_001";
    try{
    	//1、占分布式鎖。去redis占坑并設(shè)置過(guò)期時(shí)間 setIfAbsent()操作是原子性的
        final String uuid = UUID.randomUUID().toString();
        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,uuid,30,TimeUnit.SECONDS);
        if(!result){
            // 代表已經(jīng)加鎖了
            return "error_code";
        }
        // 從redis 中拿當(dāng)前庫(kù)存的值
        int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
        if(stock > 0){
            int realStock = stock - 1;
            stringRedisTemplate.opsForValue().set("stock",realStock + "");
            System.out.println("扣減成功，剩余庫(kù)存：" + realStock);
        }else{
            System.out.println("扣減失敗，庫(kù)存不足");
        }
    }finally {
        // 釋放鎖 使用lua腳本解鎖  保證原子性
        String script = "if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
        stringRedisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), Arrays.asList(lockKey), uuid);
    }
    return "end";
}

但是我們思考一下，不斷的延長(zhǎng)過(guò)期時(shí)間真的合適嗎？設(shè)置短了存在超時(shí)自動(dòng)釋放的問(wèn)題，設(shè)置長(zhǎng)了又會(huì)出現(xiàn)宕機(jī)后一段時(shí)間鎖無(wú)法釋放的問(wèn)題，雖然不會(huì)再出現(xiàn)“死鎖”。針對(duì)這個(gè)問(wèn)題，如何解決呢？

我們應(yīng)該要開(kāi)啟一個(gè)守護(hù)線程進(jìn)行監(jiān)聽(tīng)。將超時(shí)時(shí)間設(shè)置默認(rèn)30s，線程每10s調(diào)用一次判斷鎖還是否存在，如果存在則延長(zhǎng)鎖的超時(shí)時(shí)間。

1.6 案例6-Redisson分布式鎖

SpringBoot集成Redisson步驟

引入依賴

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

初始化客戶端

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

Redisson實(shí)現(xiàn)分布式鎖

@RestController
public class IndexController {
    @Autowired
    private RedissonClient redisson;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /**
     * 模擬下單減庫(kù)存的場(chǎng)景
     * @return
     */
    @RequestMapping(value = "/duduct_stock")
    public String deductStock(){
        String lockKey = "product_001";
        // 1.獲取鎖對(duì)象
        RLock redissonLock = redisson.getLock(lockKey);
        try{
            // 2.加鎖
            redissonLock.lock();  // 等價(jià)于 setIfAbsent(lockKey,"wangcp",10,TimeUnit.SECONDS);
            // 從redis 中拿當(dāng)前庫(kù)存的值
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));
            if(stock > 0){
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock",realStock + "");
                System.out.println("扣減成功，剩余庫(kù)存：" + realStock);
            }else{
                System.out.println("扣減失敗，庫(kù)存不足");
            }
        }finally {
            // 3.釋放鎖
            redissonLock.unlock();
        }
        return "end";
    }
}

Redisson 分布式鎖實(shí)現(xiàn)原理圖

Redisson 底層源碼分析

我們點(diǎn)擊 lock() 和 unlock() 方法，查看源碼，最終看到以下代碼

//加鎖 
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
            "if (redis.call('exists', KEYS[1]) == 0) then " +
                    "redis.call('hincrby', 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.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
}
//解鎖
protected RFuture<Boolean> unlockInnerAsync(long threadId) {
    return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
            "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                    "return nil;" +
                    "end; " +
                    "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                    "if (counter > 0) then " +
                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                    "return 0; " +
                    "else " +
                    "redis.call('del', KEYS[1]); " +
                    "redis.call('publish', KEYS[2], ARGV[1]); " +
                    "return 1; " +
                    "end; " +
                    "return nil;",
            Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
}

沒(méi)錯(cuò)，加鎖最終執(zhí)行的就是這段 lua 腳本語(yǔ)言。

這段lua腳本命令在Redis中執(zhí)行時(shí)，會(huì)被當(dāng)成一條命令來(lái)執(zhí)行，能夠保證原子性，故要不都成功，要不都失敗。

我們?cè)谠创a中看到Redssion的許多方法實(shí)現(xiàn)中很多都用到了lua腳本，這樣能夠極大的保證命令執(zhí)行的原子性。

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;

腳本的主要邏輯為：

exists 判斷 key 是否存在
當(dāng)判斷不存在則設(shè)置 key
然后給設(shè)置的key追加過(guò)期時(shí)間

這樣來(lái)看其實(shí)和我們前面案例5中的實(shí)現(xiàn)方法本質(zhì)沒(méi)啥區(qū)別，都是使用底層都是lua。只不過(guò)redisson做了更多的判斷，考慮的更加的周全。而且他還完善了我們案例5中的缺陷，他實(shí)現(xiàn)了一個(gè)看門狗機(jī)制。

Redisson鎖"看門狗"源碼

private void renewExpiration() {
    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
    if (ee == null) {
        return;
    }
    
    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {
            ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
            if (ent == null) {
                return;
            }
            Long threadId = ent.getFirstThreadId();
            if (threadId == null) {
                return;
            }
            
            RFuture<Boolean> future = renewExpirationAsync(threadId);
            future.onComplete((res, e) -> {
                if (e != null) {
                    log.error("Can't update lock " + getRawName() + " expiration", e);
                    EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                    return;
                }
                
                if (res) {
                    // reschedule itself
                    renewExpiration();
                } else {
                    cancelExpirationRenewal(null);
                }
            });
        }
    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    
    ee.setTimeout(task);
}
protected RFuture<Boolean> renewExpirationAsync(long threadId) {
    return evalWriteAsync(getRawName(), 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.singletonList(getRawName()),
            internalLockLeaseTime, getLockName(threadId));
}

這段代碼是在加鎖后開(kāi)啟一個(gè)守護(hù)線程進(jìn)行監(jiān)聽(tīng)。Redisson超時(shí)時(shí)間默認(rèn)設(shè)置30s，線程每10s調(diào)用一次判斷鎖還是否存在，如果存在則延長(zhǎng)鎖的超時(shí)時(shí)間。

現(xiàn)在，我們?cè)倩剡^(guò)頭來(lái)看看案例5中的加鎖代碼與原理圖，其實(shí)完善到這種程度已經(jīng)可以滿足很多公司的使用了，并且很多公司也確實(shí)是這樣用的。但我們?cè)偎伎枷率欠襁€存在問(wèn)題呢？例如以下場(chǎng)景：

眾所周知 Redis 在實(shí)際部署使用時(shí)都是集群部署的，那在高并發(fā)場(chǎng)景下我們加鎖，當(dāng)把key寫(xiě)入到master節(jié)點(diǎn)后，master還未同步到slave節(jié)點(diǎn)時(shí)master宕機(jī)了，原有的slave節(jié)點(diǎn)經(jīng)過(guò)選舉變?yōu)榱诵碌膍aster節(jié)點(diǎn)，此時(shí)可能就會(huì)出現(xiàn)鎖失效問(wèn)題。
通過(guò)分布式鎖的實(shí)現(xiàn)機(jī)制我們知道，高并發(fā)場(chǎng)景下只有加鎖成功的請(qǐng)求可以繼續(xù)處理業(yè)務(wù)邏輯。那就出現(xiàn)了大伙都來(lái)加鎖，但有且僅有一個(gè)加鎖成功了，剩余的都在等待。其實(shí)分布式鎖與高并發(fā)在語(yǔ)義上就是相違背的，我們的請(qǐng)求雖然都是并發(fā)，但Redis幫我們把請(qǐng)求進(jìn)行了排隊(duì)執(zhí)行，也就是把我們的并行轉(zhuǎn)為了串行。串行執(zhí)行的代碼肯定不存在并發(fā)問(wèn)題了，但是程序的性能肯定也會(huì)因此受到影響。

針對(duì)這些問(wèn)題，我們?cè)俅嗡伎冀鉀Q方案

在思考解決方案時(shí)我們首先想到CAP原則（一致性、可用性、分區(qū)容錯(cuò)性），那么現(xiàn)在的Redis就是滿足AP(可用性、分區(qū)容錯(cuò)性)，如果想要解決該問(wèn)題我們就需要尋找滿足CP(一致性、分區(qū)容錯(cuò)性)的分布式系統(tǒng)。首先想到的就是zookeeper，zookeeper的集群間數(shù)據(jù)同步機(jī)制是當(dāng)主節(jié)點(diǎn)接收數(shù)據(jù)后不會(huì)立即返回給客戶端成功的反饋，它會(huì)先與子節(jié)點(diǎn)進(jìn)行數(shù)據(jù)同步，半數(shù)以上的節(jié)點(diǎn)都完成同步后才會(huì)通知客戶端接收成功。并且如果主節(jié)點(diǎn)宕機(jī)后，根據(jù)zookeeper的Zab協(xié)議（Zookeeper原子廣播）重新選舉的主節(jié)點(diǎn)一定是已經(jīng)同步成功的。
那么問(wèn)題來(lái)了，Redisson與zookeeper分布式鎖我們?nèi)绾芜x擇呢？答案是如果并發(fā)量沒(méi)有那么高，可以用zookeeper來(lái)做分布式鎖，但是它的并發(fā)能力遠(yuǎn)遠(yuǎn)不如Redis。如果你對(duì)并發(fā)要求比較高的話，那就用Redis，偶爾出現(xiàn)的主從架構(gòu)鎖失效的問(wèn)題其實(shí)是可以容忍的。
關(guān)于第二個(gè)提升性能的問(wèn)題，我們可以參考ConcurrentHashMap的鎖分段技術(shù)的思想，例如我們代碼的庫(kù)存量當(dāng)前為1000，那我們可以分為10段，每段100，然后對(duì)每段分別加鎖，這樣就可以同時(shí)執(zhí)行10個(gè)請(qǐng)求的加鎖與處理，當(dāng)然有要求的同學(xué)還可以繼續(xù)細(xì)分。但其實(shí)Redis的Qps已經(jīng)達(dá)到10W+了，沒(méi)有特別高并發(fā)量的場(chǎng)景下也是完全夠用的。

以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: