秒杀系统的总结实现(单机秒杀，分布式绕道，我没学)

**实现基本工具：**redis -------->rabbitMq----------->dao

**基本问题：**只是列出一些最基本的问题，当然还有其他的问题，比如接口安全性等等

1. 瞬时涌入大量请求，导致服务器卡死瘫痪
2. 大量用户同时请求削减库存，导致超卖问题

压测结果：

使用了Jmeter进行压测，具体测试流程不在赘述，直接贴出结果，同时抢购50件

1. 瞬时500并发

   次数：

   

   库存：

   

   订单：

   

   订单是否重复？执行语句 select count(*) from participation group by limit_activity_id,user_id;

   

2. 瞬时5000并发

   次数：

   

   库存：

   

   订单：

   

   是否重复：

   

3. 瞬时10000并发

   次数：

   

   库存：

   

   订单：

   

   是否重复：

   

4. 20000次也测试过，么有问题，但是电脑差点报废

解决方案：

1. 该方法目前技术栈无法完善解决，只了解到使用nginx的漏桶进行限流，以及先进入rabbitmq进行削峰

2. 主要解决库存问题，防止库存出现错误

   **主要流程：**本项目是一个骑行APP项目，出现秒杀是因为，项目业务中有一个限时活动奖励的模块，冠军可得丰厚的奖励，限时开启，限时结束。主要就是让本应该mysql进行抗压的操作，转移到redis进行，redis本身具有速度快，抗压能力高，操作命令都是原子性的特点，所以很合适不过，通过redis预减库存，预减成功后，将其加入RabbitMq队列中，使用消费者操作数据库，并加上事务，主要防守手段就是层层判断筛选过滤漏网之鱼。

   数据库

   表名：limit_activity (限时活动表)

   

   表名：price (奖品)

   

   表名：participation (参加人员表,类似于商品订饭表)

   

   Controller层

   //用于获取限时活动，也就是说获取limit表中的数据，很简单，一个查询
   @GetMapping(value = "/getLimitActivity")
   public ResponseResult getLimitActivity() {
       return ResponseResult.ok(activeService.getLimitActivity());
   }
   //秒杀成功查询，由于使用队列延迟入库，所以需要前端进行轮询查找，这里有三个返回值
   //1 -----> 表示活动参与(秒杀)成功
   //0 -----> 表示虽然参加秒杀了，但是没有秒杀成功，也就是有人顶替了
   //-1 -----> 表示还在秒杀中，没有结果
   @GetMapping(value = "/secKillSuccess")
   public ResponseResult getKillSuccess(Long id) {
       return ResponseResult.ok(activeService.findParticipation(id));
   }
   //秒杀操作，返回值有四个
   //-2 -----> 已经访问过秒杀接口，是为了防止用户通过非法手段，重复访问某些接口
   //-1 -----> 表示已经参加过秒杀了，同样过滤非法请求
   //0 ------> 库存不足，秒杀失败
   //1 ------> 秒杀参与成功，对我们来说，也就是加入了队列，但还需配合上和轮询接口
   @GetMapping(value = "/secKill")
   public ResponseResult secKill(Long id) {
       return ResponseResult.ok(activeService.secKill(id));
   }
   

   Service层

   //不多解释，查找限时活动列表，简单的查询
   //sql ---> status属性不用看，这是为了配合前端查询活动时，直接查询到用户是否参与过的属性，数据库中没有这个字段
   //select id,depart,`end`,start_time,end_time,activity_start_Time,remain,price,(select count(*) from participation where user_id = #{userId,jdbcType=INTEGER} and limit_activity_id = limit_activity.id) as status from limit_activity order by start_time desc
   @Override
   public List<LimitActivity> getLimitActivity() {
       return activeDao.findAllLimitActivity(RequestUtil.getUserId());
   }
   //插入活动记录，类似于商品秒杀中的订单
   //sql -----> 
   //insert into participation(limit_activity_id, user_id, time) values (#{limitActivityId,jdbcType=BIGINT}, #{userId,jdbcType=BIGINT}, #{time,jdbcType=TIMESTAMP})
   @Override
   public Integer insertParticipation(Participation participation) {
       return activeDao.insertParticipation(participation);
   }
   
   //减库存,注意这里的sql，通常写法会写成下面这样，这类sql无法防守超卖问题，也就是说可能导致remain负值
   //update limit_activity set remain = remain - 1 where id = #{id,jdbcType=BIGINT}
   //真正sql ----> 加了一层库存判断，可以有效防止库存变为负数，但无法防止订单超卖，也就是说可能订单数大于库存
   //update limit_activity set remain = remain - 1 where id = #{id,jdbcType=BIGINT} and remain > 0
   @Override
   public Integer minusRemain(Long id) {
       return activeDao.minusRemain(id);
   }
   
   // 查看是否已经入MySql库，用于轮询秒杀结果查询
   @Override
   public Integer findParticipation(Long activityId) {
       Participation participation = new Participation();
       participation.setLimitActivityId(activityId);
       participation.setUserId(RequestUtil.getUserId());
       //查询数据库中是否存在这个人的订单,具体sql不再赘述
       int result = activeDao.hasParticipation(participation);
       // 秒杀成功
       if (result > 0) {
           return 1;
           //过程中可能被别人抢先，返回失败，但一般几率很小
       } else if (activeDao.findRemain(activityId) < 1) {
           return -1;
       } else {
           //队列还未执行到该用户的入库操作，继续轮询
           return 0;
       }
   }
   
   //真正的秒杀接口
   //首先在成员变量初始化三个前缀，用于存储redis，分别为
   // private static final String ACTIVITY_PREFIX = "activity:id:";
   // private static final String USER_PREFIX = "activity:user:";
   // private static final String REPEAT_REQUEST = "activity:repeat:";
   //充分利用redis命令的原子性
   @Override
   public Integer secKill(Long activityId) {
       //项目使用的jwt+shiro，在通过shiro时会自动将userID写入到RequestAttribute中，然后使用工具类获取
       Long userId = RequestUtil.getUserId();
       //拼接限时活动id
       String activityKey = ACTIVITY_PREFIX + activityId;
       //拼接用户id，这个key是用来记录已经参与秒杀的用户列表id
       String userKey = USER_PREFIX + activityId;
       //拼接重复请求id，记录每个用户请求数，防止同一用户多次请求，类似于黄牛~
       String repeat = REPEAT_REQUEST + activityId + ":" + userId;
       //将用户请求数自增1，如果结果大于1，那证明已经请求过了，直接过滤掉,原子性，无并发问题
       if (RedisUtil.incr(repeat, 1) > 1) {
           return -2;
       }
       //判断用户是否已经参加过秒杀，如果已经参加过，直接过滤掉，原子性，无并发问题
       if (RedisUtil.sHasKey(userKey, userId)) {
           //表示已经参加过了，过滤非法请求
           return -1;
       }
       //预检库存，充分利用原子性
       if (RedisUtil.decr(activityKey, 1) < 0) {
   	    //可加入这句话，这样的话redis中的库存永远是0,好看
           //有些人会说，两个原子性的操作，组合在一起，就不是原子性的了，但我在这里压测了很多遍，也没有发生库存异常，我个人理解可能原因是，这两个原子操作在一起，并且使用了if语句判断，所以他只能有一种情况。而我们通常所说的组合不安全的例子也就是这个,ConcurrentHashMap
           // K k = map.get(key);
           // V newV = k == null ? 1 : k + 1;
           // map.put(k,newV);
           //这个语句与此处的区别就是，两个线程先后获得k，重点就在于第二个语句，发现都为null，那结果都是1
           //而这里的情况却不同，decr本身是原子性，所以他这个语句是原子性，加上if判断，他只会有一种情况进入这个if条件，所以说：两个原子性的操作，组合在一起，就不是原子性的 ---->  这个说法不太准确，分具体逻辑业务，具体查看最底下的解释
           RedisUtil.incr(activityKey, 1);
           //库存不足
           return 0;
       }
       //我这里先入redis库，默认认为该人已经秒杀成功，当然如果在RabbitMq发生了宕机，导致后续入MySql库失败，那也只能认用户倒霉
       RedisUtil.sSet(userKey, userId);
       //封装参加记录(订单)
       Participation participation = new Participation();
       participation.setTime(Timestamp.valueOf(LocalDateTime.now()));
       participation.setLimitActivityId(activityId);
       participation.setUserId(userId);
       //发送到消息队列,我这里使用了简单的消息队列，按道理来说，需要加一个死信队列配合使用，防止原生队列发生不可预知的错误而丢失消息
       // 这里的队列名称要与你创建的队列名称相同，我用的是"LimitSeckill"
       rabbitTemplate.convertAndSend(RabbitMqConfig.SEC_KILL, participation);
       //入队成功，给前端标识，让他轮询结果
       return 1;
   }
   

   RabbitMQ的消费端

   //这里就是异步处理用户订单插入生成的消费端，他就是避免了瞬时访问数据库的问题，RabbitMQ的作用不再赘述
   //concurrency属性，也就是说开了五个消费者执行同一件事，将对数据库本该瞬时万级的访问减少到5，不设参数就是一个消费者
   //强烈建议必须开启手动ACK，原因就是，如果开启自动ACK，数据库插入如果失败，他也ACK了的话，就会导致数据丢失，可能会导致库存遗留问题，比如：请求了1000个请求，但中间没有出现问题，刚刚好100个任务进入了队列，一旦100个任务中有失败的，就会导致这个请求无效
   @RabbitListener(queues = RabbitMqConfig.SEC_KILL, concurrency = "5")
   // 参数：delivery ---- 用于手动ACK的参数，编号   participation ---- 需要操作的订单   channel ----- 同样也是ACK必须
   public void executeLimitActivity(@Header(AmqpHeaders.DELIVERY_TAG) long delivery, Participation participation, Channel channel) {
       //手动开启事务，方法上加@Transaction注解，不一定会生效，因为多线程。所以我开了手动，避免出问题
       TransactionStatus transaction = dataSourceTransactionManager.getTransaction(transactionDefinition);
       try {
           // 层层验证，查看数据库库存是否还有
           Integer remain = activeDao.findRemain(participation.getLimitActivityId());
           //如果没有，直接返回，并且确认消息
           if (remain < 1) {
               //这里我直接将redis的预库存设为0，因为数据库都没了。
               RedisUtil.set(ACTIVITY_PREFIX + participation.getLimitActivityId(), 0);
               //提交事务
               dataSourceTransactionManager.commit(transaction);
               //向RabbitMq确认消费成功
               channel.basicAck(delivery, false);
               return;
           }
           //第二层验证，查看订单库中是否已经存在该订单，防止多重抢单，解决黄牛，如果有，直接返回
           if (activeDao.hasParticipation(participation) >= 1) {
               //提交事务
               dataSourceTransactionManager.commit(transaction);
               //向RabbitMq确认消费成功
               channel.basicAck(delivery, false);
               return;
           }
           // 执行真正的逻辑，也就是减库存，插入订单，按理说，这里不会再出现任何问题，除非数据库宕机，或者MySQL本身出现了问题，但我们手动控制了事务和ACK，也不怕他出问题
           //插入订单
           activeDao.insertParticipation(participation);
           //库存减一，这里的Sql语句有讲究，详情请看上面的service的解释
           activeDao.minusRemain(participation.getLimitActivityId());
           //提交事务
           dataSourceTransactionManager.commit(transaction);
           //确认消息消费成功
           channel.basicAck(delivery, false);
       } catch (Exception e) {
           try {
               // 告诉RabbitMQ，这个消息处理失败，需要你重新入队，下一次我继续尝试处理，但这里一定要注意，自己的程序一定要没问题，如果是你代码本身导致的异常，那么他每次运行都会失败放入队列，这就造成了无限循环，拿出来--->失败---->放进去,这里就类似于重试机制
               channel.basicReject(delivery, true);
           } catch (IOException ex) {
               ex.printStackTrace();
           }
           // 表明事务中发生了错误，手动回滚，保证数据一致性
           dataSourceTransactionManager.rollback(transaction);
           e.printStackTrace();
       }
   }
   

   存在的问题

   1. 很明显，只解决了秒杀中最核心的库存问题，保证不超卖，不遗留，但是其他的优化什么的没有做，这里就不做赘述，优化方案过于高级，目前不在本人的技术范围内。
   2. 有一个浪费资源的问题，请看秒杀接口中，我为了保证原子性，并没有对存入redis的三个key进行过期操作，清理等，久而久之就会导致占用内存，甚至会占用过多，解决办法也很简单。你可以在发布秒杀任务的时候，定义一个定时任务，在秒杀任务结束时，清除相关的key。

   Redis组合原子操作的原子性的理解(单机redis的理解，分布式的不同的redis肯定不适用！)

   //他为什么是原子性呢，使用if包裹，decr本身是原子性的，这是这里的字节码，无论某个线程在哪一行阻塞，都不会导致不安全
   //LCONST_1 //这里是读操作，你阻塞不阻塞。和我没关系
   //INVOKESTATIC com/cycling/utils/RedisUtil.decr (Ljava/lang/String;J)J //这里一定是串行，redis单命令保证原子性
   //LCONST_0 //这里同样是读
   //LCMP  // 这里是比较大小，那为什么这里阻塞也没有关系，因为走到这里，比较的双方已经固定，并且，预减decr操作是串行，保证原子性，他的结果一定不一样，就是1001个线程执行了预检，1001个结果也一定不同，不可能会得到两个一样的数据。如果此时有1000个库存，那只会有最后的第1001个库存会进来。而在第1000次之后的所有请求，他都会进来，并且是顺序进来
   if (RedisUtil.decr(activityKey, 1) < 0) {
       // 那如果第1000次后来的线程中，有一个在这里阻塞了呢？确实会存在这个问题，但是，无论你阻塞了多少个，你上一个decr语句已经执行，
       //也就是说，你进来多少次我减多少次。假如有十个线程进来，发现都预检之后都小于0，都会进来，但其中几个在这里被阻塞了，那又有什么关系，
       我现在库存已经被预减了，库存就是-10，你别的线程再怎么加也只能加一，并且他incr也是原子性，所以最后属于自己的那份还是得自己来加回来。
       RedisUtil.incr(activityKey, 1);
       //库存不足
       return 0;
   }
   

   但是，只有这种情况可以保证原子性，也就是原子加减并原子获取

   //这就不是原子性了
   String old = RedisUtil.get(key);
   //假如两个线程同时执行，第一个线程执行到这里阻塞住了，第二个线程获取old之后发现也为null，也走到了这里，
   //此时第二个线程判断后将newValue赋值为1，
   //第一个线程恢复运行，后面就不再赘述。
   String newValue = old == null ? 1 : old + 1;
   RedisUtil.set(key,newValue)