实战分布式之电商高并发秒杀收单核心要点及代码实现

说罢秒杀网关相关的核心要点，我们接着聊聊秒杀收单相关的核心要点与代码实现。

本文重点说明以下几点：

1. 业务场景概述
2. 通过消息队列异步收单
3. 实际库存扣减
4. 实际下单操作

业务场景概述

首先对业务场景进行概述。

完整的业务流可参考 实战分布式之电商高并发秒杀场景总览

秒杀收单核心业务逻辑如下：

1. 秒杀下单消费者从MQ中获取到下单消息，开始下单操作
2. 首先进行下单前的消息幂等校验，对于已经存在的下单消息不予消费
3. 接着进行真实的库存判断，如果库存不够扣减则不再消费，这里应当通过消息推送告知用户商品已售罄，提示用户下次再来
4. 如果库存足够，则扣减库存并下单。这两者在同一个本地事务域中，保证扣减完库存一定能够下单成功
5. 下单成功后，通过消息推送通知用户对秒杀订单进行付款，付款后进行后续的发货等操作

通过消息队列异步收单

接着讲解下如何通过消息队列进行异步收单。

关于如何对消息进行封装，可以参考 实战分布式之电商高并发秒杀网关核心要点及代码实现 , 这里不再赘述。

定义消费者客户端

我们需要定义一个进行下单操作的消费者客户端，并对秒杀收单消息进行订阅。

@PostConstruct
public void init() {
    defaultMQPushConsumer = new DefaultMQPushConsumer(MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getConsumerGroup());
    defaultMQPushConsumer.setNamesrvAddr(nameSrvAddr);
    // 从头开始消费
    defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    // 消费模式:集群模式
    defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
    // 注册监听器
    defaultMQPushConsumer.registerMessageListener(messageListener);
    // 订阅所有消息
    try {
        defaultMQPushConsumer.subscribe(MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getTopic(), "*");
        // 启动消费者
        defaultMQPushConsumer.start();
    } catch (MQClientException e) {
        LOGGER.error("[秒杀下单消费者]--SecKillChargeOrderConsumer加载异常!e={}", LogExceptionWapper.getStackTrace(e));
        throw new RuntimeException("[秒杀下单消费者]--SecKillChargeOrderConsumer加载异常!", e);
    }
    LOGGER.info("[秒杀下单消费者]--SecKillChargeOrderConsumer加载完成!");
}

接着需要实现秒杀收单核心的逻辑，也就是实现我们自己的MessageListenerConcurrently。

@Component
public class SecKillChargeOrderListenerImpl implements MessageListenerConcurrently {

    /**
    * 秒杀核心消费逻辑
    * @param msgs
    * @param context
    * @return
    */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

定义一个类实现接口MessageListenerConcurrently，通过@Component标注为一个SpringBean。

这就是秒杀客户端的主要骨架代码。

进行库存校验

我们虽然在秒杀网关已经对库存进行了校验，但那是不可靠的。

原因在于秒杀网关的库存是在启动时预热的，后续扣减是基于缓存进行的。核心目的在于减少对数据库的压力。

当下单消费者进行真实下单的时候就需要对库存进行校验，此时数据库的压力已经很小了，因为在网关的前置校验已经对流量进行了大幅度的削弱。

我们看一下真实库存校验逻辑

// 库存校验
String prodId = chargeOrderMsgProtocol.getProdId();
SecKillProductDobj productDobj = secKillProductService.querySecKillProductByProdId(prodId);
// 取库存校验
int currentProdStock = productDobj.getProdStock();
if (currentProdStock <= 0) {
    LOGGER.info("[decreaseProdStock]当前商品已售罄,消息消费成功!prodId={},currStock={}", prodId, currentProdStock);
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

此处先对库存进行了查询并校验是否已经小于等于0，如果等于0说明已经售罄，则不再进行消费。

实际下单操作

实际下单操作与实际库存扣减处于同一个本地事务中，核心代码逻辑如下：

// 减库存
if (!secKillProductService.decreaseProdStock(prodId)) {
    LOGGER.info("[insertSecKillOrder]orderId={},prodId={},下单前减库存失败,下单失败!", orderId, prodId);
    // TODO 此处可给用户发送通知，告知秒杀下单失败，原因：商品已售罄
    return false;
}
// 设置产品名称
SecKillProductDobj productInfo = secKillProductService.querySecKillProductByProdId(prodId);
orderInfoDO.setProdName(productInfo.getProdName());
try {
    insertCount = secKillOrderMapper.insertSecKillOrder(orderInfoDO);
} catch (Exception e) {
    LOGGER.error("[insertSecKillOrder]orderId={},秒杀订单入库[异常],事务回滚,e={}", orderId, LogExceptionWapper.getStackTrace(e));
    String message =
            String.format("[insertSecKillOrder]orderId=%s,秒杀订单入库[异常],事务回滚", orderId);
    throw new RuntimeException(message);
}
if (insertCount != 1) {
    LOGGER.error("[insertSecKillOrder]orderId={},秒杀订单入库[失败],事务回滚,e={}", orderId);
    String message =
            String.format("[insertSecKillOrder]orderId=%s,秒杀订单入库[失败],事务回滚", orderId);
    throw new RuntimeException(message);
}
return true;

我们首先进行减库存操作，如果减库存失败，则事务回滚。

如果扣减成功则进行下单操作，下单操作失败则事务回滚，同时对库存进行回滚。此处通过Spring的声明式事务对事务进行处理。使用默认事务传播级别 Propagation.REQUIRED 即可。

小结

本文主要对高并发秒杀场景下的收单部分的重点逻辑进行了讲解，对库存真实扣减、真实下单部分的逻辑和注意点以代码的形式做了较为直观的展现。

到此，实战分布式之电商高并发秒杀场景的系列就暂时告一段落。

实际上，在收单之后还有支付、物流相关的操作，它们都能够通过使用RocketMQ之类的消息引擎进行异步化处理。思路和秒杀下单主逻辑很相似，碍于篇幅就暂时不做讲解了，感兴趣的同学可以参考之前的思路自己实现，聪明的你一定能够举一反三。更多的实战相关的总结与分享，我会及时呈现，那么我们下篇文章再见。