背景
某某在一家在线购物商城工作，最近来了一个新需求，需要他负责开发一个商品秒杀模块，而且需求很紧急，老板要求必须尽快上线。
方案
某某一开始是这么做的，直接用数据库锁进行控制，获取秒杀商品数量并加锁，如果数量大于零则成功，否则秒杀失败。
@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：
/**
* 生产者
*/
@Service
public class RedisSender {
@Autowired
private StringRedisTemplate stringRedisTemplate;
public void sendChannelMess(String channel, String message) {
stringRedisTemplate.convertAndSend(channel, message);
}
}
消费者 RedisConsumer：
/**
* 消费者
*/
@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~。本故事纯属虚构，谁也不怪，如有雷同，纯属巧合。
欢迎工作一到五年的Java工程师朋友们加入JavaQQ群：219571750，群内提供免费的Java架构学习资料（里面有高可用、高并发、高性能及分布式、Jvm性能调优、Spring源码，MyBatis，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多个知识点的架构资料）合理利用自己每一分每一秒的时间来学习提升自己，不要再用"没有时间“来掩饰自己思想上的懒惰！趁年轻，使劲拼，给未来的自己一个交代！
