高并发秒杀场景

前言

秒杀业务概述

• 秒杀本质上属于短时突发性高并发访问问题，业务特点如下：

  1. 定时触发，流量在瞬间突增
  2. 秒杀请求中常常只有部分能够成功
  3. 秒杀商品数量往往有限，不能超卖，但能接受少卖
  4. 不要求立即返回真实下单结果

• 通过对秒杀核心业务流程进行异步化，我们能够将主流程分为收单、下单两个阶段。

秒杀流程–收单

1. 用户访问秒杀入口，将秒杀请求提交给秒杀平台收单网关，平台对秒杀请求进行前置校验
2. 校验通过后，将下单请求通过缓存/队列/线程池等中间层进行提交，在投递完成同时的同时就给用户返回“排队中”
3. 对于前置校验失败的下单请求同步返回秒杀下单失败,到此，对用户侧的交互就告一段落。
4. 收单过程中，将秒杀订单放入 RocketMQ 中间层中。

秒杀流程–下单

• 下单流程中，平台的压力通过中间层的缓冲其实已经小了很多，

  • 一方面是因为在用户下单的同步校验过程中就过滤掉了部分非法请求；
  • 另一方面，我们通过在中间层做一些限流、过滤等逻辑对下单请求做限速、压单等操作，将下单请求在内部慢慢消化，尽可能减少流量对平台持久层的冲击。

• 这里其实就体现了中间层 “削峰填谷” 的特点。

• 下单业务逻辑。

  1. 秒杀订单服务获取中间层的下单请求，进行真实的下单前校验，这里主要进行库存的真实校验
  2. 扣减库存（或称锁库存）成功后，发起真实的下单操作。扣减库存（锁库存）与下单操作一般在一个事务域中
  3. 下单成功后，平台往往会发起消息推送，告知用户下单成功，并引导用户进行支付操作
  4. 用户一段时间（如：30mins）没有支付，则订单作废，库存恢复，给其他排队中的用户提供购买机会
  5. 如果用户支付成功，则订单状态更新，订单流转到其他子系统，如：物流系统对该支付成功的处理中订单进行发货等后续处理
  6. 到此，基本上就是秒杀业务的核心主流程。

• 进一步抽象 秒杀请求->中间层->真实下单 这个场景，是不是很像我们经常用到的一种异步业务处理模式？

实战

数据库结构

• H2数据库sql脚本如下

  DROP TABLE IF EXISTS t_seckill_product;
  DROP TABLE IF EXISTS t_seckill_order;
  
  
  CREATE TABLE t_seckill_product (
    id int primary key auto_increment,
    gmt_create timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_update timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    prod_id varchar NOT NULL DEFAULT '-1' COMMENT '商品id',
    prod_name varchar NOT NULL DEFAULT '-1' COMMENT '商品名称',
    prod_status int NOT NULL DEFAULT '0' COMMENT '商品状态,0-上架，1-下架',
    prod_stock int NOT NULL DEFAULT '0' COMMENT '商品库存',
    prod_price DECIMAL(10, 3) NOT NULL DEFAULT  '0.000' COMMENT '商品售价',
    version int NOT NULL DEFAULT '0' COMMENT '更新版本号'
  );
  
  CREATE TABLE `t_seckill_order` (
    id int primary key auto_increment,
    gmt_create timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_update timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    record_status tinyint NOT NULL DEFAULT '0' COMMENT '记录状态 0 正常 1 已删除',
    order_id varchar UNIQUE NOT NULL DEFAULT '-1' COMMENT '代理商订单号',
    order_status tinyint NOT NULL DEFAULT '1' COMMENT '订单状态，1 初始化 2 处理中 3 失败 0 成功',
    user_phoneno varchar NOT NULL DEFAULT '-1' COMMENT '用户手机号',
    prod_id varchar NOT NULL DEFAULT '-1' COMMENT '商品id',
    prod_name varchar NOT NULL DEFAULT '-1' COMMENT '商品名称',
    charge_money decimal(10,3) NOT NULL DEFAULT '0.000' COMMENT '交易金额',
    charge_time datetime DEFAULT NULL COMMENT '订单下单时间',
    finish_time datetime DEFAULT NULL COMMENT '订单结束时间'
  );

• mysql脚本

  DROP TABLE IF EXISTS t_seckill_product;
  DROP TABLE IF EXISTS t_seckill_order;
  
  
  CREATE TABLE t_seckill_product (
    id int primary key auto_increment,
    gmt_create timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_update timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    prod_id varchar(32) NOT NULL DEFAULT '-1' COMMENT '商品id',
    prod_name varchar(50) NOT NULL DEFAULT '-1' COMMENT '商品名称',
    prod_status int NOT NULL DEFAULT '0' COMMENT '商品状态,0-上架，1-下架',
    prod_stock int NOT NULL DEFAULT '0' COMMENT '商品库存',
    prod_price DECIMAL(10, 3) NOT NULL DEFAULT  '0.000' COMMENT '商品售价',
    version int NOT NULL DEFAULT '0' COMMENT '更新版本号'
  );
  
  CREATE TABLE `t_seckill_order` (
    id int primary key auto_increment,
    gmt_create timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,
    gmt_update timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    record_status tinyint NOT NULL DEFAULT '0' COMMENT '记录状态 0 正常 1 已删除',
    order_id varchar(32) NOT NULL DEFAULT '-1' COMMENT '代理商订单号',
    order_status tinyint NOT NULL DEFAULT '1' COMMENT '订单状态，1 初始化 2 处理中 3 失败 0 成功',
    user_phoneno varchar(11) NOT NULL DEFAULT '-1' COMMENT '用户手机号',
    prod_id varchar(32) NOT NULL DEFAULT '-1' COMMENT '商品id',
    prod_name varchar(30) NOT NULL DEFAULT '-1' COMMENT '商品名称',
    charge_money decimal(10,3) NOT NULL DEFAULT '0.000' COMMENT '交易金额',
    charge_time datetime DEFAULT NULL COMMENT '订单下单时间',
    finish_time datetime DEFAULT NULL COMMENT '订单结束时间'
  );

