優(yōu)惠券超發(fā)事故：扣了我3個月績效...
文章來源：https://c1n.cn/Xz8kf
問題拋出
問題引發(fā)
問題解決
問題拋出
在近期的項目里面有一個功能是領(lǐng)取優(yōu)惠券的功能。
問題描述：每一個優(yōu)惠券一共發(fā)行多少張，每個用戶可以領(lǐng)取多少張，如：A 優(yōu)惠券一共發(fā)行 120 張，每一個用戶可以領(lǐng)取 140 張。
<!--減優(yōu)惠券庫存的SQL-->
<update id="reduceStock">
update coupon set stock = stock - 1 where id = #{coupon_id}
</update>
上面的代碼按照我們的邏輯是沒有問題，我通過使用 PostMan 軟件測試也是沒有問題，但是上面的代碼確實是有問題的。
在測試的時候是測試的 id 為 19 的這條數(shù)據(jù)，測試完之后這里的總發(fā)行數(shù)量（stock）居然變成了 -1（也就是超發(fā)了一張）。
問題引發(fā)
上面這張圖是整個領(lǐng)取優(yōu)惠券的流程（上圖并沒有使用流程圖來畫，我覺的這樣畫可能表達更清楚一些），在藍色的框那里就是出現(xiàn)超扣減庫存的時候。為啥這樣說呢？
如果同時來了兩個線程（你可以理解成是兩個請求），比如先來的那個請求通過了檢查（線程 A），這時線程 A 還沒有扣減庫存，這時線程 B 經(jīng)過一翻操作也通過了這個檢查優(yōu)惠券是否可領(lǐng)取的方法，然后線程 A 和線程 B 依次扣減庫存或者是同時扣減庫存。
清楚了問題引發(fā)的原因，那就來看看如何解決它們。
問題解決
| 解決方案 1（Java 代碼加鎖）
在引起超發(fā)原因的那張圖內(nèi)可以看出，導致這一問題的根本原因是多個線程同時訪問這個領(lǐng)取優(yōu)惠券的方法，那只要保證在同一段只有一個線程進入到這個方法就可以了。
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)過 Jmeter 的壓測優(yōu)惠券并沒有出現(xiàn)超發(fā)的情況。
雖然這樣可以解決超發(fā)的問題，但是在項目中我們不可以這樣寫，原因如下：
synchronized 的作用范圍是單個 JVM 實例，如果是集群部署系統(tǒng)這里的加鎖你可以理解成失效。
在使用了 synchronized 加鎖后，就會形成串行等待的問題，當一個線程 A 在領(lǐng)取優(yōu)惠券方法內(nèi)執(zhí)行過久時，其它線程會等待直到線程 A 執(zhí)行結(jié)束。
| 解決方案 2（SQL 層面解決超發(fā)）
<update id="reduceStock">
update coupon set stock = stock - 1 where id = #{coupon_id} and stock > 0
</update>
MySQL 默認使用的是 InnoDB 引擎，使用 InnoDB 時在修改某一個記錄的時候會將這條記錄上鎖，所以這個修改數(shù)據(jù)時不會出現(xiàn)多個線程同時修改數(shù)據(jù)。這樣也可以避免優(yōu)惠券超發(fā)。
如果在業(yè)務(wù)中只要有庫存就可以發(fā)放優(yōu)惠券的可以使用上面這種方式。
<update id="reduceStock">
update product set stock=stock-1 where stock=#{上一次的庫存} and id = 1 and stock>0
</update>
上面這種方式會存在 ABA 的問題，當然如果業(yè)務(wù)不在意 ABA 問題可以使用上面的 sql，不過性能可能差一點，如果stock不匹配，這條sql也就失效了。
<update id="reduceStock">
update product set stock=stock-1,versioin = version+1 where id = 1 and stock>0 and version=#{上一次的版本號}
</update>
上面的這三條 SQL 層面的代碼都可以解決優(yōu)惠券超發(fā)的問題，具體使用那種就根據(jù)業(yè)務(wù)來選擇了。
| 解決方案 3（通過 Redis 分布式鎖來解決問題）
在分布式鎖中我們應(yīng)該考濾如下：
排他性，在分布式集群中，同一個方法，在同一個時間只能被某一臺機器上的一個線程執(zhí)行
容錯性，當一個線程上鎖后，如果機器突然的宕機，如果不釋放鎖，此時這條數(shù)據(jù)將會被鎖死
還要注意鎖的粒度，鎖的開銷
滿足高可用，高性能，可重入
當這個命令執(zhí)行成功的時候會返回 1，執(zhí)行失敗會返回 0，我們就可以通過這個特性來判斷是否獲取到了鎖。
String key = "lock:coupon:" + couponId;
try{
if(setnx(key,"1")){
//獲取到鎖
//設(shè)置Key的時期時間
exp(key,30,TimeUnit.MILLISECONDS);
try{
//業(yè)務(wù)邏輯
}finally{
del(key);
}
}else{
//獲取鎖失敗，遞歸調(diào)用這個方法，或者使用for進行自旋獲取鎖
}
}
這方法里面設(shè)置 key 的過期時間的原因是，當機器突然的宕機后，即使沒有釋放掉鎖，他也會在一段時間后將這個鎖釋放，避免導致死鎖。
String key = "lock:coupon:" + couponId;
String threadId = Thread.currentThread().getId();
try{
if(setnx(key,threadId)){
//獲取到鎖
//設(shè)置Key的時期時間
exp(key,30,TimeUnit.MILLISECONDS);
try{
//業(yè)務(wù)邏輯
}finally{
if(get(key) == threadId){
del(key);
}
}
}else{
//獲取鎖失敗，遞歸調(diào)用這個方法，或者使用for進行自旋獲取鎖
}
}
通過上面這種方法就可以解決誤刪除 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 其實也可以不用，寫成 uuid 也可以，但是在上面 setnx 的時候，那個值也要寫成 uuid。
但是這樣還要存在一個鎖自動續(xù)期的問題，你可以開一個守護線程，每隔多久給他續(xù)期一次，或者是直接將這個過期時間延長一些。
在 Redis 中也有一些官方推薦的分布式鎖的方式。我最后是使用的這種方式。
| 解決方案 4（使用 Redis 推薦的方式）
https://redis.io/docs/reference/patterns/distributed-locks/
這個有多種實現(xiàn)方式，比如：Golang，Java，PHP。
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.17.4</version>
</dependency>
@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);
}
}
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();
}
通過這種方法也可以解決優(yōu)惠券超發(fā)的問題 ，這也是 Rediss 官網(wǎng)推薦的一種方式。
使用這種方式也無需關(guān)心 key 過期時間續(xù)期的問題，因為在 Redisson 一旦加鎖成功，就會啟動一個 watch dog，你可以將它理解成一個守護線程，它默認會每隔 30 秒檢查一下，如果當前客戶端還占有這把鎖，它會自動對這個鎖的過期時間進行延長。
Config config = new Config();
config.setLockWatchdogTimeout();
如上就是我在解決優(yōu)惠券超發(fā)時的一個思路。
完
往期推薦
SQL優(yōu)化的魅力！從 30248s 到 0.001s
消息隊列消息丟失和消息重復發(fā)送的處理策略
CentOS 停服！我們有哪些頂流的國產(chǎn)操作系統(tǒng)
有道無術(shù)，術(shù)可成；有術(shù)無道，止于術(shù)
歡迎大家關(guān)注Java之道公眾號
好文章，我在看??