最近在想着把秒杀业务这个模块的细节做一些具体的思考，以自问自答的形式来展开，也是对技术进一步回顾和提升

在去年开始的做了一个Spring Boot + Spring MVC + MyBatis框架的一个秒杀业务模块，利用shiro组件来实现登录注册，用mysql来作为数据库，RabbitMq来进行应用解耦和异步通信以及利用其死信队列来进行实现订单超时处理，最后在高压Jmeter测试下，发现了许多问题，比如说库存超卖（数据库total的值成为了负数),用户恶意刷单(同一个用户疯狂点击button导致一个用户抢购了很多件商品),这些问题我是通过了一个是优化mysql语句以及用Redis/Redission/Zookeeper来实现分布式锁来进行，解决最后也是可以通过测试。

如何用Spring搭建这个多模块项目？

model模块依赖api模块，server模块依赖model模块，层层依赖。
model：mybatis或者跟数据库mysql相关的类文件与配置文件等等。
sever中的启动类：加载配置文件spring-jdbc.xml
构建好相应的模块之后，就需要往相应的模块添加依赖，即只需要在pom.xml中加入相应的依赖即可
值得一题的是，为了让整个项目在前后端分离开发的情况下，前后端的接口交互更加规范（比如响应信息的规范等等），在这里我们采用了通用的一个状态码枚举类StatusCode 跟 一个通用的响应结果类BaseResponse，用于后端在返回响应信息给到前端时进行统一封装。

介绍一下整体业务流程和数据库是怎么实现的?

在进行数据库设计之后，采用Mybatis逆向工程生成相应的实体类Entity、操作Sql的接口Mapper以及写动态Sql的配置文件Mapper.xml。

待秒杀商品列表与详情功能是怎么开发的？

控制器的将方法在获取到待秒杀商品的列表信息后，将通过modelMap的形式将数据列表返回给到前端的页面list.jsp中进行渲染
服务->接口->Mapper来操作数据库
sql判断逻辑是当前服务器的时间now()处于秒杀时间范围内且total库存大于0，设置状态为cankill 为1 否则为0
获取商品详情同样的服务->接口->Mapper 主要是基于Mybatis在配置文件中 动态写sql 关于主键的精确查询 返回给前端的数据放在modelmap中
为了避免有人“跳过页面的请求，直接恶意刷后端接口”，在该页面仍然再次进行了一次判断detail.cankill == 1,在后面执行“抢购/秒杀”请求时，后端接口还会再次进行判断的，所有这些都是为了安全考虑

如何实现用户登录认证？

将整合权限认证-授权框架Shiro，实现用户的登录认证功能，主要是在用户进行秒杀和抢购商品，限制用户进行登陆！对于特定的url进行过滤。shiro本身就是一个很好的用户身份认证和权限授权框架，可以实现用户的登录，权限，资源授权，会话管理。这里主要涉及到的是用户身份的认证和用户的登录功能。
MD5加密的方式进行匹配，数据库的user表中用户的密码字段存储的正是采用MD5加密后的加密串
当前端提交“用户登录”请求时，将以“提交表单”的形式将用户名、密码提交到后端UserController控制器对应的登录方法中，该方法首先会进行最基本的参数判断与校验，校验通过之后，会调用Shiro内置的组件SecurityUtils.getSubject().login()方法执行登录操作，其中的登录操作将主要在 “自定义的Realm的具体验证方法”中执行。最主要的是开发自己的Realm，开发自定义的Realm方法，并实现其中的用户登录方法。
当用户登录成功时（即用户名和密码的取值跟数据库的user表相匹配），我们会借助Shiro的Session会话机制将当前用户的信息存储至服务器会话中，并缓存一定时间！（在这里是3600s，即1个小时）！这里是保持用户的登录状态。
我们需要借助Shiro的组件ShiroFilterFactoryBean 实现“用户是否登录”的判断，以及借助FilterChainDefinitionMap拦截一些需要授权的链接URL，定义相应的URL对应需要的权限，FactoryBean会对需要登录的URL进行拦截，当用户访问这些网站时候，要求用户需要登录。同时，ShiroFilterFactoryBean也设定了重定向的具体网页。

商品秒杀是具体怎么完成的?

用户的Id可以通过Shiro 的会话模块Session进行获取
请求到Controller以后，先开始进行校验之后提取session，调用秒杀服务返回是否秒杀成功。其中killDto封装待秒杀id信息和用户id。
update中进行了where判断了total>0 实现了二次检验 库存是否扣减成功，如果成功执行了update那么返回值为1 创建订单，开始异步生成秒杀消息。
与前端交互的时候，提交的数据是采用application/json的格式提交的，即json的格式！并采用POST的请求方法进行交互！

分布式唯一ID生成订单编号是如何实现？

