大型车祸现场，电商秒杀超卖，这个锅到底有谁来背？

背景

小明在一家在线购物商城工作，最近来了一个新需求，需要他负责开发一个商品秒杀模块，而且需求很紧急，老板要求必须尽快上线。

方案

小明一开始是这么做的，直接用数据库锁进行控制，获取秒杀商品数量并加锁，如果数量大于零则成功，否则秒杀失败。

    @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~。本故事纯属虚构，谁也不怪，如有雷同，纯属巧合。

源码

分布式秒杀现场：