背景
小明在一家在線購物商城工作，最近來了一個新需求，需要他負責開發一個商品秒殺模塊，而且需求很緊急，老板要求必須盡快上線。
方案
小明一開始是這么做的，直接用數據庫鎖進行控制，獲取秒殺商品數量並加鎖，如果數量大於零則成功，否則秒殺失敗。
@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~。本故事純屬虛構，誰也不怪，如有雷同，純屬巧合。