• 初始化数据如下

  INSERT INTO t_seckill_product VALUES ('1', '2019-06-14 13:57:36', '2019-06-14 13:57:54', 'pid_0001', 'iphoneX2019新款', '0', '100', '5999.000', '0');
  INSERT INTO t_seckill_product VALUES ('2', '2019-06-14 13:57:36', '2019-06-14 13:57:54', 'pid_0002', '小米9SE', '0', '300', '1200.000', '0');
  INSERT INTO t_seckill_product VALUES ('3', '2019-06-14 13:57:36', '2019-06-14 13:57:54', 'pid_0003', '华为MATE20', '0', '400', '2000.000', '0');

依赖管理

• 使用maven进行管理

  <dependencies>
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-web</artifactId>
      </dependency>
      <!--mybatis-->
      <dependency>
          <groupId>org.mybatis.spring.boot</groupId>
          <artifactId>mybatis-spring-boot-starter</artifactId>
      </dependency>
      <!-- rocketmq -->
      <dependency>
          <groupId>org.apache.rocketmq</groupId>
          <artifactId>rocketmq-spring-boot-starter</artifactId>
      </dependency>
      <dependency>
          <groupId>com.h2database</groupId>
          <artifactId>h2</artifactId>
          <scope>runtime</scope>
      </dependency>
      <dependency>
          <groupId>com.alibaba</groupId>
          <artifactId>fastjson</artifactId>
      </dependency>
      <!-- redis -->
      <dependency>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-starter-data-redis</artifactId>
      </dependency>
      <!-- redis连接池 -->
      <dependency>
          <groupId>org.apache.commons</groupId>
          <artifactId>commons-pool2</artifactId>
      </dependency>
  </dependencies>

编写配置类

application.yml

• yaml文件如下

  server:
    port: 8070
  
  logging:
    file: #配置日志输出的文件
      path: logs/seckill-gateway-service.log
    pattern:
      console: "%-12(%d{yyyy-MM-dd HH:mm:ss.SSS}) |-%-5level [%thread] %c [%L] -| %msg%n"
      file: "%-12(%d{yyyy-MM-dd HH:mm:ss.SSS}) |-%-5level [%thread] %c [%L] -| %msg%n"
    level:
      root: INFO
      com.snowalker: DEBUG
  
  spring:
    application:
      name: seckill-gateway-service
    cache:
      redis:
        # Redis缓存数据统一设置有效期为1分钟
        time-to-live: 60000
    redis:
      host: 192.168.56.101
      port: 6379
      password: 654321
      database: 0 # 数据库索引，默认0
      timeout: 5000 # 连接超时，单位ms
      ssl: false
      # 哨兵模式
      #    sentinel:
      #      master: mymaster
      #      nodes: 192.168.56.101:26379,192.168.56.101:26380 # 哨兵的IP:Port列表
      # 集群模式
      #    cluster:
      #      nodes: 192.168.40.201:7100,192.168.40.201:7200,192.168.40.201:7300,192.168.40.201:7400,192.168.40.201:7500,192.168.40.201:7600
      #      max-redirects: 3  # 重定向的最大次数
      lettuce: # jedis或lettuce, 连接池配置，springboot2.0中使用jedis或者lettuce配置连接池，默认为lettuce连接池
        pool:
          max-active: 8 # 连接池最大连接数（使用负值表示没有限制）
          max-wait: -1 # 连接池分配连接最大阻塞等待时间（阻塞时间到，抛出异常。使用负值表示无限期阻塞）
          max-idle: 8 # 连接池中的最大空闲连接数
          min-idle: 0 # 连接池中的最小空闲连接数
    datasource:
      schema:
        - classpath:db/schema.sql
      data:
        - classpath:db/data.sql
      url: jdbc:h2:mem:testdbsa
      driver-class-name: org.h2.Driver
      username: sa
      password:
    # H2配置
    h2:
      console:
        enabled: true
        path: /h2
        settings:
          web-allow-others: true
          trace: true
  mybatis:
    # 映射实体地址
    type-aliases-package: tk.fulsun.gateway.common.dao.dataobject
    # xml配置文件地址
    mapper-locations: classpath:mapper/*.xml
    # mybatis全局配置,与configuration 不能同时存在
    # config-location: classpath:mybatis/mybatis-config.xml
    configuration:
      # 开启驼峰命名
      map-underscore-to-camel-case: true
      #当传入null的时候对应的jdbctype
      jdbc-type-for-null: null
      #用map接受查询结果时，会自动将查询结果为null的字段忽略
      #查询到0条记录时 会接收到一个所有key值都为null的map
      #只查询一个字段，而用map接收 会接收到一个为null的map
      call-setters-on-nulls: true
  
  rocketmq:
    nameserver: 192.168.56.101:9876

配置rocketmq

• 从配置文件中获取nameservice 地址

  import org.springframework.beans.factory.annotation.Value;
  import org.springframework.stereotype.Component;
  
  /**
   * @author fulsun
   * @description: RocketMQ NameServer配置
   * @date 6/15/2021 4:14 PM
   */
  @Component
  public class MQNamesrvConfig {
    @Value("${rocketmq.nameserver}")
    String namesrv;
  }

配置redis

