pringboot集成rabbitmq商品秒杀业务实战

2021-11-21 14:31:48 浏览数 (1)

#springboot集成rabbitmq商品秒杀业务实战（流量削峰）消息队列如何实现流量削峰？

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

在这里插入图片描述

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

一，准备工作：

数据库有一张商品表，库存量是100。现在有1000个消费者准备开抢这100个库存。

t_product表维护商品编号与商品库存剩余数量。编号No123321的这种商品的库存量有100个。

在这里插入图片描述

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

在这里插入图片描述

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

二，springboot开始集成rabbitmq

1，加入amqp的依赖

代码语言：txt复制
        <dependency>
代码语言：txt复制
            <groupId>org.springframework.amqp</groupId>
代码语言：txt复制
            <artifactId>spring-rabbit</artifactId>
代码语言：txt复制
        </dependency>

2，配置application.yml配置文件

代码语言：txt复制
spring:
代码语言：txt复制
  datasource:
代码语言：txt复制
    driver-class-name: com.mysql.cj.jdbc.Driver
代码语言：txt复制
    url: jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=true&serverTimezone=UTC
代码语言：txt复制
    username: root
代码语言：txt复制
    password: 1234
代码语言：txt复制
  rabbitmq:
代码语言：txt复制
    host: 101.201.101.206
代码语言：txt复制
    username: guest
代码语言：txt复制
    password: guest
代码语言：txt复制
    publisher-confirms: true  # 开启Rabbitmq发送消息确认机制，发送消息到队列并触发回调方法
代码语言：txt复制
    publisher-returns: true
代码语言：txt复制
    listener:
代码语言：txt复制
      simple:
代码语言：txt复制
        concurrency: 10 #消费者数量
代码语言：txt复制
        max-concurrency: 10 #最大消费者数量
代码语言：txt复制
        prefetch: 1 #限流（消费者每次从队列获取的消息数量）
代码语言：txt复制
        auto-startup: true  #启动时自动启动容器
代码语言：txt复制
        acknowledge-mode: manual #开启ACK手动确认模式

3，RabbitConfig配置类

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

2 ， 配置启用rabbitmq事务

代码语言：txt复制
package com.aaa.springredis.controller;
代码语言：txt复制
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
代码语言：txt复制
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
代码语言：txt复制
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
代码语言：txt复制
import org.springframework.amqp.support.converter.MessageConverter;
代码语言：txt复制
import org.springframework.context.annotation.Bean;
代码语言：txt复制
import org.springframework.stereotype.Component;
代码语言：txt复制
@Component
代码语言：txt复制
public class RabbitConfig {
代码语言：txt复制
    /**
代码语言：txt复制
     * 定义消息转换实例 ，转化成 JSON传输
     *
     * @return Jackson2JsonMessageConverter
     */
    @Bean
    public MessageConverter integrationEventMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }
代码语言：txt复制
    /**
代码语言：txt复制
     * 配置启用rabbitmq事务
     *
     * @param connectionFactory connectionFactory
     * @return RabbitTransactionManager
     */
    @Bean
    public RabbitTransactionManager rabbitTransactionManager(CachingConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }
}

4，初始化rabbitmq的回调函数

<font

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

回调函数的使用前提是配置文件中开启了rabitmq消息确认机制

在这里插入图片描述

Constructor >> @Autowired >> @PostConstruct

代码语言：txt复制
      @Autowired
代码语言：txt复制
        RabbitTemplate rabbitTemplate;
代码语言：txt复制
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitController.class);
代码语言：txt复制
    @PostConstruct
代码语言：txt复制
    private void init(){
代码语言：txt复制
        /**
代码语言：txt复制
         * 消息发送到交换器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，写抢单测试类

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

代码语言：txt复制
        private int userId=0;
代码语言：txt复制
        //开始抢单
代码语言：txt复制
        @RequestMapping("/begin")
代码语言：txt复制
        @ResponseBody
代码语言：txt复制
        public void begin(){
代码语言：txt复制
           userId  ;
代码语言：txt复制
           this.send(userId);
代码语言：txt复制
        }

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

代码语言：txt复制
        @RequestMapping("/send")
代码语言：txt复制
        @ResponseBody
代码语言：txt复制
        public String send(Integer messge){
代码语言：txt复制
                //第一个参数：交换机名字  第二个参数：Routing Key的值  第三个参数：传递的消息对象
代码语言：txt复制
                rabbitTemplate.convertAndSend("test.direct","test",messge);
代码语言：txt复制
                return "发送消息成功";
代码语言：txt复制
        }

2，配置rabbitmq监听方法

rabbitmq监听上篇文章也说过了，作用就是监听指定队列中收到来自交换机的消息，来一条收一条，收完为止！

通过ACK 确认是否被正确接收，每个Message都要被确认，可以手动去

ACK或自动ACK，如果信息消费失败的话会拒绝当前消息，并把消息返回原队列。

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

代码语言：txt复制
    @Autowired
代码语言：txt复制
    RabbitController controller;
代码语言：txt复制
    /**
代码语言：txt复制
     * @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);
        }
    }

购买商品的方法

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

3，jweter工具测试并发

jweter压力测试工具如何使用百度吧，这里忽略！

控制台打印

在这里插入图片描述

在这里插入图片描述

而数据库中的库存变成了0

在这里插入图片描述

购买记录中存放了抢单成功的用户id（100条记录）

在这里插入图片描述

当然，剩下的900个用户都抢单失败了！

<font

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

<font

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

实战结束！

seckill](https://links.jianshu.com/go?to=https://github.com/fantongxue666/rabbitmq-

seckill)

rabbitmq 腾讯云测试服务 压力测试 消息队列CMQ版 kafka

0 人点赞