大型車禍現場，電商秒殺超賣，這個鍋到底有誰來背？

背景

小明在一家在線購物商城工作，最近來了一個新需求，需要他負責開發一個商品秒殺模塊，而且需求很緊急，老闆要求必須儘快上線。

方案

小明一開始是這麼做的，直接用數據庫鎖進行控制，獲取秒殺商品數量並加鎖，如果數量大於零則成功，否則秒殺失敗。

 @Override
 @Transactional
 public Result startSeckilDBPCC_ONE(long seckillId, long userId) {
 //獲取秒殺商品數量並加鎖
 String nativeSql = "SELECT number FROM seckill WHERE seckill_id=? FOR UPDATE";
 Object object = dynamicQuery.nativeQueryObject(nativeSql, new Object[]{seckillId});
 Long number = ((Number) object).longValue();
 if(number>0){
 nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=?";
 dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
 SuccessKilled killed = new SuccessKilled();
 killed.setSeckillId(seckillId);
 killed.setUserId(userId);
 killed.setState((short)0);
 killed.setCreateTime(new Timestamp(new Date().getTime()));
 dynamicQuery.save(killed);
 return Result.ok(SeckillStatEnum.SUCCESS);
 }else{
 return Result.error(SeckillStatEnum.END);
 }
 }

寫了併發線程，跑了一下，沒問題，搞定！但是，小明轉頭一想，老闆曾經說過，這次活動宣傳力度很大，有可能會有很多用戶參與活動。恰好項目中使用了 Redis 作為緩存，何不借用一下 Redis 的發佈訂閱功能，實現秒殺隊列，從而減輕後端數據庫的訪問壓力，提升服務性能！這可是個升職加薪，當上總經理，出任CTO，迎娶白富美的好機會。說幹就幹，複製、黏貼一把擼，很快小明就把消息隊列方案搞定了。

事故

開發、測試、上線一條龍，活動開始了，秒殺商品是 100 部蘋果手機，活動結束以後，居然產生了 106 個訂單！老闆很生氣，後果很嚴重，這個鍋必須有人得背，嚇得小明趕緊仔細複查複製粘貼的代碼。

監聽配置 RedisSubListenerConfig ：

@Configuration
public class RedisSubListenerConfig {
 //初始化監聽器
 @Bean
 RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
 MessageListenerAdapter listenerAdapter) {
 RedisMessageListenerContainer container = new RedisMessageListenerContainer();
 container.setConnectionFactory(connectionFactory);
 container.addMessageListener(listenerAdapter, new PatternTopic("seckill"));
 return container;
 }
 //利用反射來創建監聽到消息之後的執行方法
 @Bean
 MessageListenerAdapter listenerAdapter(RedisConsumer redisReceiver) {
 return new MessageListenerAdapter(redisReceiver, "receiveMessage");
 }
 //使用默認的工廠初始化redis操作模板
 @Bean
 StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
 return new StringRedisTemplate(connectionFactory);
 }
}

生產者 RedisSender：

/** 

 * 生產者
 * @author 爪哇筆記 By https://blog.52itstyle.vip
 */
@Service
public class RedisSender {
 @Autowired
 private StringRedisTemplate stringRedisTemplate;
 public void sendChannelMess(String channel, String message) {
 stringRedisTemplate.convertAndSend(channel, message);
 }
}

消費者 RedisConsumer：

/**
 * 消費者
 * @author 爪哇筆記 By https://blog.52itstyle.vip
 */
@Service
public class RedisConsumer {
 
 @Autowired
 private ISeckillService seckillService;
 @Autowired
 private RedisUtil redisUtil;
 
 public void receiveMessage(String message) {
 //收到通道的消息之後執行秒殺操作
 String[] array = message.split(";");
 if(redisUtil.getValue(array[0])==null){//control層已經判斷了，其實這裡不需要再判斷了
 Result result = seckillService.startSeckilDBPCC_TWO(Long.parseLong(array[0]), Long.parseLong(array[1]));
 if(result.equals(Result.ok(SeckillStatEnum.SUCCESS))){
 WebSocketServer.sendInfo(array[0], "秒殺成功");//推送給前臺
 }else{
 WebSocketServer.sendInfo(array[0], "秒殺失敗");//推送給前臺
 redisUtil.cacheValue(array[0], "ok");//秒殺結束
 }
 }else{
 WebSocketServer.sendInfo(array[0], "秒殺失敗");//推送給前臺
 }
 } 

}