要满足以下两点？
（1）保证订单编号的生成逻辑要快、稳定，减少时间延迟
（2）要保证生成的订单编号全局唯一、不重复、趋势递增、有时序性
借助了一张数据库表 random_code 来存储生成的订单编号，设定订单编号code唯一，如果高并发订单编号出现重复，那么在插入数据库表的时候必然会出现错误。
思考的时候，采用了两种方式，第一种方式是基于随机数生成唯一ID 时间戳加四位随机数流水号 出现“重复，生成了重复的订单编号”！鉴于此种“基于随机数生成”的方式在高并发的场景下并不符合的要求
第二种方法推特的“雪花算法” ，跟UUID的比较：UUID36位，相对比较长，另外UUID一般是无序的。雪花算法生成的ID简单，64位刚刚是Long型，且是根据时间自增的。

消息异步发送是怎么实现的？

是用RabbitMQ实现的，RabbitMQ实现消息异步通信、业务服务模块解耦、接口限流、消息分发等功能。
RabbitMQ的消息发送组件RabbitTemplate
我们需要在RabbitmqConfig配置类中创建队列、交换机、路由绑定等Bean组件，三个组件
接着在RabbitSenderService 中写一段发送消息的方法，接收“订单编号”参数，在数据库中查询其对应的详细订单记录，充当“消息”并发送至RabbitMQ的队列，等待被监听消费。
最后，是在通用的消息接收服务类RabbitReceiverService中实现消息的接收，

SpringBoot发送邮件服务是怎么实现的？

基于JavaMail服务开发一个通用的发送邮件服务，用于发送邮件通知消息，与RabbitMq异步发送消息的逻辑进行整合彻底实现“用户秒杀成功后，异步发送邮件通知消息给到用户邮箱，告知用户尽快进行付款”的功能！
采用的是qq邮箱作为主邮箱账号，相应的SMTP服务器也是采用QQ邮箱的，其中，spring.mail.password 指的是在QQ邮箱后台开通POP3/SMTP服务 时腾讯官方给的“密钥”（授权码）
MailDto类主要统一封装了在发送邮件时所需要的字段信息，比如接收人、邮件标题、邮件内容等等，体现了面向对象的重要性。
最后是在RabbitMQ通用的消息接收服务类” RabbitReceiverService整合发送邮件服务

RabbitMQ死信队列如何处理超时未支付的订单？

传统的做法是采用“定时器的方式”，定时轮询获取已经超过指定时间的订单，然后执行一系列的处理措施
“死信队列”，顾明思议，是可以延时、延迟一定的时间再处理消息的一种特殊队列，它相对于“普通的队列”而言，可以实现“进入死信队列的消息不立即处理，而是可以等待一定的时间再进行处理”的功能！而普通的队列则不行，即进入队列后的消息会立即被对应的消费者监听消费。

死信队列由三大核心组件组成：死信交换机+死信路由+TTL（消息存活时间~非必需的），而死信队列又可以由“面向生产者的基本交换机+基本路由”绑定而成。生产者首先是将消息发送至“基本交换机+基本路由”所绑定而成的消息模型中，即间接性地进入到死信队列中，当过了TTL，消息将“挂掉”，从而进入下一个中转站，即“面下那个消费者的死信交换机+死信路由”所绑定而成的消息模型中。
首先，需要在RabbitmqConfig配置类创建死信队列的消息模型
首先进入 基本交换机+基本路由 -> 死信队列
过了TTL
进入死信交换机+死信路由->真正队列 的绑定

RabbitMQ发送消息服务类RabbitSenderService中开发“发送消息入死信队列”的功能
指定了消息的存活时间TTL
最后，是需要在RabbitMQ通用的消息监听服务类RabbitReceiverService 中监听“真正队列”中的消息并进行处理：在这里我们是对该订单进行失效处理
添加订单成功后，两步：加入异步邮件消息通知和加入死信队列。

如何处理超时未支付的订单

许多订单记录恰好在某个TTL集中失效，准备交由RabbitMQ的死信队列进行处理时，此时恰好RabbitMQ的服务挂掉了，RabbitMQ的服务突然挂掉，味着在那个时候那些本该被失效处理的订单记录却迟迟没有得到处理。即使重启了RabbitMQ服务，消息也重新进行了处理，但是却有可能在这段时间内用户“重新下不了单”或者其他一些意想不到的问题
“定时器”或者叫“定时任务调度”的方式来辅助处理，@Scheduled，设定一下Cron时间，询获取status = 0（即状态为未支付）的订单，并更新失效掉那些时间已经超过了TTL的订单，set status = -1。
如果项目有多个这样的定时任务，多个定时任务的执行时间Cron如果挨得很近，那么效率一定是瓶颈，这是因为如果只是上面那样写，那么始终只会有一个、单一的线程在执行上面的定时任务。对“定时任务调度”我们加入“线程池”的配置。

你再JMeter压力测试下遇到了哪些问题？

出现了total库存超卖以及同一个人可以抢购多次（可以理解为用户在前端界面疯狂点击），问题产生的根源在于高并发的情况下生成订单的流程的处理并没有为检验流程的处理赢得足够的时间，即“生成一条秒杀成功后的订单记录” 并没有及时的为 “判断用户是否已经秒杀过了~是否已经有对应的订单记录了” 的流程很好的服务！

如何解决的？

