springboot集成rabbitmq商品秒杀业务实战（流量削峰）

客官°小女子只卖身不卖艺 2022-12-10 12:25 772阅读 0赞

转载：[springboot集成rabbitmq商品秒杀业务实战（流量削峰）][springboot_rabbitmq]

*消息队列如何实现流量削峰？*

要对流量进行削峰，最容易想到的解决方案就是用消息队列来缓冲瞬时流量，把同步的直接调用转换成异步的间接推送，中间通过一个队列在一端承接瞬时的流量洪峰，在另一端平滑地将消息推送出去。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70]

这里就不讲`springboot`和`rabbitmq`如何集成了，参考文章[https://www.cnblogs.com/fantongxue/p/12493497.html][https_www.cnblogs.com_fantongxue_p_12493497.html]

## 一，准备工作： ##

数据库有一张商品表，库存量是100。现在有1000个消费者准备开抢这100个库存。  
`t_product`表维护商品编号与商品库存剩余数量。编号No123321的这种商品的库存量有100个。  
![在这里插入图片描述][20200416143645828.png]  
`t_product_record`维护抢到商品的用户ID。理论上`t_product`表开抢后的 记录数量应该是100条(共有100个人抢到了商品)。  
![在这里插入图片描述][20200416143703771.png]  
我们使用压力测试工具`jweter`对其进行并发性测试。

## 二，springboot开始集成rabbitmq ##

### 1，加入amqp的依赖 ###

<dependency>
 <groupId>org.springframework.amqp</groupId>
 <artifactId>spring-rabbit</artifactId>
 </dependency>
 
 
 
 
 
 1
 2
 3
 4

### 2，配置application.yml配置文件 ###

spring:
      datasource:
        driver-class-name: com.mysql.cj.jdbc.Driver
        url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC
        username: root
        password: 1234

rabbitmq:  
host: 101.201.101.206  
username: guest  
password: guest  
publisher-confirms: true \# 开启Rabbitmq发送消息确认机制，发送消息到队列并触发回调方法  
publisher-returns: true  
listener:  
simple:  
concurrency: 10 \#消费者数量  
max-concurrency: 10 \#最大消费者数量  
prefetch: 1 \#限流（消费者每次从队列获取的消息数量）  
auto-startup: true \#启动时自动启动容器  
acknowledge-mode: manual \#开启ACK手动确认模式

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20

### 3，RabbitConfig配置类 ###

1， 定义消息转换实例 ，转化成 `JSON`传输  
2 ， 配置启用`rabbitmq`事务

package com.aaa.springredis.controller;

import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;  
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;  
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;  
import org.springframework.amqp.support.converter.MessageConverter;  
import org.springframework.context.annotation.Bean;  
import org.springframework.stereotype.Component;

@Component  
public class RabbitConfig \{

/**
 * 定义消息转换实例 ，转化成 JSON传输
 *
 * @return Jackson2JsonMessageConverter
 */
 @Bean
 public MessageConverter integrationEventMessageConverter() {
 return new Jackson2JsonMessageConverter();
 }
 
 /**
 * 配置启用rabbitmq事务
 *
 * @param connectionFactory connectionFactory
 * @return RabbitTransactionManager
 */
 @Bean
 public RabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory connectionFactory) {
 return new RabbitTransactionManager(connectionFactory);
 }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22
 *  23
 *  24
 *  25
 *  26
 *  27
 *  28
 *  29
 *  30
 *  31
 *  32
 *  33

### 4，初始化rabbitmq的回调函数 ###

说明：被`@PostConstruct`修饰的方法会在服务器加载`Servlet`的时候运行，并且只会被服务器执行一次。如果想在生成对象时完成某些初始化操作，而偏偏这些初始化操作又依赖于依赖注入，那么久无法在构造函数中实现。为此，可以使用`@PostConstruct`注解一个方法来完成初始化，`@PostConstruct`注解的方法将会在依赖注入完成后被自动调用。

回调函数的使用前提是配置文件中开启了`rabitmq`消息确认机制  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 1]

