1. 前言
秒杀本质上属于短时突发性高并发访问问题，业务特点如下：
- 定时触发，流量在瞬间突增
- 秒杀请求中常常只有部分能够成功
- 秒杀商品数量往往有限，不能超卖，但能接受少卖
- 不要求立即返回真实下单结果
本文主要讲解秒杀场景中 RocketMQ 实战使用，不详细讲解秒杀其他业务流程。
下面是秒杀流程图：
想要了解具体实现的，参见详细代码：大佬源码
2. 秒杀业务概述
通过对秒杀核心业务流程进行异步化，我们能够将主流程分为收单、下单两个阶段。
2.1 秒杀流程--收单
- 用户访问秒杀入口，将秒杀请求提交给秒杀平台收单网关，平台对秒杀请求进行前置校验
- 校验通过后，将下单请求通过缓存/队列/线程池等中间层进行提交，在投递完成同时的同时就给用户返回“排队中”
- 对于前置校验失败的下单请求同步返回秒杀下单失败
到此，对用户侧的交互就告一段落。
收单过程中，将秒杀订单放入 RocketMQ 中间层中。
2.2 秒杀流程--下单
下单流程中，平台的压力通过中间层的缓冲其实已经小了很多，之所以会少，一方面是因为在用户下单的同步校验过程中就过滤掉了部分非法请求；另一方面，我们通过在中间层做一些限流、过滤等逻辑对下单请求做限速、压单等操作，将下单请求在内部慢慢消化，尽可能减少流量对平台持久层的冲击。这里其实就体现了中间层 “削峰填谷” 的特点。
基于上述前提，我们简单总结下秒杀下单部分的业务逻辑。
- 秒杀订单服务获取中间层的下单请求，进行真实的下单前校验，这里主要进行库存的真实校验
- 扣减库存（或称锁库存）成功后，发起真实的下单操作。扣减库存（锁库存）与下单操作一般在一个事务域中
- 下单成功后，平台往往会发起消息推送，告知用户下单成功，并引导用户进行支付操作
- 用户一段时间（如：30mins）没有支付，则订单作废，库存恢复，给其他排队中的用户提供购买机会
- 如果用户支付成功，则订单状态更新，订单流转到其他子系统，如：物流系统对该支付成功的处理中订单进行发货等后续处理
到此，基本上就是秒杀业务的核心主流程。
进一步抽象 秒杀请求->中间层->真实下单 这个场景，是不是很像我们经常用到的一种异步业务处理模式？
相信有心的你已经看出来了，没错，这就是 “生产者-消费者” 模式。
“生产者-消费者”模式 在进程内，常常通过 阻塞队列 或者 “等待-通知” 等机制实现，在服务之间则往往通过消息队列实现，这也是本次实战所采用的技术实现手段。本文将通过 RocketMQ 消息队列，对秒杀下单进行解耦，实现削峰填谷、提高系统吞吐量的目的。
接下来将具体讲解怎么使用 RocketMQ 实现上述场景。
3. 实战
3.1 结构
- 用户访问秒杀网关seckill-gateway-service，对感兴趣的商品发起秒杀操作。特别的，对于商品信息，在系统初始化的时候已经加载到 seckill-gateway-service。在进行前置库存校验的时候，依据缓存已经做了一次用户下单流量的过滤
- 网关对秒杀订单进行充分的预校验之后，将秒杀下单消息投递到 RocketMQ 中，同步向用户返回排队中
- 秒杀订单平台 seckill-order-service 订阅秒杀下单消息，对消息进行幂等处理，并对商品库存进行真实校验后，进行真实下单操作
3.2 数据库结构
3.3 NameServer配置
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Component
public class MQNamesrvConfig {
@Value("${rocketmq.nameServer.offline}")
String offlineNamesrv;
@Value("${rocketmq.nameServer.aliyun}")
String aliyunNamesrv;
/**
* 根据环境选择nameServer地址
* @return
*/
public String nameSrvAddr() {
String envType = System.getProperty("envType");
//System.out.println(envType);
if (StringUtils.isBlank(envType)) {
throw new IllegalArgumentException("please insert envType");
}
switch (envType) {
case "offline" : {
return offlineNamesrv;
}
case "aliyun" : {
return aliyunNamesrv;
}
default : {
throw new IllegalArgumentException("please insert right envType, offline/aliyun");
}
}
}
}
复制代码3.4 消息协议
这里通过实现 BaseMsg 的模板方法 encode、decode（分别表示对消息进行编码、解码），通过对this对象进行属性设置，实现了消息协议的自编解码。
/**
* @desc 基础协议类
*/
public abstract class BaseMsg {
public Logger LOGGER = LoggerFactory.getLogger(this.getClass());
/**版本号，默认1.0*/
private String version = "1.0";
/**主题名*/
private String topicName;
public abstract String encode();
public abstract void decode(String msg);
public String getVersion() {
return version;
}
public void setVersion(String version) {
this.version = version;
}
public String getTopicName() {
return topicName;
}
public void setTopicName(String topicName) {
this.topicName = topicName;
}
@Override
public String toString() {
return "BaseMsg{" +
"version='" + version + '\'' +
", topicName='" + topicName + '\'' +
'}';
}
}
复制代码/**
* @className OrderNofityProtocol
* @desc 订单结果通知协议
*/
public class ChargeOrderMsgProtocol extends BaseMsg implements Serializable {
private static final long serialVersionUID = 73717163386598209L;
/**订单号*/
private String orderId;
/**用户下单手机号*/
private String userPhoneNo;
/**商品id*/
private String prodId;
/**用户交易金额*/
private String chargeMoney;
private Map<String, String> header;
private Map<String, String> body;
@Override
public String encode() {
// 组装消息协议头
ImmutableMap.Builder headerBuilder = new ImmutableMap.Builder<String, String>()
.put("version", this.getVersion())
.put("topicName", MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getTopic());
header = headerBuilder.build();
body = new ImmutableMap.Builder<String, String>()
.put("orderId", this.getOrderId())
.put("userPhoneNo", this.getUserPhoneNo())
.put("prodId", this.getProdId())
.put("chargeMoney", this.getChargeMoney())
.build();
ImmutableMap<String, Object> map = new ImmutableMap.Builder<String, Object>()
.put("header", header)
.put("body", body)
.build();
// 返回序列化消息Json串
String ret_string = null;
ObjectMapper objectMapper = new ObjectMapper();
try {
ret_string = objectMapper.writeValueAsString(map);
} catch (JsonProcessingException e) {
throw new RuntimeException("ChargeOrderMsgProtocol消息序列化json异常", e);
}
return ret_string;
}
@Override
public void decode(String msg) {
Preconditions.checkNotNull(msg);
ObjectMapper mapper = new ObjectMapper();
try {
JsonNode root = mapper.readTree(msg);
// header
this.setVersion(root.get("header").get("version").asText());
this.setTopicName(root.get("header").get("topicName").asText());
// body
this.setOrderId(root.get("body").get("orderId").asText());
this.setUserPhoneNo(root.get("body").get("userPhoneNo").asText());
this.setChargeMoney(root.get("body").get("chargeMoney").asText());
this.setProdId(root.get("body").get("prodId").asText());
} catch (IOException e) {
throw new RuntimeException("ChargeOrderMsgProtocol消息反序列化异常", e);
}
}
public String getOrderId() {
return orderId;
}
public ChargeOrderMsgProtocol setOrderId(String orderId) {
this.orderId = orderId;
return this;
}
public String getUserPhoneNo() {
return userPhoneNo;
}
public ChargeOrderMsgProtocol setUserPhoneNo(String userPhoneNo) {
this.userPhoneNo = userPhoneNo;
return this;
}
public String getProdId() {
return prodId;
}
public ChargeOrderMsgProtocol setProdId(String prodId) {
this.prodId = prodId;
return this;
}
public String getChargeMoney() {
return chargeMoney;
}
public ChargeOrderMsgProtocol setChargeMoney(String chargeMoney) {
this.chargeMoney = chargeMoney;
return this;
}
@Override
public String toString() {
return "ChargeOrderMsgProtocol{" +
"orderId='" + orderId + '\'' +
", userPhoneNo='" + userPhoneNo + '\'' +
", prodId='" + prodId + '\'' +
", chargeMoney='" + chargeMoney + '\'' +
", header=" + header +
", body=" + body +
"} " + super.toString();
}
}
复制代码3.5 秒杀订单生产者初始化
通过 @PostConstruct 方式加载（即 init() 方式）
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.gateway.common.config.MQNamesrvConfig;
import org.apache.rocketmq.gateway.common.util.LogExceptionWapper;
import org.apache.rocketmq.message.constant.MessageProtocolConst;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
/**
* @className SecKillChargeOrderProducer
* @desc 秒杀订单生产者初始化
*/
@Component
public class SecKillChargeOrderProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(SecKillChargeOrderProducer.class);
@Autowired
MQNamesrvConfig namesrvConfig;
@Value("${rocketmq.acl.accesskey}")
String aclAccessKey;
@Value("${rocketmq.acl.accessSecret}")
String aclAccessSecret;
private DefaultMQProducer defaultMQProducer;
@PostConstruct
public void init() {
defaultMQProducer =
new DefaultMQProducer
(MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getProducerGroup(),
new AclClientRPCHook(new SessionCredentials(aclAccessKey, aclAccessSecret)));
defaultMQProducer.setNamesrvAddr(namesrvConfig.nameSrvAddr());
// 发送失败重试次数
defaultMQProducer.setRetryTimesWhenSendFailed(3);
try {
defaultMQProducer.start();
} catch (MQClientException e) {
LOGGER.error("[秒杀订单生产者]--SecKillChargeOrderProducer加载异常!e={}", LogExceptionWapper.getStackTrace(e));
throw new RuntimeException("[秒杀订单生产者]--SecKillChargeOrderProducer加载异常!", e);
}
LOGGER.info("[秒杀订单生产者]--SecKillChargeOrderProducer加载完成!");
}
public DefaultMQProducer getProducer() {
return defaultMQProducer;
}
}
复制代码3.6 秒杀订单入队（生产者）
/**
* 平台下单接口
* @param chargeOrderRequest
* @return
*/
@RequestMapping(value = "charge.do", method = {RequestMethod.POST})
public @ResponseBody Result chargeOrder(@ModelAttribute ChargeOrderRequest chargeOrderRequest) {
ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
String sessionId = attributes.getSessionId();
// 下单前置参数校验
if (!secKillChargeService.checkParamsBeforeSecKillCharge(chargeOrderRequest, sessionId)) {
return Result.error(CodeMsg.PARAM_INVALID);
}
// 前置商品校验
String prodId = chargeOrderRequest.getProdId();
if (!secKillChargeService.checkProdConfigBeforeKillCharge(prodId, sessionId)) {
return Result.error(CodeMsg.PRODUCT_NOT_EXIST);
}
// 前置预减库存
if (!secKillProductConfig.preReduceProdStock(prodId)) {
return Result.error(CodeMsg.PRODUCT_STOCK_NOT_ENOUGH);
}
// 秒杀订单入队
return secKillChargeService.secKillOrderEnqueue(chargeOrderRequest, sessionId);
}
复制代码生产者：secKillChargeService::secKillOrderEnqueue
/**
* 秒杀订单入队
* @param chargeOrderRequest
* @param sessionId
* @return
*/
@Override
public Result secKillOrderEnqueue(ChargeOrderRequest chargeOrderRequest, String sessionId) {
// 订单号生成,组装秒杀订单消息协议
String orderId = UUID.randomUUID().toString();
String phoneNo = chargeOrderRequest.getUserPhoneNum();
//消息封装
ChargeOrderMsgProtocol msgProtocol = new ChargeOrderMsgProtocol();
msgProtocol.setUserPhoneNo(phoneNo)
.setProdId(chargeOrderRequest.getProdId())
.setChargeMoney(chargeOrderRequest.getChargePrice())
.setOrderId(orderId);
String msgBody = msgProtocol.encode();
LOGGER.info("秒杀订单入队,消息协议={}", msgBody);
DefaultMQProducer mqProducer = secKillChargeOrderProducer.getProducer();
// 组装RocketMQ消息体
Message message = new Message(MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getTopic(), msgBody.getBytes());
try {
// 消息发送
SendResult sendResult = mqProducer.send(message);
//判断SendStatus
if (sendResult == null) {
LOGGER.error("sessionId={},秒杀订单消息投递失败,下单失败.msgBody={},sendResult=null", sessionId, msgBody);
return Result.error(CodeMsg.BIZ_ERROR);
}
if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
LOGGER.error("sessionId={},秒杀订单消息投递失败,下单失败.msgBody={},sendResult=null", sessionId, msgBody);
return Result.error(CodeMsg.BIZ_ERROR);
}
ChargeOrderResponse chargeOrderResponse = new ChargeOrderResponse();
BeanUtils.copyProperties(msgProtocol, chargeOrderResponse);
LOGGER.info("sessionId={},秒杀订单消息投递成功,订单入队.出参chargeOrderResponse={},sendResult={}", sessionId, chargeOrderResponse.toString(), JSON.toJSONString(sendResult));
return Result.success(CodeMsg.ORDER_INLINE, chargeOrderResponse);
} catch (Exception e) {
int sendRetryTimes = mqProducer.getRetryTimesWhenSendFailed();
LOGGER.error("sessionId={},sendRetryTimes={},秒杀订单消息投递异常,下单失败.msgBody={},e={}", sessionId, sendRetryTimes, msgBody, LogExceptionWapper.getStackTrace(e));
}
return Result.error(CodeMsg.BIZ_ERROR);
}
复制代码3.7 秒杀消费
3.7.1 定义消费者客户端
秒杀下单消费者
@Component
public class SecKillChargeOrderConsumer {
private static final Logger LOGGER = LoggerFactory.getLogger(SecKillChargeOrderConsumer.class);
@Autowired
MQNamesrvConfig namesrvConfig;
@Value("${rocketmq.acl.accesskey}")
String aclAccessKey;
@Value("${rocketmq.acl.accessSecret}")
String aclAccessSecret;
private DefaultMQPushConsumer defaultMQPushConsumer;
@Resource(name = "secKillChargeOrderListenerImpl")
private MessageListenerConcurrently messageListener;
@PostConstruct
public void init() {
defaultMQPushConsumer =
new DefaultMQPushConsumer(
MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getConsumerGroup(),
new AclClientRPCHook(new SessionCredentials(aclAccessKey, aclAccessSecret)),
// 平均分配队列算法，hash
new AllocateMessageQueueAveragely());
defaultMQPushConsumer.setNamesrvAddr(namesrvConfig.nameSrvAddr());
// 从头开始消费
defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
// 消费模式:集群模式
// 集群：同一条消息 只会被一个消费者节点消费到
// 广播：同一条消息 每个消费者都会消费到
defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
// 注册监听器
defaultMQPushConsumer.registerMessageListener(messageListener);
// 设置每次拉取的消息量，默认为1
defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
// 订阅所有消息
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加载完成!");
}
}
复制代码3.7.2 实现秒杀收单核心逻辑
实现秒杀收单核心的逻辑，也就是实现我们自己的MessageListenerConcurrently。
@Component
public class SecKillChargeOrderListenerImpl implements MessageListenerConcurrently {
private static final Logger LOGGER = LoggerFactory.getLogger(SecKillChargeOrderListenerImpl.class);
@Resource(name = "secKillOrderService")
SecKillOrderService secKillOrderService;
@Autowired
SecKillProductService secKillProductService;
/**
* 秒杀核心消费逻辑
* @param msgs
* @param context
* @return
*/
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
try {
for (MessageExt msg : msgs) {
// 消息解码
String message = new String(msg.getBody());
int reconsumeTimes = msg.getReconsumeTimes();
String msgId = msg.getMsgId();
String logSuffix = ",msgId=" + msgId + ",reconsumeTimes=" + reconsumeTimes;
LOGGER.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-接收到消息,message={},{}", message, logSuffix);
// 反序列化协议实体
ChargeOrderMsgProtocol chargeOrderMsgProtocol = new ChargeOrderMsgProtocol();
chargeOrderMsgProtocol.decode(message);
LOGGER.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-反序列化为秒杀入库订单实体chargeOrderMsgProtocol={},{}", chargeOrderMsgProtocol.toString(), logSuffix);
// 消费幂等:查询orderId对应订单是否已存在
String orderId = chargeOrderMsgProtocol.getOrderId();
OrderInfoDobj orderInfoDobj = secKillOrderService.queryOrderInfoById(orderId);
if (orderInfoDobj != null) {
LOGGER.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-当前订单已入库,不需要重复消费!,orderId={},{}", orderId, logSuffix);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
// 业务幂等:同一个prodId+同一个userPhoneNo只有一个秒杀订单
OrderInfoDO orderInfoDO = new OrderInfoDO();
orderInfoDO.setProdId(chargeOrderMsgProtocol.getProdId())
.setUserPhoneNo(chargeOrderMsgProtocol.getUserPhoneNo());
Result result = secKillOrderService.queryOrder(orderInfoDO);
if (result != null && result.getCode().equals(CodeMsg.SUCCESS.getCode())) {
LOGGER.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-当前用户={},秒杀的产品={}订单已存在,不得重复秒杀,orderId={}",
orderInfoDO.getUserPhoneNo(), orderInfoDO.getProdId(), orderId);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
// 秒杀订单入库
OrderInfoDO orderInfoDODB = new OrderInfoDO();
BeanUtils.copyProperties(chargeOrderMsgProtocol, orderInfoDODB);
// 库存校验
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;
}
// 正式下单
if (secKillOrderService.chargeSecKillOrder(orderInfoDODB)) {
LOGGER.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-秒杀订单入库成功,消息消费成功!,入库实体orderInfoDO={},{}", orderInfoDO.toString(), logSuffix);
// 模拟订单处理，直接修改订单状态为处理中
secKillOrderService.updateOrderStatusDealing(orderInfoDODB);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
} catch (Exception e) {
LOGGER.info("[秒杀订单消费者]消费异常,e={}", LogExceptionWapper.getStackTrace(e));
}
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
复制代码3.7.3 秒杀实际入库
实际下单操作与实际库存扣减处于同一个本地事务中
/**
* 秒杀订单入库
* @param orderInfoDO
* @return
*/
@Transactional(rollbackFor = Exception.class)
@Override
public boolean chargeSecKillOrder(OrderInfoDO orderInfoDO) {
int insertCount = 0;
String orderId = orderInfoDO.getOrderId();
String prodId = orderInfoDO.getProdId();
// 减库存
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;
}
复制代码4. 小结&参考资料
小结
再看看一遍流程图，不懂的地方，看看源码。
对于下单后的支付、物流等操作都可以通过使用 RocketMQ 进行异步化处理。
本文全部代码来源于大佬源码
仅为学习研究 RocketMQ 实战。