• 设置redistemplate的序列化方式使用FastJson

  @Configuration
  public class RedisConfig {
  
    /** 自定义RedisTemplate - 配置序列化配置，默认是jdk */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
      // 为了开发的方便，一般直接使用 <String, Object>
      RedisTemplate<String, Object> template = new RedisTemplate<>();
      template.setConnectionFactory(factory); // redis连接的线程安全工厂，源码也是这样配置的
  
      // Json序列化配置
      GenericFastJsonRedisSerializer fastJsonRedisSerializer = new GenericFastJsonRedisSerializer();
      // String 的序列化
      StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
      // key采用String的序列化方式
      template.setKeySerializer(stringRedisSerializer);
      // value序列化方式采用jackson
      template.setValueSerializer(fastJsonRedisSerializer);
      // hash的key也采用String的序列化方式
      template.setHashKeySerializer(stringRedisSerializer);
      // hash的value序列化方式采用jackson
      template.setHashValueSerializer(fastJsonRedisSerializer);
      // 非spring容器必须执行
      template.afterPropertiesSet();
      // 设置其他默认的序列化方式为fastjson
      template.setDefaultSerializer(fastJsonRedisSerializer);
      return template;
    }
  }

配置RestTemplate

• 配置超时时间

  @Configuration
  public class RestTemplateConfig {
    @Bean
    public RestTemplate restTemplate(ClientHttpRequestFactory factory) {
      return new RestTemplate(factory);
    }
  
    @Bean
    public RestTemplate restTemplateNew(ClientHttpRequestFactory factory) {
      return new RestTemplate(factory);
    }
  
    @Bean
    public ClientHttpRequestFactory simpleClientHttpRequestFactory() {
      SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
      /** 读超时单位为ms */
      factory.setReadTimeout(10000);
      /** 连接超时单位为ms */
      factory.setConnectTimeout(10000);
      return factory;
    }
  }

MBG逆向工程

• MBG 插件生成对应的mapper文件和实体类

  <build>
      <plugins>
          <plugin>
              <!-- MBG 插件 -->
              <groupId>org.mybatis.generator</groupId>
              <artifactId>mybatis-generator-maven-plugin</artifactId>
              <!--<overwrite>true</overwrite> 覆盖生效从1.3.7开始 -->
              <version>1.4.0</version>
              <configuration>
                  <configurationFile>
                      ${basedir}/src/main/resources/db/generatorConfig.xml
                  </configurationFile>
                  <overwrite>true</overwrite>
                  <verbose>true</verbose>
              </configuration>
              <dependencies>
                  <dependency>
                      <groupId>mysql</groupId>
                      <artifactId>mysql-connector-java</artifactId>
                      <version>8.0.25</version>
                  </dependency>
              </dependencies>
          </plugin>
      </plugins>
  </build>

• MBG脚本

  <!DOCTYPE generatorConfiguration PUBLIC
    "-//mybatis.org//DTD MyBatis Generator Configuration 1.0//EN"
    "http://mybatis.org/dtd/mybatis-generator-config_1_0.dtd">
  <generatorConfiguration>
    <!--
        mybatis逆向工程配置说明详见官网：http://mybatis.org/generator/running/runningWithMaven.html
        mybatis 标签属性说明：http://mybatis.org/generator/configreference/xmlconfig.html#
    -->
    <!-- targetRuntime: MyBatis3DynamicSql/MyBatis3/MyBatis3Simple	-->
    <context id="mysql" defaultModelType="flat" targetRuntime="MyBatis3">
      <!-- 生成的Java文件的编码 -->
      <property name="javaFileEncoding" value="UTF-8"/>
      <!-- 格式化java代码 -->
      <property name="javaFormatter" value="org.mybatis.generator.api.dom.DefaultJavaFormatter"/>
      <!-- 格式化XML代码 -->
      <property name="xmlFormatter" value="org.mybatis.generator.api.dom.DefaultXmlFormatter"/>
      <!--添加分隔符-->
      <property name="beginningDelimiter" value="'"></property>
      <property name="endingDelimiter" value="'"></property>
  
  
      <!--默认生成getter/setter方法，使用插件忽略生成getter/setter方法-->
      <!--<plugin type="com.mybatis.plugin.IngoreSetterAndGetterPlugin" />-->
      <!--用于在实体类中实现java.io.Serializable接口-->
      <plugin type="org.mybatis.generator.plugins.SerializablePlugin"></plugin>
      <!--用于重写equals 和 hashCode 方法-->
      <plugin type="org.mybatis.generator.plugins.EqualsHashCodePlugin">
        <property name="useEqualsHashCodeFromRoot" value="true"/>
      </plugin>
      <!--用于生成 toString 方法-->
      <plugin type="org.mybatis.generator.plugins.ToStringPlugin">
        <property name="useToStringFromRoot" value="true"/>
      </plugin>
  
      <!--生成注释信息的配置-->
      <commentGenerator>
        <!--阻止生成注释，默认为false-->
        <property name="suppressAllComments" value="true"></property>
        <!--阻止生成的注释包含时间戳，默认为false-->
        <property name="suppressDate" value="true"></property>
        <!--注释是否添加数据库表的备注信息，默认为false-->
        <property name="addRemarkComments" value="true"></property>
      </commentGenerator>
  
      <!--数据库连接信息-->
      <jdbcConnection driverClass="com.mysql.cj.jdbc.Driver"
        connectionURL="jdbc:mysql://192.168.56.101:3306/springbootstudy?useUnicode=true&amp;characterEncoding=UTF-8&amp;zeroDateTimeBehavior=convertToNull&amp;serverTimezone=Asia/Shanghai&amp;useSSL=false"
        userId="test"
        password="123456">
        <!-- MySQL 不支持 schema 或者 catalog 所以需要添加这个,官方文档有 -->
        <property name="nullCatalogMeansCurrent" value="true"/>
      </jdbcConnection>
  
      <!--javaTypeResolver:节点用于指定和配置 Java 类型解析器。-->
      <javaTypeResolver>
        <!--默认的解析器可能会将数据库类型 decimal 或 numberic 解析为Short、Integer、Long等 Java 类型，-->
        <property name="forceBigDecimals" value="ture"/>
        <!--是否不强制将数据库类型 date, time 和 timestamp 解析为 Date;
        默认为false，如果为true，
        解析规则将变成：date -> LocalDate，time -> LocalTime，timestamp -> LocalDateTime-->
        <property name="useJSR310Types" value="true"/>
      </javaTypeResolver>
  
      <!--javaModelGenerator : 节点用于配置实体类生成器-->
      <!--targetPackage：生成实体类存放的包名 targetProject：指定目标项目路径，可以使用绝对路径或者相对路径-->
      <javaModelGenerator targetPackage="tk.fulsun.gateway.common.dao.dataobject"
        targetProject="F:\clone\springboot-study\project-seckill-gateway-service\src\main\java">
        <!-- constructorBased为true就会使用构造方法入参，为false使用setter方法入参，默认为false -->
        <!--<property name="constructorBased" value="false"></property>-->
        <!--enableSubPackages 是否在targetPackage基础上生成子包。默认为false。当为true时，会将表所在 schema 名作为子包名-->
        <!--<property name="enableSubPackages" value="false" />-->
        <!--immutable：用于配置实体类属性是否可变，如果为true，不管constructorBased设置，都会使用构造方法入参不会生成setter方法。
    如果为false实体类属性可以改变，默认为false-->
        <!--<property name="immutable" value="false"></property>-->
        <!--设置所有实体类的基类-->
        <!--<property name="rootClass" value="类的全限定名"></property>-->
        <!--在setter方法中是否对传入字符串进行 trim 操作-->
        <property name="trimStrings" value="true"/>
      </javaModelGenerator>
  
      <!--sqlMapGenerator : 节点用于配置 XML 生成器-->
      <sqlMapGenerator targetPackage="mapper"
        targetProject="F:\clone\springboot-study\project-seckill-gateway-service\src\main\resources">
        <!--enableSubPackages ： 是否在targetPackage基础上生成子包。默认为false。当为true时，会将表所在 schema 名作为子包名-->
        <!--<property name="enableSubPackages" value="false" />-->
      </sqlMapGenerator>
  
      <!-- javaClientGenerator：最多配置一个，用于生成mapper接口
        MyBatis3：ANNOTATEDMAPPER、MIXEDMAPPER、XMLMAPPER
        MyBtais3Simple：ANNOTATEDMAPPER、XMLMAPPER
    -->
      <javaClientGenerator type="XMLMAPPER" targetPackage="tk.fulsun.gateway.common.dao"
        targetProject="F:\clone\springboot-study\project-seckill-gateway-service\src\main\java">
      </javaClientGenerator>
  
      <!--表的配置-->
      <table schema="db_mybatis" tableName="%"></table>
    </context>
  </generatorConfiguration>