*Constructor >> @Autowired >> @PostConstruct*

@Autowired
 RabbitTemplate rabbitTemplate;
 private static final Logger LOGGER = LoggerFactory.getLogger(RabbitController.class);
 
 
 @PostConstruct
 private void init(){
 /**
 * 消息发送到交换器Exchange后触发回调。
 * 使用该功能需要开启确认，spring-boot中配置如下：
 * spring.rabbitmq.publisher-confirms = true
 */
 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
 @Override
 public void confirm(CorrelationData correlationData, boolean b, String s) {
 if (b) {
 System.out.println("消息已确认 cause:{"+s+"} - {"+correlationData+"}");
 } else {
 System.out.println("消息未确认 cause:{"+s+"} - {"+correlationData+"}");
 }
 }
 });
 /**
 * 通过实现ReturnCallback接口，
 * 如果消息从交换器发送到对应队列失败时触发
 * 比如根据发送消息时指定的routingKey找不到队列时会触发
 * 使用该功能需要开启确认，spring-boot中配置如下：
 * spring.rabbitmq.publisher-returns = true
 */
 rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
 @Override
 public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
 LOGGER.error("消息被退回:{}", message);
 LOGGER.error("消息使用的交换机:{}", exchange);
 LOGGER.error("消息使用的路由键:{}", routingKey);
 LOGGER.error("描述:{}", replyText);
 }
 });
 }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22
 *  23
 *  24
 *  25
 *  26
 *  27
 *  28
 *  29
 *  30
 *  31
 *  32
 *  33
 *  34
 *  35
 *  36
 *  37
 *  38
 *  39
 *  40

## 三，开始测试 ##

### 1，写抢单测试类 ###

写抢单测试类，我们使用`jweter`压力测试工具开启1000个线程进行测试（开启多线程并发测试），所以为了区别每一个模拟的用户，使用`userId`累加的方式进行区分。

private int userId=0;
            //开始抢单
            @RequestMapping("/begin")
            @ResponseBody
            public void begin(){ 
               userId++;
               this.send(userId);
            }
    
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8

而上面的`send`方法就是把接收到的用户请求发送到`rabbitmq`消息中间件中。

@RequestMapping("/send")
            @ResponseBody
            public String send(Integer messge){ 
                    //第一个参数：交换机名字 第二个参数：Routing Key的值 第三个参数：传递的消息对象
                    rabbitTemplate.convertAndSend("test.direct","test",messge);
                    return "发送消息成功";
            }
    
     
      
     
      
       1
       2
       3
       4
       5
       6
       7

### 2，配置rabbitmq监听方法 ###

`rabbitmq`监听上篇文章也说过了，作用就是监听指定队列中收到来自交换机的消息，来一条收一条，收完为止！  
通过`ACK` 确认是否被正确接收，每个`Message`都要被确认，可以手动去 `ACK`或自动`ACK`，如果信息消费失败的话会拒绝当前消息，并把消息返回原队列。

从队列中收到用户的`userId`，然后进行购买商品模拟操作（减少一个库存，新增一条购买记录）

@Autowired
 RabbitController controller;
 /**
 * @RabbitListener 可以标注在类上面，需配合 @RabbitHandler 注解一起使用
 * @RabbitListener 标注在类上面表示当有收到消息的时候，就交给 @RabbitHandler 的方法处理，具体使用哪个方法处理，
 * 根据 MessageConverter 转换后的参数类型
 *
 * 使用 @Payload 和 @Headers 注解可以消息中的 body 与 headers 信息
 *
 * 通过 ACK 确认是否被正确接收，每个 Message 都要被确认（acknowledged），可以手动去 ACK 或自动 ACK
 */
 @RabbitListener(queues = "test") //指定监听的队列名
 public void receiver(@Payload Integer userId, @Headers Channel channel, Message message) throws IOException {
 LOGGER.info("用户{}开始抢单", userId);
 try {
 //处理消息
 controller.robbingProduct(userId);

// 确认消息已经消费成功  
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);  
\} catch (Exception e) \{   
LOGGER.error(“消费处理异常:\{\} - \{\}”, userId, e);  
// 拒绝当前消息，并把消息返回原队列  
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);  
\}  
\}

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18
 *  19
 *  20
 *  21
 *  22
 *  23
 *  24
 *  25
 *  26

