SpringBoot整合RabbitMQ之典型应用场景实战二

RabbitMQ 实战：并发量配置与消息确认机制

实战背景

对于消息模型中的 listener 而言，默认情况下是“单消费实例”的配置，即“一个 listener 对应一个消费者”，这种配置对于上面所讲的“异步记录用户操作日志”、“异步发送邮件”等并发量不高的场景下是适用的。但是在对于秒杀系统、商城抢单等场景下可能会显得很吃力！

我们都知道，秒杀系统跟商城抢单均有一个共同的明显的特征，即在某个时刻会有成百上千万的请求到达我们的接口，即瞬间这股巨大的流量将涌入我们的系统，我们可以采用下面一图来大致体现这一现象：

当到了“开始秒杀”、“开始抢单”的时刻，此时系统可能会出现这样的几种现象：

• 应用系统配置承载不了这股瞬间流量，导致系统直接挂掉，即传说中的“宕机”现象；

• 接口逻辑没有考虑并发情况，数据库读写锁发生冲突，导致最终处理结果跟理论上的结果数据不一致（如商品存库量只有 100，但是高并发情况下，实际表记录的抢到的用户记录数据量却远远大于 100）；

• 应用占据服务器的资源直接飙高，如 CPU、内存、宽带等瞬间直接飙升，导致同库同表甚至可能同 host 的其他服务或者系统出现卡顿或者挂掉的现象；

于是乎，我们需要寻找解决方案！对于目前来讲，网上均有诸多比较不错的解决方案，在此我顺便提一下我们的应用系统采用的常用解决方案，包括：

• 我们会将处理抢单的整体业务逻辑独立、服务化并做集群部署；

• 我们会将那股巨大的流量拒在系统的上层，即将其转移至 MQ 而不直接涌入我们的接口，从而减少数据库读写锁冲突的发生以及由于接口逻辑的复杂出现线程堵塞而导致应用占据服务器资源飙升；

• 我们会将抢单业务所在系统的其他同数据源甚至同表的业务拆分独立出去服务化，并基于某种 RPC 协议走 HTTP 通信进行数据交互、服务通信等等；

• 采用分布式锁解决同一时间同个手机号、同一时间同个 IP 刷单的现象；

下面，我们用 RabbitMQ 来实战上述的第二点！即我们会在“请求” -> "处理抢单业务的接口" 中间架一层消息中间件做“缓冲”、“缓压”处理，如下图所示：

并发量配置与消息确认机制

正如上面所讲的，对于抢单、秒杀等高并发系统而言，如果我们需要用 RabbitMQ 在 “请求” - “接口” 之间充当限流缓压的角色，那便需要我们对 RabbitMQ 提出更高的要求，即支持高并发的配置，在这里我们需要明确一点，“并发消费者”的配置其实是针对 listener 而言，当配置成功后，我们可以在 MQ 的后端控制台应用看到 consumers 的数量，如下所示：

其中，这个 listener 在这里有 10 个 consumer 实例的配置，每个 consumer 可以预监听消费拉取的消息数量为 5 个（如果同一时间处理不完，会将其缓存在 mq 的客户端等待处理！）

另外，对于某些消息而言，我们有时候需要严格的知道消息是否已经被 consumer 监听消费处理了，即我们有一种消息确认机制来保证我们的消息是否已经真正的被消费处理。在 RabbitMQ 中，消息确认处理机制有三种：Auto - 自动、Manual - 手动、None - 无需确认，而确认机制需要 listener 实现 ChannelAwareMessageListener 接口，并重写其中的确认消费逻辑。在这里我们将用 “手动确认” 的机制来实战用户商城抢单场景。

1.在 RabbitMQConfig 中配置确认消费机制以及并发量的配置

2.消息模型的配置信息

3.RabbitMQ 后端控制台应用查看此队列的并发量配置

4.listener 确认消费处理逻辑：在这里我们需要开发抢单的业务逻辑，即“只有当该商品的库存 >0 时，抢单成功，扣减库存量，并将该抢单的用户信息记录入表，异步通知用户抢单成功！”

5.紧接着我们采用 CountDownLatch 模拟产生高并发时的多线程请求（或者采用 jmeter 实施压测也可以！），每个请求将携带产生的随机数：充当手机号 -> 充当消息，最终入抢单队列！在这里，我模拟了 50000 个请求，相当于 50000 手机号同一时间发生抢单的请求，而设置的产品库存量为 100，这在 product 数据库表即可设

6.将抢单请求的手机号信息压入队列，等待排队处理

7.在最后我们写个 Junit 或者写个 Controller，进行 initService.generateMultiThread(); 调用模拟产生高并发的抢单请求即可

@RestController
    public class ConcurrencyController {

    private static final Logger log= LoggerFactory.getLogger(HelloWorldController.class);

    private static final String Prefix="concurrency";

    @Autowired
    private InitService initService;

    @RequestMapping(value = Prefix+"/robbing/thread",method = RequestMethod.GET)
    public BaseResponse robbingThread(){
        BaseResponse response=new BaseResponse(StatusCode.Success);
        initService.generateMultiThread();
        return response;
    }}

8.最后，我们当然是跑起来，在控制台我们可以观察到系统不断的在产生新的请求（线程）– 相当于不断的有抢单的手机号涌入我们的系统，然后入队列，listener 监听到请求之后消费处理抢单逻辑！最后我们可以观察两张数据库表：商品库存表、商品成功抢单的用户记录表 - 只有当库存表中商品对应的库存量为 0、商品成功抢单的用户记录刚好 100 时 即表示我们的实战目的以及效果已经达到了！！

总结：如此一来，我们便将 request 转移到我们的 mq，在一定程度缓解了我们的应用以及接口的压力！当然，实际情况下，我们的配置可能远远不只代码层次上的配置，比如我们的 mq 可能会做集群配置、负载均衡、商品库存的更新可能会考虑分库分表、库存更新可能会考虑独立为库存 Dubbo 服务并通过 Rest Api 异步通信交互并独立部署等等。这些优化以及改进的目的其实无非是为了能限流、缓压、保证系统稳定、数据的一致等！而我们的 MQ，在其中可以起到不可磨灭的作用，其字如其名：“消息队列”，而队列具有 “先进先出” 的特点，故而所有进入 MQ 的消息都将 “乖巧” 的在 MQ 上排好队，先来先排队，先来先被处理消费，由此一来至少可以避免 “瞬间时刻一窝蜂的 request 涌入我们的接口” 的情况！

附注：在用 RabbitMQ 实战上述高并发抢单解决方案，其实我也在数据库层面进行了优化，即在读写存库时采用了“类似乐观锁”的写法，保证：抢单的请求到来时有库存，更新存库时保证有库存可以被更新！

彩蛋：本博文继续分享介绍了RabbitMQ典型应用业务场景的实战-并发系统下RabbitMQ的限流作用以及基于SpringBoot微服务项目的实战，另外也介绍了消息确认机制的配置实战跟并发量配置，下篇博文将继续分享死信队列的相关内容及其实战，相关源码数据库可以来这里下载

https://pan.baidu.com/s/1KUuz_eeFXOKF3XRMY2Jcew

学习过程有任何问题均可以与我交流，QQ：1974544863！感兴趣的童鞋可以关注一下我的微信公众号！

原文：http://blog.51cto.com/13877966/2297182