群友分享的優(yōu)惠券超發(fā)解決思路！

問(wèn)題拋出

在近期的項(xiàng)目里面有一個(gè)功能是領(lǐng)取優(yōu)惠券的功能。

「問(wèn)題描述：」

每一個(gè)優(yōu)惠券一共發(fā)行多少?gòu)垼總€(gè)用戶可以領(lǐng)取多少?gòu)垼?/p>

如：A優(yōu)惠券一共發(fā)行120張，每一個(gè)用戶可以領(lǐng)取140張，當(dāng)一個(gè)用戶領(lǐng)取優(yōu)惠券成功的時(shí)候，把領(lǐng)取的記錄寫(xiě)入到另外一個(gè)表中(這張表我們暫且稱為表B)。

<!--減優(yōu)惠券庫(kù)存的SQL-->
<update id="reduceStock">
     update coupon set stock = stock - 1 where id = #{coupon_id}
</update>

上面的代碼按照我們的邏輯是沒(méi)有問(wèn)題，我通過(guò)使用PostMan軟件測(cè)試也是沒(méi)有問(wèn)題，但是上面的代碼確實(shí)是有問(wèn)題的。

往往我們寫(xiě)的一些業(yè)務(wù)功能，在低并發(fā)的時(shí)候很多的問(wèn)題會(huì)體現(xiàn)不出來(lái)。所以這個(gè)領(lǐng)取優(yōu)惠券的功能我通過(guò)Jmeter軟件來(lái)進(jìn)行壓測(cè)。

?

這里配置了一下，大概會(huì)發(fā)送500次請(qǐng)求，那來(lái)驗(yàn)證下優(yōu)惠券會(huì)不會(huì)出現(xiàn)超發(fā)的問(wèn)題

?

執(zhí)行結(jié)果，里面沒(méi)有出現(xiàn)異常什么的，樣本為500。包括在匯總報(bào)告里面也出現(xiàn)了一些返回信息是「優(yōu)惠券不足的信息」，那來(lái)看下數(shù)據(jù)庫(kù)里面的優(yōu)惠券的總發(fā)行數(shù)量有沒(méi)有變成負(fù)數(shù)呢？也就是有沒(méi)有超發(fā)。

在測(cè)試的時(shí)候是測(cè)試的「id為19」的這條數(shù)據(jù)，測(cè)試完之后這里的總發(fā)行數(shù)量(stock)居然變成了-1(也就是超發(fā)了一張)。

問(wèn)題引發(fā)

在解決這個(gè)問(wèn)題之前，先來(lái)看下這個(gè)問(wèn)題是如何引發(fā)出來(lái)的。

上面這張圖是整個(gè)領(lǐng)取優(yōu)惠券的流程(上圖并沒(méi)有使用流程圖來(lái)畫(huà)，我覺(jué)的這樣畫(huà)可能表達(dá)更清楚一些)，在藍(lán)色的框那里就是出現(xiàn)超扣減庫(kù)存的時(shí)候。為啥這樣說(shuō)呢？

如果同時(shí)來(lái)了兩個(gè)線程(你可以理解成是兩個(gè)請(qǐng)求)，比如先來(lái)的那個(gè)請(qǐng)求通過(guò)了檢查(線程A)，這時(shí)線程A還沒(méi)有扣減庫(kù)存，這時(shí)線程B經(jīng)過(guò)一翻操作也通過(guò)了這個(gè)檢查優(yōu)惠券是否可領(lǐng)取的方法，然后線程A和線程B依次扣減庫(kù)存或者是同時(shí)扣減庫(kù)存，這樣就會(huì)出現(xiàn)優(yōu)惠券超領(lǐng)的情況。

清楚了問(wèn)題引發(fā)的原因，那就來(lái)看看如何解決它們。

解決方案一(Java代碼加鎖)

在引起超發(fā)原因的那張圖內(nèi)可以看出，導(dǎo)致這一問(wèn)題的根本原因是多個(gè)線程同時(shí)訪問(wèn)這個(gè)領(lǐng)取優(yōu)惠券的方法，那只要保證在同一段只有一個(gè)線程進(jìn)入到這個(gè)方法就可以了。

上面貼的代碼就可以改成下面這樣：

synchronized (this){
    LoginUser loginUser = LoginInterceptor.threadLocal.get();
    CouponDO couponDO = couponMapper.selectOne(new QueryWrapper<CouponDO>()
                                    .eq("id", couponId)
                                    .eq("category", categoryEnum.name()));
    if(couponDO == null){
        throw new BizException(BizCodeEnum.COUPON_NO_EXITS);
    }
    this.checkCoupon(couponDO,loginUser.getId());

    //構(gòu)建領(lǐng)券記錄
    CouponRecordDO couponRecordDO = new CouponRecordDO();
    BeanUtils.copyProperties(couponDO,couponRecordDO);
    couponRecordDO.setCreateTime(new Date());
    couponRecordDO.setUseState(CouponStateEnum.NEW.name());
    couponRecordDO.setUserId(loginUser.getId());
    couponRecordDO.setUserName(loginUser.getName());
    couponRecordDO.setCouponId(couponDO.getId());
    couponRecordDO.setId(null);

    int row = couponMapper.reduceStock(couponId);
    if(row == 1){
        couponRecordMapper.insert(couponRecordDO);
    }else{
        log.info("發(fā)送優(yōu)惠券失敗:{},用戶:{}",couponDO,loginUser);
    }
}