收单操作

编写商品的service类

• 提供秒杀商品的查询，修改功能

  public interface SecKillProductService {
    /**
     * 获取秒杀商品列表
     *
     * @return
     */
    List<TSeckillProduct> queryTSeckillProductList();
  
    /**
     * 根据产品id查询产品信息
     *
     * @param prodId
     * @return
     */
    TSeckillProduct queryProdById(String prodId);
  
    /**
     * 修改产品信息
     *
     * @param updateProdData
     * @return
     */
    void updateProdInfo(TSeckillProduct updateProdData);
  
    /**
     * 预减库存
     *
     * @param prodId
     * @return
     */
    boolean preReduceProdStock(String prodId);
  }

• 实现类：通过生成的mapper操作数据库

  @Service(value = "TSeckillProductService")
  public class SecKillProductServiceImpl implements SecKillProductService {
  
      @Autowired private TSeckillProductMapper seckillProductMapper;
      @Autowired private RedisTemplate redisTemplate;
  
      @Override
      public List<TSeckillProduct> queryTSeckillProductList() {
          return seckillProductMapper.selectByExample(null);
      }
  
      @Override
      public TSeckillProduct queryProdById(String prodId) {
          TSeckillProductExample example = new TSeckillProductExample();
          Criteria criteria = example.createCriteria();
          criteria.andProdIdEqualTo(prodId);
          List<TSeckillProduct> products = seckillProductMapper.selectByExample(example);
          return products != null && products.size() > 0 ? products.get(0) : null;
      }
  
      @Override
      public void updateProdInfo(TSeckillProduct updateProdData) {
          seckillProductMapper.updateByPrimaryKeySelective(updateProdData);
      }
  
      @Override
      public boolean preReduceProdStock(String prodId) {
          Preconditions.checkNotNull(prodId, "请确保prodId非空!");
          synchronized (this) {
              TSeckillProduct product = (TSeckillProduct) redisTemplate.opsForValue().get(prodId);
              int prodStock = product.getProdStock();
              if (prodStock <= 0) {
                  log.info("prodId={},prodStock={},当前秒杀商品库存已不足!", prodId, prodStock);
                  return false;
              }
  
              int afterPreReduce = prodStock - 1;
              if (afterPreReduce < 0) {
                  // 预减库存后小于0,说明库存预减失败,库存已不足
                  log.info("prodId={} 预减库存失败,当前库存已不足!", prodId);
                  return false;
              }
              // 预减库存成功,回写库存
              log.info("prodId={} 预减库存成功,当前扣除后剩余库存={}!", prodId, afterPreReduce);
              product.setProdStock(afterPreReduce);
              redisTemplate.opsForValue().set(prodId, product, 86400, TimeUnit.SECONDS);
              return true;
          }
      }
  }

