(秒杀项目) 4.8 异步化扣减库存（核心）

一、安装RocketMQ

官方下载地址：http://rocketmq.apache.org/dowloading/releases/
打开下载地址找到要下载的版本、复制链接。

然后再Linux中使用命令
wget 链接进行下载，然后使用unzip命令解压缩。
解压完之后给他最高的可执行权限
执行 chmod -R 777 rocketmq-all-4.8.0-bin-release
更改配置
修改相应的配置、应为默认的内存大小是4个G、我们要根据需求改成我们合适的大小。
修改runserver.sh

vim runserver.sh

然后修改第82行的配置为如图所示、改好之后保存退出。

再改runbroker.sh、改好之后保存退出。
再去改conf文件夹下的配置文件broker.conf、增加23、24行
至此就配置好了。

启动RocketMQ
进入解压文件目录下、输入如下命令后台启动
先启动mqnamesrv、输入启动命令后有如下日志输出表示成功

nohup sh ./bin/mqnamesrv -n localhost:9876 &

再启动mqbroker 、输入启动命令后有如下日志输出表示成功

nohup sh ./bin/mqbroker -n localhost:9876 -c ./conf/broker.conf &

二、异步扣减库存（核心）

1. 为什么要异步扣减库存

因为下单时候扣减库存马上要去数据库修改数据、直接去数据库修改数据是比较消耗性能的、并发量一大会造成性能很差体验很不好、所以我们可以直接在下单的时候先不去修改数据库中的库存表，先修改缓存中的数据，然后过一段时间再去修改数据库中的数据，这样就可以大大提高秒杀下单时的并发性能了，而这个功能通过rocketmq消息队列中的事务型消息实现的。

上面说到先扣减缓存中的数据、然后使用异步的方式去扣减数据库中的数据。这样的话就涉及到数据的一致性问题了，就是本地缓存中扣减了库存但有可能消息在传送的过程中没有被消费者消费从而导致数据库中没有扣减库存，那么如何去保证本地事务和消费者消费的一致性呢。这就要使用到rocketmq中的事务型消息了。

官方文档：https://github.com/apache/rocketmq/blob/master/docs/cn/design.md

2. 下单时保证本地事务和消费者消费的一致性问题(重中之重)

这个过程使用到了rocketmq中的事务性消息。总共分成了8步

生产者先向mq服务（mqserver由servername和broker server组成）发送一个消息，这个消息在内部为hard消息是一种半成品消息，意思是向mqserver打招呼，要向他发送消息了。
然后mqserver接收到了消息之后会返回一个ok，表示已经准备好了。
准备好之后在生产者中就会执行本地事务。
事务成功之后生产者会向发送一个commit消息表示事务执行成功、假设事务失败就会发送一个事务回滚消息rollback消息
但是4步骤有可能在一些网络环境差等意外情况下发送不到mqserver、这样的话这条消息是无法被处理的不管本地事务有没有执行成功、这条消息会一直卡在这。这里mq使用了回查机制来解决这个问题。
mqserver在接收到send并成功返回ok后会每隔一段时间会通知生产者检查事务有没有成功、默认第一次是30s后来每次间隔时间会越来越长。
接收到mqserver发送过来的check之后mqserver会去检查事务有没有执行成功
将事务检查的结果再次发送，成功就发送commit消息，失败发送rollback消息。
接收到commit或者rollback消息之后才会把消息放到正式队列中、接收到的是commit通知消费者消费、如果是rollback则消费者不消费会撤销这条消息。

本项目中的异步扣减库存为了保证缓存和mysql的异步扣减库存的一致性使用的就是上面mq的事务型消息实现的，只是在上面的一些步骤中加了自己的逻辑代码。

大概如下图

根据步骤进行说明：

步骤1在发送消息时发送一个要扣减库存的消息、并且生成一个流水，流水的目的是用于check的，后面会说到。
然后会在执行本地事务中创建订单扣减本地缓存中的库存、并且更新项目中是在创建订单时扣减库存的、假设扣减成功了就会到步骤4表示已经扣减成功，然后最终会被消费者消费这条消息、消费者消费消息时的逻辑就是扣减mysql数据库中的库存。
假设步骤4的消息没有成功发送到mqserver，那么上面说了会触发mqserver的2次检查机制、那么事务中检查什么比较好呢，答案是流水、流水表的锁粒度是远远比订单的锁粒度要小的、因为流水表的值是递增、所以锁的时候我们只锁一条流水，而不是整个库存。所以在步骤3的事务中会去更新流水，步骤6检查check检查时检查流水有没有更新成功，更新成功就表示事务成功没成功就表示事务失败。。
为什么要在发送消息时生成流水呢、因为如果在事务时生成流水假设事务失败，那么连本次流水都没有后面还怎么检查流水。

上面大概的逻辑做好了、接下来就是看项目中对应的代码实现了

3.代码实现

使用事务型消息逻辑实现异步化扣减库存对应项目中的实现代码大体位置如下图

更新销量代码实现

