Sentinel + Redis + Mysql + RabbitMQ 秒杀功能设计及后端代码实现

文章目录

- 前言
- 数据一致性
- 高性能
- - 动静分离
  - 静态资源缓存
  - 流控
  - 缓存数据库
  - 消息队列
  - - RabbitMQ的优点
- 高并发
- - 分布式锁
- 后端代码实现
- - 中间件
  - 表结构
  - 添加依赖
  - 公共常量
  - 实体类
  - Redission配置
  - 定时任务
  - Controller
  - 下单接口
  - 付款接口
  - 接收通道消息
  - 完整代码

前言

在开发秒杀系统功能的时候，需要考虑但不限于以下几点：
1. 确保数据一致性
2. 确保系统高性能
3. 处理高并发场景
实际上，对于不同的秒杀业务场景，需要考虑的问题也会有不同的解决方案。就比如这里我们提到的在确保数据一致性的问题，根据不同的业务需求有不同的处理方案。

数据一致性

秒杀系统的数据一致性，其中一方面体现在库存数量的计算上，我们不仅要确保商品尽可能地卖光，还要确保生成的最终订单数量不能超过预设的库存值，否则就会出现超卖的情况，这也是我们整个秒杀服务最基本的要求。
为了防止超卖，我们可以在每次生成订单前查询当前秒杀商品的剩余库存，库存不足不允许生成订单。
另一方面，生成订单后需要扣减库存，不同的扣减方案也会影响整个秒杀功能的实际表现。正常情况下购买商品分为两个步骤：下单、付款。扣款方案大致分为三种情况。

下单后扣减库存：下单后立即扣减库存，会最大限度地减少服务器的压力，因为最多只有有限个下单成功的用户。但是这种模式有一个弊端，因为我们无法保证下单成功的用户最后都会去付款，所以可能出现部分用户一直不付款，但是却一直占用资源的情况，导致商品没办法全部卖出去。
付款后扣减库存：只要用户不付款库存就一直在，会保证商品能全部卖出去。但是这种模式同样有一个弊端，如果付款订单的数量没有达到库存上限，那么所有的下单请求都会成功，会导致在某一时刻下单成功的信息激增，然后绝大部分用户付款时，由于库存不足导致无法付款成功，从而给用户带来不好的体验。
下单后预扣减库存：即下单后保留下单信息一段时间，如果没有付款就释放库存，并将下单信息置为失效，虽然没有解决什么实质上的问题，但是作为以上两种情况的折中方案中和了两者的优势和弊端，目前大部分下单网购都是使用的这种方式。

后面的DEMO里面使用的付款后扣减库存的方案，秒杀系统嘛，买不到很正常（主要是偷个懒，逻辑处理相对简单）。

高性能

动静分离

因为秒杀系统的页面一般内容一般来说是不会有变化的，所以我们大可不必每次刷新页面就去请求诸多后端接口。
所以我们上来就先要理清楚哪些信息是可以固定不变的，哪些信息是必须要后台提供的。就比如说，针对某个商品的秒杀活动，这个商品的价格、产品介绍、优惠信息一般是不会改变的，所以这些信息我们可以直接设置为页面上写死的数据，减少对后端服务的请求次数。

静态资源缓存

推荐使用 CDN （内容分发网络）， CDN 会将数据从源服务器复制到其他服务器上。当用户访问时，CDN 会根据自己的负载均衡选择一个最优的服务器，然后用户在这个最优服务器上访问内容，如果该服务器上没有目标资源，则会进行回源（从源服务器获取信息）。
因为我对 CDN 的原理也不是很了解，只了解可以提高静态页面的访问速度，所以这里不做过多的说明。

流控

为了缓解秒杀时刻的访问量带给服务器的巨大压力，我们可以在处理请求之前就适当的筛掉部分请求，即进行流控降级，比如一秒内如果有10000个请求同时命中服务器，那我们只允许其中100个请求能够进入真正的业务逻辑中，剩下的请求会返回一个降级响应，这个降级响应一般来说都是直接返回业务响应失败的信息，婉转一点可以返回“服务忙，请重试”。
Sentinel 提供了这样的流控功能，Sentinel允许我们为指定的接口设置流控规则，我们可以通过 QPS 或者并发线程数设置阈值，使用QPS的话可以控制每秒最多做出多少有效响应，而使用并发线程数则会控制系统线程数量不超过预设值。
在这里插入图片描述

缓存数据库