redis初始化加载

• 查询秒杀商品信息，将信息写入到redis中

  /**
   * @author fulsun
   * @description: 秒杀商品初始化加载，正式生产环境是有一个商品服务的，这里简化处理了
   * @date 6/15/2021 5:15 PM
   */
  @Component
  @Slf4j
  public class SecKillProductConfig implements InitializingBean {
  
      @Autowired private RedisTemplate redisTemplate;
  
      @Autowired private SecKillProductService secKillProductService;
  
      @Override
      public void afterPropertiesSet() throws Exception {
          List<TSeckillProduct> killProductList = secKillProductService.queryTSeckillProductList();
          if (killProductList == null) {
              throw new RuntimeException("请确保数据库中存在秒杀商品!");
          }
          //
          killProductList.stream()
              .forEach(
              (pojo -> {
                  String prodId = pojo.getProdId();
                  redisTemplate.opsForValue().set(prodId, pojo, 86400, TimeUnit.SECONDS);
              }));
          log.info("[SecKillProductConfig]初始化秒杀配置完成,商品信息=[{}]", JSON.toJSONString(killProductList));
      }
  }

提供秒杀服务

• 编写秒杀的前置参数校验，前置商品校验，秒杀订单入队接口，秒杀后的查询接口(参数校验)

  package tk.fulsun.gateway.common.service;
  
  /**
   * @author fulsun
   * @description: 秒杀下单service
   * @date 6/16/2021 10:17 AM
   */
  public interface SecKillChargeService {
    /**
     * 秒杀下单前置参数校验
     *
     * @param chargeOrderRequest
     * @param sessionId
     * @return
     */
    boolean checkParamsBeforeSecKillCharge(ChargeOrderRequest chargeOrderRequest, String sessionId);
  
    /**
     * 秒杀下单前置商品校验
     *
     * @param prodId
     * @param sessionId
     * @return
     */
    boolean checkProdConfigBeforeKillCharge(String prodId, String sessionId);
  
    /**
     * 秒杀订单入队
     *
     * @param chargeOrderRequest
     * @param sessionId
     * @return
     */
    Result secKillOrderEnqueue(ChargeOrderRequest chargeOrderRequest, String sessionId);
  
    /**
     * 秒杀查询前置参数校验
     *
     * @param queryOrderRequest
     * @param sessionId
     * @return
     */
    boolean checkParamsBeforeSecKillQuery(QueryOrderRequest queryOrderRequest, String sessionId);
  
    /**
     * 执行订单查询业务
     *
     * @param queryOrderRequest
     * @param sessionId
     * @return
     */
    Result queryOrder(QueryOrderRequest queryOrderRequest, String sessionId);
  }

