【RocketMQ】秒杀设计与实现

🎯 导读：本文档详细探讨了高并发场景下的秒杀系统设计与优化策略，特别是如何在短时间内处理大量请求。文档分析了系统性能指标如QPS（每秒查询率）和TPS（每秒事务数），并通过实例讲解了如何使用JMeter进行性能测试。此外，文档提供了技术选型指南，包括SpringBoot、Redis、RocketMQ等技术的应用，并给出了具体的用户量评估和服务器配置建议。最后，通过分析不同的库存扣减与订单创建实现方式，提出了使用Redis分布式锁等技术提高并发性能的方法。

秒杀

介绍

秒杀：很短的时间内，要处理大量的请求

【高并发介绍】

并发：多个任务在同一时间段内执行，cpu不停切换来执行不同任务

并行：多核CPU上，多个任务在同一时刻执行

要想高并发，硬件很重要，但是成本很高，企业希望在有限的硬件上，最优化软件的性能

性能指标

QPS

• QPS：每秒钟处理请求的数量，业务处理时间越低，QPS越高

Tomcat 的 QPS：SpringBoot的Tomcat默认是最大是200个线程，如果请求处理消耗50ms，理论QPS就是1000*200/50=4000，实际大概率会更低

可以在配置文件中设置tomcat的线程数量

【使用Jmeter测试】

如果异常很大，超过0.5%，数据就没有太大的价值

Tomcat最大连接数改成400

如果并发量非常大，一个Tomcat顶不住，可以做服务集群。

• 一个nginx可以顶住5w的QPS，再负载均衡到多个tomcat服务中

• 并发量达到30w，nginx顶不住了，使用好机器来提供虚拟IP，然后再将请求分发到多个Nginx中
  

• 个人开发，100wQPS就很强了。如果有很大的流量，可以根据用户IP拆分到不同地区的机房。一个域名下面对应很多个服务器IP，按照用户IP区域将其分发大较近的机房IP即可

TPS

每秒钟能够处理的事务或交易的数量。

怎么优化接口性能

• 减少IO（批量查询、批量插入、批量删除）
• 尽早return（例如先去Redis判断的库存够不够，再去执行扣减库存）
• 能异步就异步（减库存放到MQ）
• 锁粒度尽量小
• 事务范围尽可能小
• 前端分流（如拼图滑块、计算，有人快、有人慢，同时可以验证是否为机器人）
• 做限制（一个人针对一个商品只能抢一次优惠券，Redis setnx，抢过就不让进来了）

• seckill-web：接受秒杀请求，然后把业务交给seckill-service执行
• seckill-service：处理秒杀真实业务

技术选型

• Springboot 接收请求并操作 redis 和 MySQL
• Redis 用于缓存+分布式锁
• RocketMQ 用于解耦、削峰、异步
• MySQL 用于存放真实的商品信息
• Mybatis 用于操作数据库的orm框架

用户量评估

总用户量：50w

日活量：1-2w（用户不会天天用，除非经常做活动）

qps：2w+（怎么统计，日志，统计次数）

几台服务器（什么配置）：8C16G 4-6台

• seckill-web：4台
• seckill-service：2台

带宽：100M

技术要点

1. 通过 redis 的 setnx 对用户和商品做去重判断, 防止用户刷接口
2. 每天晚上 8 点通过定时任务把 MySQL 中参与秒杀的库存商品, 同步到 redis 中去, 做库存的预扣减, 提升接口性能
3. 通过 RocketMQ 消息中间件的异步消息, 来将秒杀的业务异步化, 进一步提升性能
4. seckill-service 使用并发消费模式, 并且设置合理的线程数量, 快速处理队列中堆积的消息
5. 使用 redis 的分布式锁+自旋锁, 对商品的库存进行并发控制, 把并发压力转移到程序中和 redis 中去, 减少 db 压力
6. 使用声明式事务注解 Transactional, 并且设置异常回滚类型, 控制数据库的原子性操作
7. 使用 jmeter 压测工具, 对秒杀接口进行压力测试, 在 8C16G 的服务器上, qps2k+, 达到压测预期

架构图