提高系统性能必然少不了缓存数据库的帮助，现在最常用之一的缓存数据库 Redis 就很适合这种秒杀场景，Redis 中的多路复用技术以及在内存上操作数据的设计造就了 Redis 的高吞吐量，使得它能够在短时间之内响应更多的查询。

消息队列

在业务开发中，有一些逻辑我们往往不需要及时完成，我们只需要关注最核心的业务，部分附带的逻辑可以推迟执行。
我们或许可以考虑使用多线程处理，但是使用多线程可能会存在以下问题：
1、线程仍旧是在当前进程之内执行的，虽然加块了接口的响应速度，但是服务器的负压还是一样的，并没有实际为服务器减少
什么压力。
2、如果逻辑中存在对持久数据库的操作，使用多线程可能在某一时刻会有大量的写操作涌入数据库，数据库的并发能力是相对比较弱的，过多的写请求可能导致数据库宕机。
所以我们可以考虑消息队列，这里使用的是 RabbitMQ 。

RabbitMQ的优点

异步执行：可以提高接口响应速度，由不同的进程服务去消费消息处理剩余工作，减轻当前服务器的压力。
削峰填谷：比如设置每秒最大的消息处理量为1000，请求过多的时候，会控制在每秒处理1000个消息，其余消息就会积压在 RabbitMQ 的队列中，这就是削峰。高峰期过后，请求量下降，但是这个时候由于存在高峰期积攒的消息，所以在短时间内它仍能够提供一定的量级的消息，这就是填谷。这样将缓解在请求高峰期带给数据库服务器的压力，避免数据库崩溃。

高并发

分布式锁

对于要求一致性的数据，比如说剩余库存，我们在查询的时候不允许有任何进程去修改这个值，以供我们做出正确的判断。在微服务项目中，一般会提供多个服务实例，其中一个服务修改库存剩余值的时候，其他任何服务进程都不允许读写该值，这个时候我们往往需要用到分布式锁去处理。
因为我们用到 Redis 做缓存数据库，所以我们可以使用 Redis 实现分布式锁。 Redission 提供了操作 Redis 和获取分布式锁的便捷方法。

后端代码实现

这里实现的是一个简单的秒杀系统，采用的是付款后再扣减库存的方案，以保证商品尽可能的卖出去，而且限制每人只能购买一次。
该DEMO功能还不够完善，下面的代码不是完整代码，完整代码链接附在最后。

中间件

Mysql：不管怎么样我们的数据最终是需要落库的，持久化数据库是必须的。
Reids：作为缓存服务器，秒杀所需的基本信息都存在 Redis 中。
RabbitMQ：逻辑中对持久化数据库的操作以及其他不重要的逻辑处理会影响接口的响应速度，所以使用 RabbitMQ 去通知另外的服务进程操作。
定时任务：用于将我们设置的秒杀信息存入到 Redis 中，可以采用定时任务框架例如 elastic-job、xxl-job等，这里使用 springframework 自带的定时任务框架。

表结构