购买商品的方法

public void robbingProduct(Integer userId){ 
 Product product = testDao.selectProductByNo("No123321");
 if (product != null && product.getTotal() > 0) { 
 //更新库存表，库存量减少1。返回1说明更新成功。返回0说明库存已经为0
 int i = testDao.updateProduct("No123321");
 if(i>0){ 
 //插入记录
 testDao.insertProductRecord(new ProductRecord("No123321", userId));
 //发送短信
 LOGGER.info("用户{}抢单成功", userId);
 }else { 
 LOGGER.error("用户{}抢单失败", userId);
 }
 } else { 
 LOGGER.error("用户{}抢单失败", userId);
 }
 }

*  1
 *  2
 *  3
 *  4
 *  5
 *  6
 *  7
 *  8
 *  9
 *  10
 *  11
 *  12
 *  13
 *  14
 *  15
 *  16
 *  17
 *  18

### 3，jweter工具测试并发 ###

`jweter`压力测试工具如何使用百度吧，这里忽略！  
控制台打印  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 3]  
而数据库中的库存变成了0  
![在这里插入图片描述][20200416152606174.png]  
购买记录中存放了抢单成功的用户`id`（100条记录）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 4]  
当然，剩下的900个用户都抢单失败了！

`rabbitmq`队列是先进先出的顺序，先来后到，1000个请求你也得给我排队，前100个请求抢单成功之后就注定了后900个请求是抢单失败的！

使用`RabbitMQ`的最主要变化就是：以前抢单操作请求直接由我们抢单应用程序执行，现在请求被转移到了`RabbitMQ`服务器中。`RabbitMQ`服务器把接收到的抢单请求进行排队，最后由`RabbitMQ`服务器把抢单请求转发到我们的抢单应用程序，这样的好处就是避免我们的抢单应用程序短时间直接处理大量请求。`RabbitMQ`服务器主要作用是减缓抢单应用程序的并发压力，相当于在我们的抢单程序之前加了一道请求缓冲区。

实战结束！  
工程地址：[https://github.com/fantongxue666/rabbitmq-seckill][https_github.com_fantongxue666_rabbitmq-seckill]

[springboot_rabbitmq]: https://blog.csdn.net/weixin_44001965/article/details/105557610?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522160108622019725255545641%2522%252C%2522scm%2522%253A%252220140713.130102334..%2522%257D&request_id=160108622019725255545641&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~first_rank_v2~rank_v28-3-105557610.pc_first_rank_v2_rank_v28&utm_term=rabbitmq%20%E5%89%8A%E5%B3%B0%E9%99%90%E6%B5%81&spm=1018.2118.3001.4187
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70]: /images/20221123/d235266b7ccd4fd3b42206ef6a51fe23.png
[https_www.cnblogs.com_fantongxue_p_12493497.html]: https://www.cnblogs.com/fantongxue/p/12493497.html
[20200416143645828.png]: /images/20221123/c0a3257b159b4bb191f39552bb3272d7.png
[20200416143703771.png]: /images/20221123/5ab7f0aeacbd4fd7a8fcc46674ce8876.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 1]: /images/20221123/a6e42da881f744b6b456e03729fdd05c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 2]: /images/20221123/2f5cbb9948a8481eb332a38d0834b4b0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 3]: /images/20221123/1ef84a4ca6bf40af972fe7af694eeb83.png
[20200416152606174.png]: /images/20221123/c707305cf68a424988b94e46d0e95431.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDAwMTk2NQ_size_16_color_FFFFFF_t_70 4]: /images/20221123/1686408fd5d64fab91bbbaa2602e560d.png
[https_github.com_fantongxue666_rabbitmq-seckill]: https://github.com/fantongxue666/rabbitmq-seckill