• 编写实现类

  package tk.fulsun.gateway.common.service.impl;
  
  import com.alibaba.fastjson.JSON;
  import java.math.BigDecimal;
  import java.util.UUID;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.commons.lang3.StringUtils;
  import org.apache.rocketmq.client.producer.DefaultMQProducer;
  import org.apache.rocketmq.client.producer.SendResult;
  import org.apache.rocketmq.client.producer.SendStatus;
  import org.apache.rocketmq.common.message.Message;
  import org.springframework.beans.BeanUtils;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.data.redis.core.RedisTemplate;
  import org.springframework.stereotype.Service;
  import tk.fulsun.gateway.common.dao.dataobject.TSeckillProduct;
  import tk.fulsun.gateway.common.dto.CodeMsg;
  import tk.fulsun.gateway.common.dto.Result;
  import tk.fulsun.gateway.common.dto.request.ChargeOrderRequest;
  import tk.fulsun.gateway.common.dto.request.QueryOrderRequest;
  import tk.fulsun.gateway.common.dto.response.ChargeOrderResponse;
  import tk.fulsun.gateway.common.dto.response.QueryOrderResponse;
  import tk.fulsun.gateway.common.service.SecKillChargeService;
  import tk.fulsun.gateway.common.utils.LogExceptionWapper;
  import tk.fulsun.gateway.mq.msg.ChargeOrderMsgProtocol;
  import tk.fulsun.gateway.mq.msg.MessageProtocolConst;
  import tk.fulsun.gateway.mq.producer.SecKillChargeOrderProducer;
  
  /**
   * @author fulsun
   * @description: 秒杀下单service实现
   * @date 6/16/2021 10:41 AM
   */
  @Slf4j
  @Service(value = "secKillChargeService")
  public class SecKillChargeServiceImpl implements SecKillChargeService {
    @Autowired private RedisTemplate redisTemplate;
  
    @Autowired private SecKillChargeOrderProducer secKillChargeOrderProducer;
  
    @Autowired private OrderQueryManager orderQueryManager;
  
    /**
     * 秒杀下单前置参数校验
     *
     * @param chargeOrderRequest
     * @param sessionId
     * @return true为校验通过
     */
    @Override
    public boolean checkParamsBeforeSecKillCharge(
        ChargeOrderRequest chargeOrderRequest, String sessionId) {
      // 入参校验
      if (chargeOrderRequest == null) {
        log.info("sessionId={},下单请求参数chargeOrderRequest为空,返回下单失败", sessionId);
        return false;
      }
      log.info(
          "sessionId={},下单开始,下单请求参数chargeOrderRequest=[{}].",
          sessionId,
          JSON.toJSONString(chargeOrderRequest));
      String userPhoneNum = chargeOrderRequest.getUserPhoneNum();
      String chargePrice = chargeOrderRequest.getChargePrice();
      String prodId = chargeOrderRequest.getProdId();
  
      if (StringUtils.isBlank(prodId)
          || StringUtils.isBlank(chargePrice)
          || StringUtils.isBlank(userPhoneNum)) {
        log.info("sessionId={},下单必要参数为空,返回下单失败", sessionId);
        return false;
      }
      // 价格合法性校验 是否>0?
      BigDecimal chargePriceDecimal = new BigDecimal(chargePrice);
      if (chargePriceDecimal.longValue() < 0) {
        log.info("sessionId={},商品交易金额小于0,价格非法,返回下单失败", sessionId);
        return false;
      }
      return true;
    }
  
    /**
     * 秒杀下单前置商品校验
     *
     * @param prodId
     * @param sessionId
     * @return
     */
    @Override
    public boolean checkProdConfigBeforeKillCharge(String prodId, String sessionId) {
      // 商品校验
      TSeckillProduct product = (TSeckillProduct) redisTemplate.opsForValue().get(prodId);
      if (product == null) {
        log.info("sessionId={},prodId={},对应的商品信息不存在,返回下单失败", sessionId, prodId);
        return false;
      }
      return true;
    }
  
    /**
     * 秒杀订单入队
     *
     * @param chargeOrderRequest
     * @param sessionId
     * @return
     */
    @Override
    public Result secKillOrderEnqueue(ChargeOrderRequest chargeOrderRequest, String sessionId) {
  
      // 订单号生成,组装秒杀订单消息协议
      String orderId = UUID.randomUUID().toString().replaceAll("-", "");
      String phoneNo = chargeOrderRequest.getUserPhoneNum();
  
      ChargeOrderMsgProtocol msgProtocol = new ChargeOrderMsgProtocol();
      msgProtocol
          .setUserPhoneNo(phoneNo)
          .setProdId(chargeOrderRequest.getProdId())
          .setChargeMoney(chargeOrderRequest.getChargePrice())
          .setOrderId(orderId);
      String msgBody = msgProtocol.encode();
      log.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) {
          log.error("sessionId={},秒杀订单消息投递失败,下单失败.msgBody={},sendResult=null", sessionId, msgBody);
          return Result.error(CodeMsg.BIZ_ERROR);
        }
        if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
          log.error("sessionId={},秒杀订单消息投递失败,下单失败.msgBody={},sendResult=null", sessionId, msgBody);
          return Result.error(CodeMsg.BIZ_ERROR);
        }
        ChargeOrderResponse chargeOrderResponse = new ChargeOrderResponse();
        BeanUtils.copyProperties(msgProtocol, chargeOrderResponse);
        log.info(
            "sessionId={},秒杀订单消息投递成功,订单入队.出参chargeOrderResponse={},sendResult={}",
            sessionId,
            chargeOrderResponse.toString(),
            JSON.toJSONString(sendResult));
        return Result.success(CodeMsg.ORDER_INLINE, chargeOrderResponse);
      } catch (Exception e) {
        int sendRetryTimes = mqProducer.getRetryTimesWhenSendFailed();
        log.error(
            "sessionId={},sendRetryTimes={},秒杀订单消息投递异常,下单失败.msgBody={},e={}",
            sessionId,
            sendRetryTimes,
            msgBody,
            LogExceptionWapper.getStackTrace(e));
      }
      return Result.error(CodeMsg.BIZ_ERROR);
    }
  
    /**
     * 查单前参数校验
     *
     * @param queryOrderRequest
     * @param sessionId
     * @return
     */
    @Override
    public boolean checkParamsBeforeSecKillQuery(
        QueryOrderRequest queryOrderRequest, String sessionId) {
      // 入参校验
      if (queryOrderRequest == null) {
        log.info("sessionId={},查询请求参数queryOrderRequest为空,返回查询失败", sessionId);
        return false;
      }
      log.info(
          "sessionId={},查询开始,查询请求参数queryOrderRequest=[{}].",
          sessionId,
          JSON.toJSONString(queryOrderRequest));
  
      String userPhoneNum = queryOrderRequest.getUserPhoneNum();
      String prodId = queryOrderRequest.getProdId();
  
      if (StringUtils.isBlank(prodId) || StringUtils.isBlank(userPhoneNum)) {
        log.info("sessionId={},查询必要参数为空,返回查询失败", sessionId);
        return false;
      }
      return true;
    }
  
    /**
     * 执行订单查询业务
     *
     * @param queryOrderRequest
     * @param sessionId
     * @return
     */
    @Override
    public Result<QueryOrderResponse> queryOrder(
        QueryOrderRequest queryOrderRequest, String sessionId) {
      return orderQueryManager.queryOrder(queryOrderRequest, sessionId);
    }
  }

秒杀API