-- 秒杀信息表
CREATE TABLE `seckill` (
  `id` int NOT NULL AUTO_INCREMENT,
  `product_id` int DEFAULT NULL COMMENT '商品ID',
  `count` int DEFAULT '0' COMMENT '秒杀库存',
  `seckill_price` decimal(10,2) DEFAULT NULL COMMENT '秒杀价格',
  `start_time` datetime DEFAULT NULL COMMENT '秒杀开始时间',
  `end_time` datetime DEFAULT NULL COMMENT '秒杀结束时间',
  `is_cached` tinyint DEFAULT '0' COMMENT '是否放入Redis缓存 0 否 1 是',
  PRIMARY KEY (`id`) USING BTREE,
  UNIQUE KEY `UK_PRODUCT_CODE` (`product_id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
-- 订单表
CREATE TABLE `seckill_orders` (
  `id` int NOT NULL AUTO_INCREMENT COMMENT 'id',
  `order_no` varchar(32) DEFAULT NULL COMMENT '订单号',
  `account_id` int DEFAULT NULL COMMENT '账户ID',
  `seckill_id` int DEFAULT NULL COMMENT '秒杀ID',
  `count` int DEFAULT NULL COMMENT '秒杀数量',
  `payment_amount` decimal(10,0) DEFAULT NULL COMMENT '应付金额',
  `checkout_time` datetime DEFAULT NULL COMMENT '下单时间',
  `status` tinyint DEFAULT '0' COMMENT '状态 0待支付 1已支付 2已取消',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=176 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

添加依赖

这里只放了部分依赖，全部依赖信息比较多，可以查看最后的源码链接
配置文件都是比较基础的单机配置，有些甚至是默认配置比如说 RabbitMQ，这里就不贴了

<!-- Sentinel 配置中心 -->
<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
    <version>2021.0.4.0</version>
</dependency>
<!-- Redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>2.6.6</version>
</dependency>
<!-- Redission -->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.9.1</version>
</dependency>
<!-- Mysql -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>
<!-- MybatisPlus -->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.1.1</version>
</dependency>
<dependency>
	 <groupId>com.baomidou</groupId>
	 <artifactId>mybatis-plus-core</artifactId>
	 <version>3.1.1</version>
	 <scope>compile</scope>
</dependency>
<!-- RabbitMQ -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>1.5.2.RELEASE</version>
</dependency>

公共常量

主要是用于查询 Redis 信息或者获取分布式锁时的前缀，可以放在公共常量类中。

public class CommonConst {
    public static String SECKILL_LOCK_USER = "SECKILL_USER_LOCK:";//用户个人锁
    public static String SECKILL_LOCK_GLOBAL = "SECKILL_GLOBAL_LOCK:";//全局锁
    public static String SECKILL_START_TIMESTEMP = "SECKILL_START_TIMESTEMP:";//秒杀开始时间
    public static String SECKILL_STOP_TIMESTEMP = "SECKILL_STOP_TIMESTEMP:";//秒杀结束时间
    public static String SECKILL_REMAIN_COUNT = "SECKILL_REMAIN_COUNT:";//秒杀商品剩余数量
    public static String SECKILL_ORDER_USERS = "SECKILL_ORDER_USERS:";//下单成功用户列表
    public static String SECKILL_SUCCEED_USERS = "SECKILL_SUCCEED_USERS:";//付款成功用户列表
}

实体类

@Data
public class Seckill {
	//对应数据库表 seckill 字段
}
@Data
public class SeckillOrders implements Serializable {
	//对应数据库表 seckill_orders 字段
}

Redission配置

这里使用 Redisssion 去操作 Redis ，因为它提供简单的锁操作。

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() {
        // 配置信息
        Config config = new Config();
        //地址、密码
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");//                .setPassword("pwd");
        return Redisson.create(config);
    }
}

定时任务

这里做了一个简单的定时任务，每隔五秒执行一次，将没有缓存过的秒杀信息放入 Redis 中

@Component
@EnableScheduling
public class SeckillTask {
    @Autowired
    SeckillDao seckillDao;
    @Autowired
    RedissonClient redission;

    /**
     * 5分钟执行一次
     * 查询没放进缓存中的秒杀任务并放入缓存
     */
    @Scheduled(cron="0/5 * * * * ?")
    public void seckillRedisCache() {
        //将秒杀信息缓存进REDIS
        QueryWrapper<Seckill> ew = new QueryWrapper<>();
        ew.eq("is_cached",0);
        List<Seckill> seckills = seckillDao.selectList(ew);

        if(CollectionUtils.isNotEmpty(seckills)){
            List<Long> cachedIds = new ArrayList<>();
            for (Seckill seckill : seckills) {
                RBucket<Integer> count = redission.getBucket(CommonConst.SECKILL_REMAIN_COUNT + seckill.getId());
                //先判断下确实没有缓存过
                if(!count.isExists()){
                    count.set(seckill.getCount());
                    redission.getBucket(CommonConst.SECKILL_START_TIMESTEMP + seckill.getId()).set(seckill.getStartTime());
                    redission.getBucket(CommonConst.SECKILL_STOP_TIMESTEMP + seckill.getId()).set(seckill.getEndTime());
                }
                cachedIds.add(seckill.getId());
            }

            //修改缓存状态
            if(CollectionUtils.isNotEmpty(cachedIds)) {
                UpdateWrapper<Seckill> uw = new UpdateWrapper<>();
                uw.setSql("is_cached = 1");
                uw.in("id", cachedIds);
                seckillDao.update(null,uw);
            }
        }
    }
}

Controller

增加 Sentinel 限流设置，被限流的接口返回 “活动火爆，请重新尝试!” 的降级响应。
需要注意的几点是：
1、除了 BlockException 这个参数外，降级的方法其他参数类型需要和被流控的接口参数类型保持一致
2、如果是通过 Sentinel 控制台去设置流控规则，程序启动后会发现在没有任何资源，这个时候我们只需要调用一次被流控的接口就好了，然后就可以为其添加流控规则。
调用前
在这里插入图片描述
调用后

@Autowired
private ISeckillService seckillService;

/**
 * 流控降级
 * */
public R<Orders> seckillFallback(Long accountId,Long pid,BlockException ex) {
    return R.failure("活动火爆，请重新尝试！");
}

//秒杀接口（通过Sentinel限流）
@PostMapping("/seckill/{accountId}/{pid}")
@SentinelResource(value="seckill",blockHandler = "seckillFallback")
public R seckill(@PathVariable("accountId") Long accountId,@PathVariable("pid") Long pid) throws Exception {
    return seckillService.seckill(accountId,pid);
}

//支付接口
@PostMapping("/killpay/{seckillOrder}")
public R killpay(@PathVariable("seckillOrder") String seckillOrder) throws Exception {
    return seckillService.killpay(seckillOrder);
}

下单接口

下单之前我们要判断活动是否开始、是否已经结束、是否已经抢购成功过、是否已有下单记录、是否还有库存等。
同时我们要处理同一用户出现的并发现象，因为过快的请求可能是非正常情况，我们可以设法拒绝一些非正常请求的继续访问。

public R seckill(Long accountId,Long kid) throws InterruptedException {
        //获取活动开始时间
        RBucket<Date> startTime = redission.getBucket(CommonConst.SECKILL_START_TIMESTEMP + kid);
        if(!startTime.isExists() || new Date().compareTo(startTime.get())<0) {//获取不到表示活动还未开始
            return R.failure("活动未开始!");
        }

        //获取活动结束时间
        RBucket<Date> stopTime = redission.getBucket(CommonConst.SECKILL_STOP_TIMESTEMP + kid);
        if(new Date().compareTo(stopTime.get())>0){//判断活动是否结束
            return R.failure("活动已结束!");
        }

        //获取用户个人锁，处理同一用户同时多次请求秒杀接口
        //采用自动释放锁的方式，500ms后自动释放，500ms内统一用户的请求视为非正常请求
        RLock lock = redission.getLock(CommonConst.SECKILL_LOCK_USER + kid + ":" + accountId);
        boolean locked = lock.tryLock(0,500,TimeUnit.MILLISECONDS);
        if(locked){

            //判断是否已经购买成功过
            RBucket<Set> succedUsers = redission.getBucket(CommonConst.SECKILL_SUCCEED_USERS + kid);
            if(succedUsers.isExists() && succedUsers.get().contains(accountId)){
                return R.failure("抢购次数已用尽!");
            }

            //判断是否有下单记录
            RBucket<Set> checkoutUsers = redission.getBucket(CommonConst.SECKILL_ORDER_USERS + kid);
            if(checkoutUsers.isExists() && checkoutUsers.get().contains(accountId)){
                return R.failure("已有下单记录，请前往支付!");
            }

			//判断是否还有库存(下单时做初步判断，防止没有库存了仍旧能下单。)
            RAtomicLong count = redission.getAtomicLong(CommonConst.SECKILL_REMAIN_COUNT + kid);
            if(!count.isExists() || count.get()<=0) {
                return R.failure("已售罄!");
            }

            //写入下单成功的人员列表 操作时要获取锁，避免其他进程读取或者操作
            RLock gwlock = redission.getLock(CommonConst.SECKILL_LOCK_GLOBAL + kid );
            if(gwlock.tryLock(1000, TimeUnit.MILLISECONDS)) {
                Set<Long> newUsers = new HashSet<>();
                if(checkoutUsers.isExists()){
                    newUsers = checkoutUsers.get();
                    newUsers.add(accountId);
                }else{
                    newUsers.add(accountId);
                }
                checkoutUsers.set(newUsers);
                //释放写锁
                gwlock.unlock();

                String secOrder = UUID.randomUUID().toString().replace("-","");//返回订单标志
                //生成下单所需基本信息，例如：账户、秒杀ID，
                SeckillOrders checkout = new SeckillOrders();
                checkout.setOrderNo(secOrder);
                checkout.setAccountId(accountId);
                checkout.setSeckillId(kid);
                checkout.setCount(1);
                checkout.setRabbitMqType(0);
                //放进消息队列中处理 我这里交换器 my-mq-exchange_A 绑定的是队列 QUEUE_A 绑定的路由键是 spring-boot-routingKey_A
                RabbitPublishUtils.sendDirectMessage(RabbitMQExchanges.EXCHANGE_A, RabbitMQRoutingKeys.ROUTINGKEY_A, checkout);

                // 最后返回下单单号供前端刷新界面使用
                return R.success(secOrder,"下单成功");
            }else{
               return R.failure("活动火爆，请重新尝试!");
            }
        }else{
            return R.failure("操作频繁!");
        }
    }

付款接口

主要涉及库存校验、库存的扣减和恢复操作。

    public R killpay(String seckillOrder) throws InterruptedException {
        //订单信息从数据库中查询 防篡改
        QueryWrapper<SeckillOrders> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("order_no",seckillOrder);
        SeckillOrders checkout = ordersDao.selectOne(queryWrapper);
        Long kid = checkout.getSeckillId();

        RBucket<Set> succedUsers = redission.getBucket(SECKILL_SUCCEED_USERS + kid);

        //库存先扣减1 写锁控制下不让其他进程读取和修改库存值
        RLock glock = redission.getLock(CommonConst.SECKILL_LOCK_GLOBAL + kid );
        if (glock.tryLock(1000, TimeUnit.MILLISECONDS)) {
            RAtomicLong count = redission.getAtomicLong(SECKILL_REMAIN_COUNT + kid);
            if (count.isExists() && count.get() > 0) {
                count.getAndDecrement();
            }else{
                glock.unlock();
                return R.failure("已售罄!");
            }

            //添加到购买成功的人员列表中
            Set<Long> newUsers = new HashSet<>();
            if(succedUsers.isExists()){
                newUsers = succedUsers.get();
                newUsers.add(checkout.getAccountId());
            }else{
                newUsers.add(checkout.getAccountId());
            }
            succedUsers.set(newUsers);
			//释放写锁
            glock.unlock();
        }

        //扣减数据库中的库存交由消息队列中去处理
        checkout.setRabbitMqType(1);
        RabbitPublishUtils.sendDirectMessage(RabbitMQExchanges.EXCHANGE_A, RabbitMQRoutingKeys.ROUTINGKEY_A, checkout);

        try {
            //TODO 调用支付接口
            //模拟支付失败的操作 用来查看库存恢复是否正常
//            int sd = 0;
//            Object sds = 10/sd;
        }catch (Exception ex){

            //支付失败 恢复商品数量 可以交由消息队列中去处理
            checkout.setRabbitMqType(2);
            RabbitPublishUtils.sendDirectMessage(RabbitMQExchanges.EXCHANGE_A, RabbitMQRoutingKeys.ROUTINGKEY_A, checkout);
            if (glock.tryLock(1000, TimeUnit.MILLISECONDS)) {
            	//缓存库存恢复
                RAtomicLong count = redission.getAtomicLong(SECKILL_REMAIN_COUNT + kid);
                count.getAndIncrement();
				//购买成功列表移除当前用户
                Set<Long> newUsers = new HashSet<>();
                if(succedUsers.isExists()){
                    newUsers = succedUsers.get();
                    newUsers.remove(checkout.getAccountId());
                }
                succedUsers.set(newUsers);
				//释放写锁
                glock.unlock();
            }
            return R.failure("支付失败");
        }

        //下单状态修改 改为已付款
        checkout.setStatus(1);
        ordersDao.updateById(checkout);

        return R.success();
    }

接收通道消息

@Component
public class RabbitMqReceiver {

    @Autowired
    SeckillOrdersDao seckillCheckoutDao;

    @Autowired
    SeckillDao seckillDao;

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = "QUEUE_A", durable = "true", ignoreDeclarationExceptions = "true"),
            exchange = @Exchange(value = "my-mq-exchange_A"), key = "spring-boot-routingKey_A", ignoreDeclarationExceptions = "true"))
    public void handleMessage(SeckillOrders checkout){
        if(null != checkout){
            switch (checkout.getRabbitMqType()){
                case 0://生成订单以及其他业务逻辑
                    Seckill seckill = seckillDao.selectById(checkout.getSeckillId());
                    BigDecimal price = seckill.getSeckillPrice();
                    BigDecimal payment = price.multiply(new BigDecimal(checkout.getCount().toString()));
                    checkout.setPaymentAmount(payment);
                    checkout.setCheckoutTime(new Date());
                    //下单信息落库
                    seckillCheckoutDao.insert(checkout);
                    break;
                case 1://扣减库存以及其他业务逻辑
                    UpdateWrapper<Seckill> updateWapper = new UpdateWrapper<>();
                    updateWapper.eq("id",checkout.getSeckillId());
                    updateWapper.setSql("count = count - 1 ");
                    seckillDao.update(null,updateWapper);
                    break;
                case 2://恢复库存以及其它业务逻辑
                    updateWapper = new UpdateWrapper<>();
                    updateWapper.eq("id",checkout.getSeckillId());
                    updateWapper.setSql("count = count + 1 ");
                    seckillDao.update(null,updateWapper);
                    break;
                default:
                    break;
            }
        }
    }
}