（Java企业 / 公司项目）使用RocketMQ组件对请求做削峰处理

我会带着你远行 2024-04-29 19:07 11阅读 0赞

### 内容 ###

*  rocketMQ基本介绍
 *  使用MQ，将购票流程一分为二。目前系统的吞吐量低，用户从购买车票到拿到票花费的时间较长。
 *  增加排队购票功能。排队提示loading。

### **什么是RocketMQ** ###

RocketMQ作为一款纯java、分布式、队列模型的开源消息中间件，支持事务消息、顺序消息、批量消息、定时消息、消息回溯等。主要功能是异步解耦和流量削峰：。

### 购票时序图 ###

目前的时序图，用户发送购票请求，服务端校验验证码，拿令牌，拿锁，然后选座购票，结束流程才会返回。服务器执行时间太长。

![6b7b1a29308b44b18686b9da745470ba.png][]

增加异步，拿锁分为两步，拿令牌锁要放在同步里，拿车次锁要放在异步里。用来防止机器人和超卖。拿到后便响应给用户，告诉用户有资格买票。异步线程中选座购票。之后前端发起轮询，调用后端的查询接口。就想下面这个图片一样的操作

![8207dffb11684f0bbb12fc157ef2b7bd.png][]

再次改进，将异步操作放到出票模块，接收购票请求的模块与购票模块分开，可以用不同的服务器，从而为选座购票功能分配更多的节点。服务端发送消息给MQ，出票模块监听MQ的消息，有购票请求就选座购票。过一段时间进行轮询，查看购票结果。

![cf6b66535a484f8aa8d3eb11870e543f.png][]

### 初始RocketMQ ###