先看orderserviceimpl、由代码可知、是通过异步方式更新销量的、通过发送一个包含itemid商品id和商品数量进行更新的。其中是消息是通过一个json形式转化为字符串之后再转为二进制消息发送给消费者消费的。
接着看更新销量数据的消费者。消费者通过订阅更新订单的消费者信息来监听数据、当接收到消息之后，通过重写onMessage方法在里面先将接收到的消息通过JSONObjective解析成字符串，然后将参数调用itemservice去更新销量。

异步化扣减库存的代码实现

缓存预热代码实现
因为在代码逻辑中要实现先扣减缓存中的数据再扣减数据库中的数据，所以秒杀活动进行之前我们应该先将商品数据加到缓存中去，这叫做缓存预热。本项目没有写后台管理模块、所以通过测试代码模拟、根据下面的代码可知通过将所有item商品全部查出来然后将商品加到redis缓存当中去。

扣减缓存中的商品数量实现
扣减缓存商品数量的实现在ItemServiceImpl中、具体代码如下

流水数据表分析
在本项目中检查的流水对应的表是如图下所示的库存日志表、表id的数据类型使用的是字符类型、不适用数字类型的原因很明显因为商品订单量很大怕越界。然后会存商品的id和要购买的数量，最后一个就是当前商品订单的状态默认是0表示没有被处理、1表示订单创建成功、2表示订单创建失败。而在异步扣减库存的事务型消息中二次检查检查的就是status的值。

对应逻辑图的最后一步的扣减库存实现

对应的实现代码

生产者的监听实现代码

本地消息也就是第3步中创建订单的实现

本地检查方法的实现(对应步骤6检查流水)

本地检查是通过检查流水的状态实现的、在实现代码中可以得到答案

第1步发送消息的实现
这里发送消息是通过controller发送的、传入的参数有用户id、商品数量、活动id

对应的service实现代码

到此异步扣减库存的步骤代码就看完了。但这样还是很乱现在整理下

三、异步化扣减库存代码实现流程整理

对应逻辑中的第一步发送消息、发送消息进入的是 controller的create方法、然后传递用户id、商品id、商品数量、活动ID过去、然后createOrderAsync进行处理、service中会先去有没有该商品如果没有就抛出一个告窑异常给controller、假如没有告窑就会去创建一个流水、创建流水就是向流水日志表中添加一条记录添加的记录id为随机生成的uuid、商品id、数量还有一个用于二次检查的status状态码默认是0、0表示该流水暂时没有被处理，接着将传过来的用户参数使用json封装成本地事务（本地事务涉及了创建订单操作）需要的参数。将商品id、数量、库存日志id使用json封装成消息参数（消息最终被消费者消费和检查是涉及到库存日志id的）。最后就是发送一个秒杀主题，扣减库存tag的消息了，投递消息后成功后，假设本地事务执行失败会向controller抛出一个订单创建失败异常、目的是让用于知道订单创建失败了。
消息发送成功之后接着就是执行本地事务了、执行本地事务是在一个本地事务生产者代码里。他会先通过监听然后执行对应的方法、接收到秒杀主题、tag为扣减库存的消息之后，就会去执行创建订单的方法，该方法会异步更新销量并且将扣减库存日志记录的状态码改为1、表示执行成功、也就是执行一个创建订单的事务创建订单成功后会返回一个commit成功的状态码，commit表示事务成功。这里对应的是图中的3
当commit之后表示事务成功、事务成功之后生产者会将信息发送给mq服务器中的队列中、然后被消费者消费，消费者消费中的逻辑对应的是扣减数据库中的库存。这里对应的是图中的最后一步。
当在并发量大本地事务成功后product发送的commit没有被mq服务器接收到就会隔一段时间执行check、这个实现是在product中实现的，他是根据库存日志中的状态码字段进行判断的、对应的是图中的check阶段的逻辑。具体逻辑如下

四、总结

异步扣减库存功能是本项目中最最核心的内容也是难点和亮点、一定要理解并熟记该流程。
面试时的热点问题有：涉及的知识点是rocketmq的特性

消息丢失问题。
面试时面试官可能会问你你的项目是怎么处理消息丢失问题的、这里涉及到的是rocketmq中的消息可靠性的特性

消费者消费失败问题
消费失败的原因有可能是消息本身的原因、也有可能是网络的原因、rocketmq中有一种消息重试的机制解决这个问题、他会将失败的消息放到一个主题为“%RETRY%+consumerGroup”的重试队列中、然后会隔一段时间去重试消费、重试的次数越多间隔的时间就越长。这样就在一定程度上解决了消费者消费失败的问题。
重复消费问题
重复消费是因为生产者发送消息时发送失败后会重试重新发送消息、这就会导致消费者消费的消息重复问题、在rocketmq中重复消费是无法避免的、但在一般情况下不会发生、当出现消息量大、网络抖动，消息重复就会是大概率事件。另外，生产者主动重发、consumer负载变化也会导致重复消息。
死信队列