這樣，經(jīng)過(guò)Jmeter的壓測(cè)優(yōu)惠券并沒(méi)有出現(xiàn)超發(fā)的情況。

雖然這樣可以解決超發(fā)的問(wèn)題，但是在項(xiàng)目中我們不可以這樣寫(xiě)，原因如下：

• synchronized的作用范圍是單個(gè)JVM實(shí)例，如果是集群部署系統(tǒng)這里的加鎖你可以理解成失效
• 在使用了synchronized加鎖后，就會(huì)形成串行等待的問(wèn)題，當(dāng)一個(gè)線程A在領(lǐng)取優(yōu)惠券方法內(nèi)執(zhí)行過(guò)久時(shí)，其它線程會(huì)等待直到線程A執(zhí)行結(jié)束

解決方案二(Sql層面解決超發(fā))

<update id="reduceStock">
     update coupon set stock = stock - 1 where id = #{coupon_id} and stock > 0
</update>

Mysql默認(rèn)使用的是InnoDB引擎，使用InnoDB時(shí)在修改某一個(gè)記錄的時(shí)候會(huì)將這條記錄上鎖，所以這個(gè)修改數(shù)據(jù)時(shí)不會(huì)出現(xiàn)多個(gè)線程同時(shí)修改數(shù)據(jù)。這樣也可以避免優(yōu)惠券超發(fā)。

如果在業(yè)務(wù)中只要有庫(kù)存就可以發(fā)放優(yōu)惠券的可以使用上面這種方式。

還有一種Sql的方式，可以將stock自身做為樂(lè)觀鎖。

<update id="reduceStock">
     update product set stock=stock-1 where stock=#{上一次的庫(kù)存}  and id = 1 and stock>0
</update>

上面這種方式會(huì)存在ABA的問(wèn)題，當(dāng)然如果業(yè)務(wù)不在意ABA問(wèn)題可以使用上面的sql，不過(guò)性能可能差一點(diǎn)，如果stock不匹配，這條sql也就失效了。

如果業(yè)務(wù)在意ABA問(wèn)題的話也可以在表中加一個(gè)version的字段，每次修改數(shù)據(jù)的時(shí)候這個(gè)字段會(huì)加1，這樣就可以避免ABA問(wèn)題

<update id="reduceStock">
     update product set stock=stock-1,versioin = version+1 where  id = 1 and stock>0 and version=#{上一次的版本號(hào)}
</update>

上面的這三條Sql層面的代碼都可以解決優(yōu)惠券超發(fā)的問(wèn)題，具體使用那種就根據(jù)業(yè)務(wù)來(lái)選擇了

解決方案三(通過(guò)Redis分布式鎖來(lái)解決問(wèn)題)

引入Redis后，當(dāng)領(lǐng)取優(yōu)惠券時(shí)會(huì)先去Redis里面去獲取鎖，當(dāng)鎖獲取成功后才可以對(duì)數(shù)據(jù)庫(kù)進(jìn)行操作

在分布式鎖中我們應(yīng)該考濾如下：

• 排他性，在分布式集群中，同一個(gè)方法，在同一個(gè)時(shí)間只能被某一臺(tái)機(jī)器上的一個(gè)線程執(zhí)行
• 容錯(cuò)性，當(dāng)一個(gè)線程上鎖后，如果機(jī)器突然的宕機(jī)，如果不釋放鎖，此時(shí)這條數(shù)據(jù)將會(huì)被鎖死
• 還要注意鎖的粒度，鎖的開(kāi)銷
• 滿足高可用，高性能，可重入

我們可以使用Redis里面的setnx命令來(lái)設(shè)置鎖，因?yàn)?code style="font-size: 14px;word-wrap: break-word;margin: 0 2px;background-color: rgba(27,31,35,.05);font-family: Operator Mono, Consolas, Monaco, Menlo, monospace;word-break: break-all;color: #3594F7;background: RGBA(59, 170, 250, .1);padding: 0 2px;border-radius: 2px;height: 21px;line-height: 22px;">setnx是原子性的操作不可被打斷

當(dāng)這個(gè)命令執(zhí)行成功的時(shí)候會(huì)返回1，執(zhí)行失敗會(huì)返回0,我們就可以通過(guò)這個(gè)特性來(lái)判斷是否獲取到了鎖。

先看下偽代碼：