數據層代碼：

@Override
@Transactional
public Result startSeckil(long seckillId,long userId) {
 //由於使用了隊列，小明這裡沒用數據庫鎖
 String nativeSql = "SELECT number FROM seckill WHERE seckill_id=?";
 Object object = dynamicQuery.nativeQueryObject(nativeSql, new Object[]{seckillId});
 Long number = ((Number) object).longValue();
 if(number>0){
 //扣庫存
 nativeSql = "UPDATE seckill SET number=number-1 WHERE seckill_id=?";
 dynamicQuery.nativeExecuteUpdate(nativeSql, new Object[]{seckillId});
 //創建訂單
 SuccessKilled killed = new SuccessKilled();
 killed.setSeckillId(seckillId);
 killed.setUserId(userId);
 killed.setState((short)0);
 Timestamp createTime = new Timestamp(new Date().getTime());
 killed.setCreateTime(createTime);
 dynamicQuery.save(killed);
 //支付
 return Result.ok(SeckillStatEnum.SUCCESS);
 }else{
 return Result.error(SeckillStatEnum.END);
 }
}

小明重新審讀了代碼，一開始小明覺得既然使用了隊列，數據庫層面就沒必要用數據庫鎖了，然後去掉了 for update，很顯然問題就出在這裡。導致超賣的因素只有一個，那就是多線程併發搶佔資源，如果業務邏輯沒有做相應的措施，很有可能導致超賣。

回到代碼來看，雖然秒殺用戶進入了隊列，但是 RedisConsumer 端有可能是多線程處理隊列數據，小明為了驗證想法，在消費端加入了以下代碼來打印線程名稱。

Thread th=Thread.currentThread();
System.out.println("Tread name:"+th.getName());

再次運行任務，果不其然，每個秒殺用戶都開啟了一個線程處理任務：

Tread name:container-1
Tread name:container-2
Tread name:container-3
Tread name:container-4
Tread name:container-5
Tread name:container-6
......

各位看官到這裡，線索已經很明確了，我們只需要把消費端改造成單線程處理，問題就迎刃而解了。

解決方案

使用 Redis 消息隊列，出現超賣問題是因為RedisMessageListenerContainer 的默認使用線程池是SimpleAsyncTaskExecutor，每次消費都會創建一個線程來處理，這樣就會有大量的新線程被創建。有興趣的小夥伴可以跟進源碼，瞭解更多詳細內容。

監聽配置 RedisSubListenerConfig 改造為 ：

@Bean
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
 MessageListenerAdapter listenerAdapter) {
 RedisMessageListenerContainer container = new RedisMessageListenerContainer();
 container.setConnectionFactory(connectionFactory);
 container.addMessageListener(listenerAdapter, new PatternTopic("seckill"));
 /**
 * 如果不定義線程池，每一次消費都會創建一個線程，如果業務層面不做限制，就會導致秒殺超賣。 

 * 此處感謝網友 DIscord
 */
 ThreadFactory factory = new ThreadFactoryBuilder()
 .setNameFormat("redis-listener-pool-%d").build();
 Executor executor = new ThreadPoolExecutor(
 1,
 1,
 5L,
 TimeUnit.SECONDS,
 new LinkedBlockingQueue<>(1000),
 factory);
 container.setTaskExecutor(executor);
 return container;
}

然後測試改造效果：

Tread name:redis-listener-pool-0
Tread name:redis-listener-pool-0
Tread name:redis-listener-pool-0
......

小結

那麼問題來了，這個鍋到底誰來背，開發、測試還是產品？這麼好的宣傳機會，直接上頭條"XX 電商系統 bug 超賣，虧損超 10W 仍堅持發貨，稱不能虧了消費者"然後超的錢相關責任人擔一部分， perfect~。本故事純屬虛構，誰也不怪，如有雷同，純屬巧合。

源碼

分佈式秒殺現場：https://gitee.com/52itstyle/spring-boot-seckill