1. 从sql语句上，对于通过userId和killId来查询是否用户抢购商品那个service中，在mybatis中加上total>0,并且在保证update库存时候中要保证扣减完之后的数量大于0，即只有在保证扣减完之后的数量大于0之下，该Sql操作后受影响的行数为1。以及在具体的生成订单上仿照“单例模式”中的“双重检验锁”，加上查询用户id和物品id绑定的订单是否已存在和是否成功向数据库插入订单消息的双重检验锁。
2. 基于Redis原子操作的：首先加入Redis的第三方依赖，接着application.properties配置Redis服务所在的Host、端口Port、链接密钥Password等信息，自定义注入Redis相关的Bean设置，RedisTemplate跟StringRedisTemplate操作组件，并指定其对应的Key、Value的序列化策略。最后，采用Redis的原子操作SETNX(SET if Not eXists 只在键 key 不存在的情况下， 将键 key 的值设置为 value 。若键 key 已经存在， 则 SETNX 命令不做任何动作)
   我们设置的key是 final String key=new StringBuffer().append(killId).append(userId).append("-RedisLock").toString();value是RandomUtil.generateOrderCode()一个随机数
   和EXPIRE方法来实现一种分布式锁。我们可以将“KillId和UserId的一一对应关系~即一个人只能抢到一个商品”组合在一起作为Key。并且设置了Key，那么就需要防止万一用个expire来在未来的某一时刻去释放。但是真正的释放是在finally delete语句中，先要判断是不是之前获取到的锁，避免误删锁。如果不释放锁或者执行Expire之前，Redis的节点宕机了，那么此时将很有可能永久进入“Key锁死”的窘境。重启后，由于之前的Key没有得到释放，故而这个Key将永远存在于缓存中，即对应的用户将不能秒杀该商品了。这是一个隐患。
3. 基于Reddison的分布式锁优化：在这里解决被锁死问题，其底层实现机制在于"Redisson内部提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期”,除此之外，Redisson还通过加锁的方法提供了leaseTime的参数来指定加锁的时间，即超过这个时间后锁便自动解开了。Boolean cacheRes=lock.tryLock(30,10,TimeUnit.SECONDS);第一个参数30s=表示尝试获取分布式锁，并且最大的等待获取锁的时间为30s,第二个参数10s=表示上锁之后，10s内操作完毕将自动释放锁，使用了Redisson分布式锁中的“可重入锁”组件。
4. 基于ZooKeeper的分布式锁优化：
   ZooKeeper对外会提供一个多层级的节点命名空间（节点称为ZNode），每个节点都用一个以斜杠（/）分隔的路径表示，而且每个节点都有父节点（根节点除外）。ZooKeeper的相关功能特性在实际使用过程中，其底层可能需要动态的添加、删减相应的节点，此时zk会提供一个Watcher监听器，用以监听那些动态新增、删减的节点，即ZooKeeper会在某些业务场景对一些节点使用上Watcher机制，监听相应的节点的动态。
   基于Watcher机制 + 动态创建、删减临时顺序节点 所实现的，值得一提的是，一个ZNode节点将代表一个路径。
   以下为ZooKeeper实现（获取）分布式锁的原理：
   （1）当前线程在获取分布式锁的时候，会在ZNode节点(ZNode节点是Zookeeper的指定节点)下创建临时顺序节点，释放锁的时候将删除该临时节点。
   （2）客户端/服务 调用createNode方法在 ZNode节点 下创建临时顺序节点，然后调用getChildren(“ZNode”)来获取ZNode下面的所有子节点，注意此时不用设置任何Watcher。
   （3）客户端/服务获取到所有的子节点path之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁，即当前线程获取到了分布式锁。
   （4）如果发现自己创建的节点并非ZNode所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，然后对其调用exist()方法，同时对其注册事件监听器。之后，让这个被关注的节点删除（核心业务逻辑执行完，释放锁的时候，就是删除该节点），则客户端的Watcher会收到相应的通知，此时再次判断自己创建的节点是否是ZNode子节点中序号最小的，如果是则获取到了锁，如果不是则重复以上步骤继续获取到比自己小的一个节点并注册监听。

//定义获取分布式锁的操作组件实例
InterProcessMutex mutex=new InterProcessMutex(curatorFramework,pathPrefix+killId+userId+"-lock");//尝试获取分布式锁
mutex.acquire(10L,TimeUnit.SECONDS) 尝试等待10秒钟获取到分布式锁，如果获取到锁就可以进行进一步操作//释放锁
mutex.release();

基于Zookeeper实现的分布式互斥锁 - InterProcessMutex需要依赖于一个zk客户端的实例，给到一条路径进行操作，在这条路径上不断创建临时有序的结点，哪一个结点的值最小将获取到锁。实际是一个串行。开始的时候会不断创建临时结点，开销还是挺大的，在实际生产环境中，既要保证并发安全且Qps也不是很高的情况下，就可以用zk。要考虑到Qps很高的情况下，就可以使用Redis和Redisson。因为启动的时候会产生很多的路径。
特点：分布式锁（基于Zookeeper），互斥锁，公平锁（监听上一临时顺序节点 + wait() / notifyAll()），可重入。