[ https://rocketmq.apache.org/][https_rocketmq.apache.org]

[快速开始][Link 1]

#### [相关概念][Link 2] ####

*  生产者：负责生产消息，一般由业务系统负责生产消息。一个消息生产者会把业务应用系统里产生的消息发送到broker服务器。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。
    
     *  topic，表示要发送的消息的主题。
     *  body 表示消息的存储内容
     *  properties 表示消息属性
     *  transactionId 会在事务消息中使用。
     *  普通消息、顺序消息、延迟消息、批量消息、事务消息。
 *  消费者：负责消费消息，一般是后台系统负责异步消费。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。
    
     *  push消费：MQ主动将消息推给客户端。
     *  pull消费：消费者主动拉取消息。
     *  一个消息可以支持多个消费者or一个消息由一个消费者消费。
 *  主题：表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是RocketMQ进行消息订阅的基本单位。

### 使用RocketMQ将购票流程一分为二 ###

一部分处理验证码，令牌，车次锁；另一部分处理选座购票逻辑。

拿到车次锁，就代表用户有条件购票，然后快速反馈用户。

**下单购票接口，只处理验证码、令**牌、车次锁，不执行选座购票逻辑

package com.jiawa.train.business.service;
    
    import cn.hutool.core.date.DateTime;
    import com.alibaba.csp.sentinel.annotation.SentinelResource;
    import com.alibaba.csp.sentinel.slots.block.BlockException;
    import com.alibaba.fastjson.JSON;
    import com.jiawa.train.business.domain.ConfirmOrder;
    import com.jiawa.train.business.dto.ConfirmOrderMQDto;
    import com.jiawa.train.business.enums.ConfirmOrderStatusEnum;
    import com.jiawa.train.business.mapper.ConfirmOrderMapper;
    import com.jiawa.train.business.req.ConfirmOrderDoReq;
    import com.jiawa.train.business.req.ConfirmOrderTicketReq;
    import com.jiawa.train.common.context.LoginMemberContext;
    import com.jiawa.train.common.exception.BusinessException;
    import com.jiawa.train.common.exception.BusinessExceptionEnum;
    import com.jiawa.train.common.util.SnowUtil;
    import jakarta.annotation.Resource;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.slf4j.MDC;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import java.util.Date;
    import java.util.List;
    
    @Service
    public class BeforeConfirmOrderService {
    
        private static final Logger LOG = LoggerFactory.getLogger(BeforeConfirmOrderService.class);
    
        @Resource
        private ConfirmOrderMapper confirmOrderMapper;
    
        @Autowired
        private SkTokenService skTokenService;
    
        // @Resource
        // public RocketMQTemplate rocket
        // public RocketMQTemplate rocketMQTemplate;
    
        @Resource
        private ConfirmOrderService confirmOrderService;
    
        @SentinelResource(value = "beforeDoConfirm", blockHandler = "beforeDoConfirmBlock")
        public Long beforeDoConfirm(ConfirmOrderDoReq req) {
            Long id = null;
            // 根据前端传值，加入排队人数
            for (int i = 0; i < req.getLineNumber() + 1; i++) {
                req.setMemberId(LoginMemberContext.getId());
                // 校验令牌余量
                boolean validSkToken = skTokenService.validSkToken(req.getDate(), req.getTrainCode(), LoginMemberContext.getId());
                if (validSkToken) {
                    LOG.info("令牌校验通过");
                } else {
                    LOG.info("令牌校验不通过");
                    throw new BusinessException(BusinessExceptionEnum.CONFIRM_ORDER_SK_TOKEN_FAIL);
                }
    
                Date date = req.getDate();
                String trainCode = req.getTrainCode();
                String start = req.getStart();
                String end = req.getEnd();
                List<ConfirmOrderTicketReq> tickets = req.getTickets();
    
                // 保存确认订单表，状态初始
                DateTime now = DateTime.now();
                ConfirmOrder confirmOrder = new ConfirmOrder();
                confirmOrder.setId(SnowUtil.getSnowflakeNextId());
                confirmOrder.setCreateTime(now);
                confirmOrder.setUpdateTime(now);
                confirmOrder.setMemberId(req.getMemberId());
                confirmOrder.setDate(date);
                confirmOrder.setTrainCode(trainCode);
                confirmOrder.setStart(start);
                confirmOrder.setEnd(end);
                confirmOrder.setDailyTrainTicketId(req.getDailyTrainTicketId());
                confirmOrder.setStatus(ConfirmOrderStatusEnum.INIT.getCode());
                confirmOrder.setTickets(JSON.toJSONString(tickets));
                confirmOrderMapper.insert(confirmOrder);
    
                // 发送MQ排队购票
                ConfirmOrderMQDto confirmOrderMQDto = new ConfirmOrderMQDto();
                confirmOrderMQDto.setDate(req.getDate());
                confirmOrderMQDto.setTrainCode(req.getTrainCode());
                confirmOrderMQDto.setLogId(MDC.get("LOG_ID"));
                String reqJson = JSON.toJSONString(confirmOrderMQDto);
                // LOG.info("排队购票，发送mq开始，消息：{}", reqJson);
                // rocketMQTemplate.convertAndSend(RocketMQTopicEnum.CONFIRM_ORDER.getCode(), reqJson);
                // LOG.info("排队购票，发送mq结束");
                confirmOrderService.doConfirm(confirmOrderMQDto);
                id = confirmOrder.getId();
            }
            return id;
        }
    
        /**
         * 降级方法，需包含限流方法的所有参数和BlockException参数
         * @param req
         * @param e
         */
        public void beforeDoConfirmBlock(ConfirmOrderDoReq req, BlockException e) {
            LOG.info("购票请求被限流：{}", req);
            throw new BusinessException(BusinessExceptionEnum.CONFIRM_ORDER_FLOW_EXCEPTION);
        }
    }

ConfirmOrderController.java中注入的ConfirmOrderService也改为BeforeConfirmOrderService

1.  **依赖注入**：
    
     *  使用 `@Resource` 或 `@Autowired` 注解，将其他服务或组件注入到当前服务中，以便在方法中使用。这些服务包括 `ConfirmOrderMapper`、`DailyTrainTicketService`、`DailyTrainCarriageService`、`DailyTrainSeatService`、`AfterConfirmOrderService`、`StringRedisTemplate` 和 `SkTokenService`。
2.  **令牌校验**：.validSkToken
    
    \*`在 方法`对before传Do入的Confirm请求`（ 方法`中req，`首先）通过中的`令牌sk（`TokenskServiceToken`）进行校验。
    
     *  如果令牌有效，则记录日志表示令牌校验通过；否则，抛出 `BusinessException` 异常，表示令牌校验失败。
3.  **获取车次锁**：
    
     *  使用 Redis 的 `setIfAbsent` 命令（通过 `redisTemplate` 实现）尝试获取一个与指定车次和日期相关的锁。
     *  如果成功获取到锁（即 `setIfAbsent` 返回 `true`），则记录日志表示成功抢到锁；否则，抛出 `BusinessException` 异常，表示获取锁失败。
4.  **发送消息队列（MQ）**：
    
     *  在获取到锁之后，代码注释中提到将发送消息到消息队列（MQ）以等待出票。但具体的发送 MQ 的代码没有展示。
     *  这意味着一旦消息被发送到 MQ，相关的出票服务将会异步处理购票请求。

此外，代码中还使用了 `@SentinelResource` 注解，这是来自 Sentinel 框架的注解，用于实现服务的熔断和降级。当服务调用出现异常或响应时间超过阈值时，Sentinel 可以自动进行熔断，避免服务雪崩。在这个例子中，如果 `beforeDoConfirm` 方法执行出现异常，将会调用 `beforeDoConfirmBlock` 方法进行降级处理。

总结来说，`BeforeConfirmOrderService` 的主要功能是进行购票前的准备工作，包括令牌校验、获取车次锁以及发送消息到消息队列以等待出票。这个服务使用了 Redis 和消息队列来实现分布式环境下的并发控制和异步处理。

**实现RocketMQ发送，spring.factories功能在Spring Boot 3.0被移除，替代方案为META-INFO/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports**

添加RocketMQ依赖

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.3</version>
    </dependency>

配置rocketmq

# rocketmq
    rocketmq:
      name-server: http://localhost:9876
      producer:
        group: default

主体枚举类

public enum RocketMQTopicEnum {
        CONFIRM_ORDER("CONFIRM_ORDER", "确认订单排队");
    
        private final String code;
        private final String desc;
    
        RocketMQTopicEnum(String code, String desc) {
            this.code = code;
            this.desc = desc;
        }
    
        @Override
        public String toString() {
            return "RocketMQTopicEnum{" +
                   "code='" + code + '\'' +
                   ", desc='" + desc + '\'' +
                   '}';
        }
    
        public String getCode() {
            return code;
        }
    
        public String getDesc() {
            return desc;
        }
    }

发送rocketmq (RocketMQTemplate)，将购票的请求参数转成json，导入工具类，然后发送。

此时可以发送rocketmq

**实现rocketmq接收**

消费类，消费发送的topic，接收收到的json

[6b7b1a29308b44b18686b9da745470ba.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/29/989facf79e8c4ac5a164784a52f8901e.png
[8207dffb11684f0bbb12fc157ef2b7bd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/29/9c6e7f237c984d58b79b9b21196189ba.png
[cf6b66535a484f8aa8d3eb11870e543f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/29/57748248757f419b915ef177ebf3d708.png
[https_rocketmq.apache.org]: https://rocketmq.apache.org/
[Link 1]: https://rocketmq.apache.org/zh/docs/4.x/introduction/02quickstart
[Link 2]: https://rocketmq.apache.org/zh/docs/4.x/introduction/03whatis