• controller接口

  package tk.fulsun.gateway.controller;
  
  import lombok.extern.slf4j.Slf4j;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.web.bind.annotation.GetMapping;
  import org.springframework.web.bind.annotation.ModelAttribute;
  import org.springframework.web.bind.annotation.PostMapping;
  import org.springframework.web.bind.annotation.RequestBody;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RestController;
  import org.springframework.web.context.request.RequestContextHolder;
  import org.springframework.web.context.request.ServletRequestAttributes;
  import tk.fulsun.gateway.common.dto.CodeMsg;
  import tk.fulsun.gateway.common.dto.Result;
  import tk.fulsun.gateway.common.dto.request.ChargeOrderRequest;
  import tk.fulsun.gateway.common.dto.request.QueryOrderRequest;
  import tk.fulsun.gateway.common.service.SecKillChargeService;
  import tk.fulsun.gateway.common.service.SecKillProductService;
  
  /**
   * @author fulsun
   * @description: TODO
   * @date 6/16/2021 1:41 PM
   */
  @Slf4j
  @RestController
  @RequestMapping("api")
  public class OrderChargeController {
      @Autowired private SecKillChargeService secKillChargeService;
      @Autowired private SecKillProductService secKillProductService;
  
      /**
     * 平台下单接口
     *
     * @param chargeOrderRequest
     * @return
     */
      @PostMapping(value = "charge")
      public Result chargeOrder(@RequestBody 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 (!secKillProductService.preReduceProdStock(prodId)) {
              return Result.error(CodeMsg.PRODUCT_STOCK_NOT_ENOUGH);
          }
          // 秒杀订单入队
          return secKillChargeService.secKillOrderEnqueue(chargeOrderRequest, sessionId);
      }
  
      /**
     * 平台查单接口
     *
     * @param queryOrderRequest
     * @return
     */
      @GetMapping(value = "query")
      public Result queryOrder(@ModelAttribute QueryOrderRequest queryOrderRequest) {
          ServletRequestAttributes attributes =
              (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
          String sessionId = attributes.getSessionId();
          // 查询前置参数校验
          if (!secKillChargeService.checkParamsBeforeSecKillQuery(queryOrderRequest, sessionId)) {
              return Result.error(CodeMsg.PARAM_INVALID);
          }
          // 查询订单
          return secKillChargeService.queryOrder(queryOrderRequest, sessionId);
      }
  }

主启动类

• 启动类上扫描Mapper

  @SpringBootApplication
  @MapperScan("tk.fulsun.gateway.common.dao")
  public class GateWayApplication {
    public static void main(String[] args) {
      SpringApplication.run(GateWayApplication.class, args);
    }
  }

测试

• postman

  

• 控制台输出

  

• mq控制台查询

  

• 查询缓存状态

  

• 上面说明收单的步骤完成了

下单操作

• 创建订单模块project-seckill-order-service

消费者

• 代码如下

  import javax.annotation.PostConstruct;
  import javax.annotation.Resource;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
  import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
  import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
  import org.apache.rocketmq.client.exception.MQClientException;
  import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
  import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Component;
  import tk.fulsun.order.config.MQNamesrvConfig;
  import tk.fulsun.order.mq.msg.MessageProtocolConst;
  import tk.fulsun.order.utils.LogExceptionWapper;
  
  /**
   * @author fulsun
   * @description: 秒杀下单消费者
   * @date 6/16/2021 3:13 PM
   */
  @Slf4j
  @Component
  public class SecKillChargeOrderConsumer {
    @Autowired MQNamesrvConfig namesrvConfig;
  
    @Resource(name = "secKillChargeOrderListenerImpl")
    private MessageListenerConcurrently messageListener;
  
    private DefaultMQPushConsumer defaultMQPushConsumer;
  
    @PostConstruct
    public void init() {
      // pull 拉取消息,push 推送消息。push本质:拉取模式,主动的拉取消息避免消费者的堆积
      defaultMQPushConsumer =
          new DefaultMQPushConsumer(
              MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getConsumerGroup(),
              null,
              // 平均分配队列算法，hash
              new AllocateMessageQueueAveragely());
      defaultMQPushConsumer.setNamesrvAddr(namesrvConfig.getNamesrv());
      // 消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
      // defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
      // 从头开始消费
      defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
      // 消费模式:集群模式
      // 集群 CLUSTERING：同一条消息 只会被一个消费者节点消费到
      // 广播 BROADCASTING：同一条消息 每个消费者都会消费到
      defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
      // 设置每次拉取的消息量，默认为1
      defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
      // 注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
      defaultMQPushConsumer.registerMessageListener(messageListener);
      // 订阅所有消息
      try {
        defaultMQPushConsumer.subscribe(
            MessageProtocolConst.SECKILL_CHARGE_ORDER_TOPIC.getTopic(), "*");
        // 启动消费者
        defaultMQPushConsumer.start();
      } catch (MQClientException e) {
        log.error(
            "[秒杀下单消费者]--SecKillChargeOrderConsumer加载异常!e={}", LogExceptionWapper.getStackTrace(e));
        throw new RuntimeException("[秒杀下单消费者]--SecKillChargeOrderConsumer加载异常!", e);
      }
      log.info("[秒杀下单消费者]--SecKillChargeOrderConsumer加载完成!");
    }
  }

监听回调

• 消费消息的时候进行预校验

  @Slf4j
  @Component
  public class SecKillChargeOrderListenerImpl implements MessageListenerConcurrently {
  
    @Autowired private SecKillOrderService secKillOrderService;
    @Autowired private 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;
          log.info("[秒杀订单消费者]-SecKillChargeOrderConsumer-接收到消息,message={},{}", message, logSuffix);
  
          // 反序列化协议实体
          ChargeOrderMsgProtocol chargeOrderMsgProtocol = new ChargeOrderMsgProtocol();
          chargeOrderMsgProtocol.decode(message);
          log.info(
              "[秒杀订单消费者]-SecKillChargeOrderConsumer-反序列化为秒杀入库订单实体chargeOrderMsgProtocol={},{}",
              chargeOrderMsgProtocol.toString(),
              logSuffix);
  
          // 消费幂等:查询orderId对应订单是否已存在
          String orderId = chargeOrderMsgProtocol.getOrderId();
          TSeckillOrder orderInfoDobj = secKillOrderService.queryOrderInfoById(orderId);
          if (orderInfoDobj != null) {
            log.info(
                "[秒杀订单消费者]-SecKillChargeOrderConsumer-当前订单已入库,不需要重复消费!,orderId={},{}",
                orderId,
                logSuffix);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
  
          // 业务幂等:同一个prodId+同一个userPhoneNo只有一个秒杀订单
          TSeckillOrder orderInfo = new TSeckillOrder();
          orderInfo.setProdId(chargeOrderMsgProtocol.getProdId());
          orderInfo.setUserPhoneno(chargeOrderMsgProtocol.getUserPhoneNo());
          Result result = secKillOrderService.queryOrder(orderInfo);
          if (result != null && result.getCode().equals(CodeMsg.SUCCESS.getCode())) {
            log.info(
                "[秒杀订单消费者]-SecKillChargeOrderConsumer-当前用户={},秒杀的产品={}订单已存在,不得重复秒杀,orderId={}",
                orderInfo.getUserPhoneno(),
                orderInfo.getProdId(),
                orderId);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
  
          // 秒杀订单入库
          TSeckillOrder orderInfoDB = new TSeckillOrder();
          BeanUtils.copyProperties(chargeOrderMsgProtocol, orderInfoDB);
          orderInfoDB.setUserPhoneno(chargeOrderMsgProtocol.getUserPhoneNo());
  
          // 库存校验
          String prodId = chargeOrderMsgProtocol.getProdId();
          TSeckillProduct productDobj = secKillProductService.querySecKillProductByProdId(prodId);
          // 取库存校验
          int currentProdStock = productDobj.getProdStock();
          if (currentProdStock <= 0) {
            log.info(
                "[decreaseProdStock]当前商品已售罄,消息消费成功!prodId={},currStock={}", prodId, currentProdStock);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
          // 正式下单
          if (secKillOrderService.chargeSecKillOrder(orderInfoDB)) {
            log.info(
                "[秒杀订单消费者]-SecKillChargeOrderConsumer-秒杀订单入库成功,消息消费成功!,入库实体orderInfoDO={},{}",
                orderInfoDB.toString(),
                logSuffix);
            // 模拟订单处理，直接修改订单状态为处理中
            secKillOrderService.updateOrderStatusDealing(orderInfoDB);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
          return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
      } catch (Exception e) {
        log.info("[秒杀订单消费者]消费异常,e={}", LogExceptionWapper.getStackTrace(e));
      }
      return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
  }

下单操作

• 将减库存和写入订单作为事务操作

• 减库存(乐观锁:对应的数据表中多添加一个版本标识字段)

  /**
   * 减库存,基于乐观锁
  *
  * @param prodId
  * @return
  */
  @Transactional(rollbackFor = Exception.class)
  @Override
  public boolean decreaseProdStock(String prodId) {
  
      TSeckillProduct productDobj = querySecKillProductByProdId(prodId);
      // 取库存
      int currentProdStock = productDobj.getProdStock();
      // 取版本号
      int beforeVersion = productDobj.getVersion();
      TSeckillProduct productDO = new TSeckillProduct();
      productDO.setProdStock(currentProdStock - 1);
      productDO.setVersion(beforeVersion + 1);
      int updateCount = 0;
      try {
          // 更新操作
          TSeckillProductExample example = new TSeckillProductExample();
          example.createCriteria()
              .andProdStockGreaterThanOrEqualTo(1)
              .andProdIdEqualTo(prodId)
              .andVersionEqualTo(beforeVersion);
          updateCount = secKillProductMapper.updateByExampleSelective(productDO, example);
      } catch (Exception e) {
          log.error(
              "[decreaseProdStock]prodId={},商品减库存[异常],事务回滚,e={}",
              prodId,
              LogExceptionWapper.getStackTrace(e));
          String message = String.format("[decreaseProdStock]prodId=%s,商品减库存[异常],事务回滚", prodId);
          throw new RuntimeException(message);
      }
      if (updateCount != 1) {
          log.error("[decreaseProdStock]prodId={},商品减库存[失败],事务回滚,e={}", prodId);
          String message = String.format("[decreaseProdStock]prodId=%s,商品减库存[失败],事务回滚", prodId);
          throw new RuntimeException(message);
      }
      log.info("[decreaseProdStock]当前减库存[成功],prodId={},剩余库存={}", prodId, currentProdStock - 1);
      return true;
  }

• 写入订单信息

  /**
     * 秒杀订单入库
     *
     * @param orderInfo
     * @return
     */
  @Override
  @Transactional(rollbackFor = Exception.class)
  public boolean chargeSecKillOrder(TSeckillOrder orderInfo) {
      int insertCount = 0;
      String orderId = orderInfo.getOrderId();
      String prodId = orderInfo.getProdId();
  
      // 减库存
      if (!secKillProductService.decreaseProdStock(prodId)) {
          log.info("[insertSecKillOrder]orderId={},prodId={},下单前减库存失败,下单失败!", orderId, prodId);
          // TODO 此处可给用户发送通知，告知秒杀下单失败，原因：商品已售罄
          return false;
      }
      // 设置产品名称
      TSeckillProduct productInfo = secKillProductService.querySecKillProductByProdId(prodId);
      orderInfo.setProdName(productInfo.getProdName());
      try {
          // 记录状态 0 正常
          orderInfo.setRecordStatus((byte) 0);
          //订单状态，1 初始化
          orderInfo.setOrderStatus((byte) 1);
          insertCount = secKillOrderMapper.insertSelective(orderInfo);
      } catch (Exception e) {
          log.error(
              "[insertSecKillOrder]orderId={},秒杀订单入库[异常],事务回滚,e={}",
              orderId,
              LogExceptionWapper.getStackTrace(e));
          String message = String.format("[insertSecKillOrder]orderId=%s,秒杀订单入库[异常],事务回滚", orderId);
          throw new RuntimeException(message);
      }
      if (insertCount != 1) {
          log.error("[insertSecKillOrder]orderId={},秒杀订单入库[失败],事务回滚,e={}", orderId);
          String message = String.format("[insertSecKillOrder]orderId=%s,秒杀订单入库[失败],事务回滚", orderId);
          throw new RuntimeException(message);
      }
      return true;
  }

并发测试

• 使用100个线程对pid_0001商品进行秒杀

  ​

参考

• https://blog.csdn.net/chachapaofan/article/details/104155403
• 大佬源码