数据库

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;--   ----------------------------
-- Table structure for goods
--   ----------------------------
DROP TABLE IF EXISTS   `goods`;
CREATE TABLE `goods`  (`id` int(11) NOT NULL AUTO_INCREMENT,`goods_name` varchar(255) CHARACTER SET   utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,`price` decimal(10, 2) NULL DEFAULT NULL,`stocks` int(255) NULL DEFAULT NULL,`status` int(255) NULL DEFAULT NULL,`pic` varchar(255) CHARACTER SET utf8mb4   COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,`create_time` datetime(0) NULL DEFAULT   NULL,`update_time` datetime(0) NULL DEFAULT   NULL,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB   AUTO_INCREMENT = 4 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci   ROW_FORMAT = Dynamic;--   ----------------------------
-- Records of goods
--   ----------------------------
INSERT INTO `goods` VALUES   (1, '小米12s', 4999.00,   1000, 2, 'xxxxxx', '2023-02-23 11:35:56', '2023-02-23 16:53:34');
INSERT INTO `goods` VALUES   (2, '华为mate50', 6999.00,   10, 2, 'xxxx', '2023-02-23 11:35:56', '2023-02-23 11:35:56');
INSERT INTO `goods` VALUES   (3, '锤子pro2', 1999.00,   100, 1, NULL, '2023-02-23 11:35:56', '2023-02-23 11:35:56');--   ----------------------------
-- Table structure for   order_records
--   ----------------------------
DROP TABLE IF EXISTS   `order_records`;
CREATE TABLE   `order_records`  (`id` int(11) NOT NULL AUTO_INCREMENT,`user_id` int(11) NULL DEFAULT NULL,`order_sn` varchar(255) CHARACTER SET   utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL,`goods_id` int(11) NULL DEFAULT NULL,`create_time` datetime(0) NULL DEFAULT   NULL,PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB   AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci   ROW_FORMAT = Dynamic;SET FOREIGN_KEY_CHECKS = 1;

创建项目选择依赖seckill-web（接受用户秒杀请求）

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.13</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.powernode</groupId><artifactId>seckill-web</artifactId><version>0.0.1-SNAPSHOT</version><name>seckill-web</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><!-- RocketMQ的依赖 --><dependency><groupId>org.apache.RocketMQ</groupId><artifactId>RocketMQ-spring-boot-starter</artifactId><version>2.2.1</version></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.14</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>

修改配置文件

server:port: 7001tomcat:threads:max: 400
spring:application:name: seckill-webredis:host: 127.0.0.1port: 6379database: 0lettuce:pool:enabled: truemax-active: 100max-idle: 20min-idle: 5
RocketMQ:name-server: 192.168.188.129:9876     # RocketMQ的nameServer地址producer:access-key: dsad secret-key: dsadasfasgroup: powernode-group        # 生产者组别，不配置会报错send-message-timeout: 3000  # 消息发送的超时时间retry-times-when-send-async-failed: 2  # 异步消息发送失败重试次数max-message-size: 4194304       # 消息的最大长度

创建SeckillController

package com.powernode.controller;import com.alibaba.fastjson.JSON;
import org.apache.RocketMQ.client.producer.SendCallback;
import org.apache.RocketMQ.client.producer.SendResult;
import org.apache.RocketMQ.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;import java.util.HashMap;
import java.util.concurrent.atomic.AtomicInteger;@RestController
public class SeckillController {@Autowiredprivate StringRedisTemplate redisTemplate;@Autowiredprivate RocketMQTemplate RocketMQTemplate;/*** 压测时自动是生成用户id*/AtomicInteger ai = new AtomicInteger(0);/*** 1.用户去重，一个用户针对一种商品只能抢购一次* 2.做库存的预扣减  拦截掉大量无效请求* 3.放入mq 异步化处理订单* userId通过登录状态拿取* @return*/@GetMapping("doSeckill")public String doSeckill(Integer goodsId /*, Integer userId*/) {int userId = ai.incrementAndGet();// unique key 唯一标记 去重String uk = userId + "-" + goodsId;// set nx  set if not exist。如果要每天刷新，key加上年月日即可，key再设置过期时间Boolean flag = redisTemplate.opsForValue().setIfAbsent("seckillUk:" + uk, "");if (!flag) {return "您已经参与过该商品的抢购，请参与其他商品抢购!";}// 假设库存已经同步了  key:goods_stock:1  val:10// 直接扣减数量，线程安全。如果先查出来，再减少，线程不安全Long count = redisTemplate.opsForValue().decrement("goods_stock:" + goodsId);// getkey  java  setkey    先查再写 再更新 有并发安全问题if (count < 0) {return "该商品已经被抢完，请下次早点来";}// 放入mqHashMap<String, Integer> map = new HashMap<>(4);map.put("goodsId", goodsId);map.put("userId", userId);RocketMQTemplate.asyncSend("seckillTopic3", JSON.toJSONString(map), new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功" + sendResult.getSendStatus());}@Overridepublic void onException(Throwable throwable) {System.err.println("发送失败" + throwable.getMessage());}});// 不能直接返回抢购成功，因为MQ可能是有问题的return "拼命抢购中，请稍后去订单中心查看";}
}

创建项目选择依赖seckill-service（处理秒杀）

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.6.13</version><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.powernode</groupId><artifactId>seckill-service</artifactId><version>0.0.1-SNAPSHOT</version><name>seckill-service</name><description>Demo project for Spring Boot</description><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.mybatis.spring.boot</groupId><artifactId>mybatis-spring-boot-starter</artifactId><version>2.3.0</version></dependency><dependency><groupId>com.MySQL</groupId><artifactId>MySQL-connector-j</artifactId><scope>runtime</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>1.2.6</version></dependency><!-- RocketMQ的依赖 --><dependency><groupId>org.apache.RocketMQ</groupId><artifactId>RocketMQ-spring-boot-starter</artifactId><version>2.2.1</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-pool2</artifactId></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.14</version></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId><configuration><excludes><exclude><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></exclude></excludes></configuration></plugin></plugins></build></project>

修改yml文件

server:port: 7002
spring:application:name: seckill-servicedatasource:driver-class-name: com.MySQL.cj.jdbc.Driverurl: jdbc:MySQL://127.0.0.1:3306/seckill?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTCusername: rootpassword: 123456redis:host: 127.0.0.1port: 6379database: 0lettuce:pool:enabled: truemax-active: 100max-idle: 20min-idle: 5
mybatis:configuration:log-impl: org.apache.ibatis.logging.stdout.StdOutImplmapper-locations: classpath*:mapper/*.xml
RocketMQ:name-server: 192.168.188.129:9876

逆向生成实体类

修改启动类

@SpringBootApplication
@MapperScan(basePackages = {"com.powernode.mapper"}) // mapper上面有@Mapper注解，这里就不用加扫描了
@EnableScheduling // 开启定时任务
public class seckillServiceApplication {public static void main(String[] args) {SpringApplication.run(seckillServiceApplication.class, args);}
}

修改GoodsMapper

List<Goods> selectSeckillGoods();

修改GoodsMapper.xml

<!--  查询数据库中需要参于秒杀的商品数据 status = 2 -->
<select id="selectSeckillGoods" resultMap="BaseResultMap">select `id`,`stocks` from goods where `status` = 2
</select>

同步MySQL数据到redis

方法1

package com.powernode.config;import com.powernode.domain.Goods;
import com.powernode.mapper.GoodsMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;import javax.annotation.PostConstruct;
import java.util.List;/*** 将MySQL的参与抢购的商品的数据* 同步到redis里面去* 在上游服务需要使用redis来做库存的预扣减*/
@Component
public class DataSyncConfig {@Autowiredprivate GoodsMapper goodsMapper;@Autowiredprivate StringRedisTemplate redisTemplate;// 业务场景是搞一个定时任务 每天10点开启// 为了 测试方便 项目已启动就执行一次/*** spring bean的生命周期* 在当前对象 实例化完以后* 属性注入以后* 执行 PostConstruct 注解的方法*/@PostConstruct// java的注解，不是Spring的注解，项目启动的时候，就执行这个方法@Scheduled(cron = "0 10 0 0 0 ?")public void initData() {List<Goods> goodsList = goodsMapper.selectSeckillGoods();if (CollectionUtils.isEmpty(goodsList)) {return;}goodsList.forEach(goods -> redisTemplate.opsForValue().set("goods_stock:" + goods.getId(), goods.getStocks().toString()));}}

不用上面的方法的话，可以在启动类中写，但是不推荐

Bean生命周期

• 实例化对象 new

• 属性赋值

• 初始化

  • spring

• boot （前：PostConstruct，或下面写法；中；后）

• @Component
  public class DataSync implements InitializingBean, BeanPostProcessor{@Override public void afterropertiesSet() throws Exception {} @Override public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException{}@Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException{}
  }
  

• 使用

• 销毁

方法2

package com.powernode.data;import com.powernode.domain.Goods;
import com.powernode.mapper.GoodsMapper;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.util.List;@Component
public class MySQLToRedis2 implements CommandLineRunner {@Resourceprivate GoodsMapper goodsMapper;@Resourceprivate StringRedisTemplate stringRedisTemplate;@Overridepublic void run(String... args) throws Exception {initData();}private void initData() {//1 查询数据库中需要参于秒杀的商品数据List<Goods> goodsList = goodsMapper.queryseckillGoods();ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();//2 把数据同步到Redisfor (Goods goods : goodsList) {operations.set("goods:" + goods.getGoodsId(), goods.getTotalStocks().toString());}}}

秒杀业务监听器

package com.powernode.listener;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.powernode.service.GoodsService;
import org.apache.RocketMQ.common.message.MessageExt;
import org.apache.RocketMQ.spring.annotation.RocketMQMessageListener;
import org.apache.RocketMQ.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;/*** 默认负载均衡模式* 默认多线程消费*/
@Component
@RocketMQMessageListener(topic = "seckillTopic3", consumerGroup = "seckill-consumer-group")
public class SeckillMsgListener implements RocketMQListener<MessageExt> {@Autowiredprivate GoodsService goodsService;@Autowiredprivate StringRedisTemplate redisTemplate;// 20sint time = 20000;/** 扣减库存* 写订单表*/@Overridepublic void onMessage(MessageExt message) {String s = new String(message.getBody());JSONObject jsonObject = JSON.parseObject(s);Integer goodsId = jsonObject.getInteger("goodsId");Integer userId = jsonObject.getInteger("userId");// 减库存，写订单表，使用同步代码块
//        synchronized (this) {
//            goodsService.realDoSeckill1(goodsId, userId);
//        }// 减库存，写订单表，使用MySQL行锁// goodsService.realDoSeckill1(goodsId, userId);// 减库存，写订单表，使用Redis自旋加锁  int current = 0;// 如果有业务因为自旋时间限制，在有限时间内没有抢得到锁，可以增加限制时间上限，或者把循环改成truewhile (current <= time) {// 一般在做分布式锁的情况下，会给锁一个过期时间，防止出现死锁Boolean flag = redisTemplate.opsForValue().setIfAbsent("goods_lock:" + goodsId, "", 10, TimeUnit.SECONDS);if (flag) {// 加锁成功try {goodsService.realDoSeckill(goodsId, userId);return;} finally {// 解锁redisTemplate.delete("goods_lock:" + goodsId);}} else {// 获取锁失败，自旋加锁current += 200;try {Thread.sleep(200);} catch (InterruptedException e) {e.printStackTrace();}}}}}

修改GoodsService

void realDoSeckill(Integer goodsId, Integer userId);

修改GoodsServiceImpl

【基础方案：有问题】

@Resource
private GoodsMapper goodsMapper;@Autowired
private OrderRecordsMapper orderRecordsMapper;/*** 扣减库存* 写订单表* @param goodsId* @param userId*/
@Override
@Transactional(rollbackFor = RuntimeException.class)
public void realDoSeckill(Integer goodsId, Integer userId) {// 扣减库存  插入订单表Goods goods = goodsMapper.selectByPrimaryKey(goodsId);int finalStock = goods.getStocks() - 1;if (finalStock < 0) {// 只是记录日志 让代码停下来   这里的异常用户无法感知throw new RuntimeException("库存不足：" + goodsId);}goods.setStocks(finalStock);goods.setUpdateTime(new Date());// insert 要么成功 要么报错  update 会出现i<=0的情况// update goods set stocks =  1 where id = 1  没有行锁int i = goodsMapper.updateByPrimaryKey(goods);if (i > 0) {// 写订单表OrderRecords orderRecords = new OrderRecords();orderRecords.setGoodsId(goodsId);orderRecords.setUserId(userId);orderRecords.setCreateTime(new Date());// 时间戳生成订单号orderRecords.setOrderSn(String.valueOf(System.currentTimeMillis()));orderRecordsMapper.insert(orderRecords);}
}

上面的实现不是线程安全的，先查了库存，然后再去修改。并发中，可能一开始库存是够的，后面被其他用户抢走了，库存不够了，但是这里的程序还会继续往下执行

【加锁方案：效率低】

加锁：库存扣减不对，性能差

原因：加事务》加锁》提交事务，MySQL默认事务隔离级别是可重复读。原本有1000件，两个人消费，按理说是998件。但实际上，A进入了方法，修改完库存，释放了锁，但是还没有提交事务，@Transactional是包住整个方法的。B线程进来获得了锁，查询数据库，还是1000件，导致两个线程业务执行完成之后，还剩下999

解决：要先提交事务，才释放锁，这样才是正确的。将代码改成锁包住事务，数据正确性保证了。但是效率还是低

分布式系统要改成分布式锁

【使用MySQL行锁(innodb才有)，并发性能不足】

• update goods set stocks = stocks - 1会触发行锁
• update goods set stocks = 具体值不会触发行锁
• stocks > 1加一个控制

/*** MySQL行锁  innodb  行锁* 分布式锁* todo 答案1** @param goodsId* @param userId*/
@Override
@Transactional(rollbackFor = RuntimeException.class)
public void realDoSeckill1(Integer goodsId, Integer userId) {// update goods set stocks = stocks - 1 ,update_time = now() where id = #{value} and stocks > 1 int i = goodsMapper.updateStocks(goodsId);if (i > 0) {// 写订单表OrderRecords orderRecords = new OrderRecords();orderRecords.setGoodsId(goodsId);orderRecords.setUserId(userId);orderRecords.setCreateTime(new Date());// 时间戳生成订单号orderRecords.setOrderSn(String.valueOf(System.currentTimeMillis()));orderRecordsMapper.insert(orderRecords);}
}

缺点：通过MySQL来控制锁，数据库压力大，如果并发数在1000以下还好，高一点还是建议其他方案

【在监听器中使用Redis自旋加锁】

详情看前面的秒杀业务监听器实现