分享

Springboot集成rabbitmq商品秒杀业务实战(流量削峰)

问题导读:
1、springboot如何集成rabbitmq?
2、RabbitConfig配置类如何设置?
3、如何初始化rabbitmq的回调函数?
4、如何配置rabbitmq监听方法?



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

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

2021-10-20_190740.jpg

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

一,准备工作:

数据库有一张商品表,库存量是100。现在有1000个消费者准备开抢这100个库存。
t_product表维护商品编号与商品库存剩余数量。编号No123321的这种商品的库存量有100个。

2021-10-20_190817.jpg
t_product_record维护抢到商品的用户ID。理论上t_product表开抢后的 记录数量应该是100条(共有100个人抢到了商品)。

2021-10-20_190851.jpg

我们使用压力测试工具jweter对其进行并发性测试。

二,springboot开始集成rabbitmq

1,加入amqp的依赖
  1.         <dependency>
  2.             <groupId>org.springframework.amqp</groupId>
  3.             <artifactId>spring-rabbit</artifactId>
  4.         </dependency>
复制代码




2,配置application.yml配置文件
  1. spring:
  2.   datasource:
  3.     driver-class-name: com.mysql.cj.jdbc.Driver
  4.     url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC
  5.     username: root
  6.     password: 1234
  7.   rabbitmq:
  8.     host: 101.201.101.206
  9.     username: guest
  10.     password: guest
  11.     publisher-confirms: true  # 开启Rabbitmq发送消息确认机制,发送消息到队列并触发回调方法
  12.     publisher-returns: true
  13.     listener:
  14.       simple:
  15.         concurrency: 10 #消费者数量
  16.         max-concurrency: 10 #最大消费者数量
  17.         prefetch: 1 #限流(消费者每次从队列获取的消息数量)
  18.         auto-startup: true  #启动时自动启动容器
  19.         acknowledge-mode: manual #开启ACK手动确认模式
复制代码




3,RabbitConfig配置类

1)定义消息转换实例 ,转化成 JSON传输

2 )配置启用rabbitmq事务
  1. <div>package com.aaa.springredis.controller;
  2. import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
  3. import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
  4. import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
  5. import org.springframework.amqp.support.converter.MessageConverter;
  6. import org.springframework.context.annotation.Bean;
  7. import org.springframework.stereotype.Component;
  8. @Component
  9. public class RabbitConfig {
  10.     /**
  11.      * 定义消息转换实例 ,转化成 JSON传输
  12.      *
  13.      * @return Jackson2JsonMessageConverter
  14.      */
  15.     @Bean
  16.     public MessageConverter integrationEventMessageConverter() {
  17.         return new Jackson2JsonMessageConverter();
  18.     }
  19.     /**
  20.      * 配置启用rabbitmq事务
  21.      *
  22.      * @param connectionFactory connectionFactory
  23.      * @return RabbitTransactionManager
  24.      */
  25.     @Bean
  26.     public RabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory connectionFactory) {
  27.         return new RabbitTransactionManager(connectionFactory);
  28.     }
  29. }</div>
复制代码



4,初始化rabbitmq的回调函数

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

回调函数的使用前提是配置文件中开启了rabitmq消息确认机制
2021-10-20_190942.jpg

Constructor >> @Autowired >> @PostConstruct
  1.       @Autowired
  2.         RabbitTemplate rabbitTemplate;
  3.     private static final Logger LOGGER = LoggerFactory.getLogger(RabbitController.class);
  4.     @PostConstruct
  5.     private void init(){
  6.         /**
  7.          * 消息发送到交换器Exchange后触发回调。
  8.          * 使用该功能需要开启确认,spring-boot中配置如下:
  9.          * spring.rabbitmq.publisher-confirms = true
  10.          */
  11.     rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
  12.         @Override
  13.         public void confirm(CorrelationData correlationData, boolean b, String s) {
  14.             if (b) {
  15.                 System.out.println("消息已确认 cause:{"+s+"} - {"+correlationData+"}");
  16.             } else {
  17.                 System.out.println("消息未确认 cause:{"+s+"} - {"+correlationData+"}");
  18.             }
  19.         }
  20.     });
  21.         /**
  22.          * 通过实现ReturnCallback接口,
  23.          * 如果消息从交换器发送到对应队列失败时触发
  24.          * 比如根据发送消息时指定的routingKey找不到队列时会触发
  25.          * 使用该功能需要开启确认,spring-boot中配置如下:
  26.          * spring.rabbitmq.publisher-returns = true
  27.          */
  28.         rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
  29.             @Override
  30.             public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
  31.                 LOGGER.error("消息被退回:{}", message);
  32.                 LOGGER.error("消息使用的交换机:{}", exchange);
  33.                 LOGGER.error("消息使用的路由键:{}", routingKey);
  34.                 LOGGER.error("描述:{}", replyText);
  35.             }
  36.         });
  37.     }