String key = "lock:coupon:" + couponId;
try{
 if(setnx(key,"1")){
  //獲取到鎖
  //設(shè)置Key的時(shí)期時(shí)間
  exp(key,30,TimeUnit.MILLISECONDS);
  try{
   //業(yè)務(wù)邏輯
  }finally{
   del(key);
  }
    }else{
        //獲取鎖失敗，遞歸調(diào)用這個(gè)方法，或者使用for進(jìn)行自旋獲取鎖
    }
}

這方法里面設(shè)置key的過(guò)期時(shí)間的原因是，當(dāng)機(jī)器突然的宕機(jī)后，即使沒(méi)有釋放掉鎖，他也會(huì)在一段時(shí)間后將這個(gè)鎖釋放，避免導(dǎo)致死鎖。

雖然看上面的代碼是沒(méi)有問(wèn)題的，但是它是存在一個(gè)誤刪除key的問(wèn)題

為了避免這個(gè)問(wèn)題，可以將setnx命令設(shè)置的那個(gè)值，設(shè)置成當(dāng)前線程的ID,在刪除的時(shí)候判斷這個(gè)線程ID是不是與當(dāng)前線程的Id相同就可以了。

String key = "lock:coupon:" + couponId;
String threadId = Thread.currentThread().getId();
try{
 if(setnx(key,threadId)){
  //獲取到鎖
  //設(shè)置Key的時(shí)期時(shí)間
  exp(key,30,TimeUnit.MILLISECONDS);
  try{
   //業(yè)務(wù)邏輯
  }finally{
   if(get(key) == threadId){
    del(key);
   }
  }
    }else{
        //獲取鎖失敗，遞歸調(diào)用這個(gè)方法，或者使用for進(jìn)行自旋獲取鎖
    }
}

通過(guò)上面這種方法就可以解決誤刪除key的問(wèn)題。

在finally中的這個(gè)判斷和刪除key的代碼不是原子性的，我們可以通過(guò)lua腳本的方式來(lái)實(shí)現(xiàn)它們之間的原子性，將刪除key的代碼修改成如下：

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

redisTemplate.execute(new DefaultRedisScript<>(script, Integer.class), Arrays.asList(key), threadId);

?

這里的threadId其實(shí)也可以不用，寫(xiě)成uuid也可以，但是在上面setnx的時(shí)候，那個(gè)值也要寫(xiě)成uuid

?

但是這樣還要存在一個(gè)鎖自動(dòng)續(xù)期的問(wèn)題，你可以開(kāi)一個(gè)守護(hù)線程，每隔多久給他續(xù)期一次，或者是直接將這個(gè)過(guò)期時(shí)間延長(zhǎng)一些。

在Redis中也有一些官方推薦的分布式鎖的方式。我最后是使用的這種方式

解決方案四(使用Redis推薦的方式)

官網(wǎng)地址：https://redis.io/docs/reference/patterns/distributed-locks/

這個(gè)有多種實(shí)現(xiàn)方式，比如：Golang,Java,Php

引入Redisson包

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

配置RedissoneClient

@Configuration
public class AppConfig {

    @Value("${spring.redis.host}")
    private String redisHost;

    @Value("${spring.redis.port}")
    private String redisPort;

    @Bean
    public RedissonClient redisson(){
        Config config = new Config();
        config.useSingleServer().setAddress("redis://" + redisHost + ":" + redisPort);
        return Redisson.create(config);
    }
}

配置好RedissonClient后，通過(guò)getLock方法獲取到鎖對(duì)象后，在我們的Service層中就可以通過(guò)lock和unlock來(lái)進(jìn)行加鎖和釋放鎖了，這樣還是很方便的。

public JsonData addCoupon(long couponId, CouponCategoryEnum categoryEnum) {
    String key = "lock:coupon:" + couponId;
    RLock rLock = redisson.getLock(key);
    LoginUser loginUser = LoginInterceptor.threadLocal.get();
    rLock.lock();
    try{
       //業(yè)務(wù)邏輯
    }finally {
        rLock.unlock();
    }
    return JsonData.buildSuccess();
}

通過(guò)這種方法也可以解決優(yōu)惠券超發(fā)的問(wèn)題，這也是Rediss官網(wǎng)推薦的一種方式。

使用這種方式也無(wú)需關(guān)心key過(guò)期時(shí)間續(xù)期的問(wèn)題，因?yàn)樵赗edisson一旦加鎖成功，就會(huì)啟動(dòng)一個(gè)watch dog，你可以將它理解成一個(gè)守護(hù)線程，它默認(rèn)會(huì)每隔30秒檢查一下，如果當(dāng)前客戶端還占有這把鎖，它會(huì)自動(dòng)對(duì)這個(gè)鎖的過(guò)期時(shí)間進(jìn)行延長(zhǎng)。

也可以通過(guò)下面的方法設(shè)置watch dog的檢測(cè)時(shí)間間隔

Config config = new Config();
config.setLockWatchdogTimeout();

如上就是微信群友在解決優(yōu)惠券超發(fā)時(shí)的一個(gè)思路。感謝他的熱心分享！