复制代码




三,开始测试
1,写抢单测试类

写抢单测试类,我们使用jweter压力测试工具开启1000个线程进行测试(开启多线程并发测试),所以为了区别每一个模拟的用户,使用userId累加的方式进行区分。
  1.        private int userId=0;
  2.         //开始抢单
  3.         @RequestMapping("/begin")
  4.         @ResponseBody
  5.         public void begin(){
  6.            userId++;
  7.            this.send(userId);
  8.         }
  9. 而上面的send方法就是把接收到的用户请求发送到rabbitmq消息中间件中。
  10.         @RequestMapping("/send")
  11.         @ResponseBody
  12.         public String send(Integer messge){
  13.                 //第一个参数:交换机名字  第二个参数:Routing Key的值  第三个参数:传递的消息对象
  14.                 rabbitTemplate.convertAndSend("test.direct","test",messge);
  15.                 return "发送消息成功";
  16.         }
复制代码




2,配置rabbitmq监听方法

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

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

  1.    @Autowired
  2.     RabbitController controller;
  3.     /**
  4.      * @RabbitListener 可以标注在类上面,需配合 @RabbitHandler 注解一起使用
  5.      * @RabbitListener 标注在类上面表示当有收到消息的时候,就交给 @RabbitHandler 的方法处理,具体使用哪个方法处理,
  6.      * 根据 MessageConverter 转换后的参数类型
  7.      *
  8.      * 使用 @Payload 和 @Headers 注解可以消息中的 body 与 headers 信息
  9.      *
  10.      * 通过 ACK 确认是否被正确接收,每个 Message 都要被确认(acknowledged),可以手动去 ACK 或自动 ACK
  11.      */
  12.     @RabbitListener(queues = "test") //指定监听的队列名
  13.     public void receiver(@Payload Integer userId, @Headers Channel channel, Message message) throws IOException {
  14.         LOGGER.info("用户{}开始抢单", userId);
  15.         try {
  16.             //处理消息
  17.             controller.robbingProduct(userId);
  18. //             确认消息已经消费成功
  19.             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
  20.         } catch (Exception e) {
  21.             LOGGER.error("消费处理异常:{} - {}", userId, e);
  22. //             拒绝当前消息,并把消息返回原队列
  23.             channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
  24.         }
  25.     }
复制代码


购买商品的方法
  1. public void robbingProduct(Integer userId){
  2.                 Product product = testDao.selectProductByNo("No123321");
  3.                 if (product != null && product.getTotal() > 0) {
  4.                     //更新库存表,库存量减少1。返回1说明更新成功。返回0说明库存已经为0
  5.                     int i = testDao.updateProduct("No123321");
  6.                     if(i>0){
  7.                         //插入记录
  8.                         testDao.insertProductRecord(new ProductRecord("No123321", userId));
  9.                         //发送短信
  10.                         LOGGER.info("用户{}抢单成功", userId);
  11.                     }else {
  12.                         LOGGER.error("用户{}抢单失败", userId);
  13.                     }
  14.                 } else {
  15.                     LOGGER.error("用户{}抢单失败", userId);
  16.                 }
  17.         }
复制代码




3,jweter工具测试并发

jweter压力测试工具如何使用百度吧,这里忽略!
控制台打印
2021-10-20_191028.jpg

2021-10-20_191053.jpg

而数据库中的库存变成了0
2021-10-20_191123.jpg

购买记录中存放了抢单成功的用户id(100条记录)
2021-10-20_191154.jpg

当然,剩下的900个用户都抢单失败了!

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

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

作者:本少爷来了
来源:https://www.jianshu.com/p/f5e28df429b8


最新经典文章,欢迎关注公众号


已有(1)人评论

跳转到指定楼层
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /5 下一条