Springboot快速集成RocketMQ

╰+哭是因爲堅強的太久メ 2024-03-29 16:14 80阅读 0赞

##  ##

**目录**

一、RocketMQ？是个啥

官方说明:

具有以下特性：

二、RocketMQ消息队列的特点、优势

RocketMQ的特点和优势（可跳过看三的整合代码）

编辑 三、RocketMQ环境安装

SpringBoot环境中使用RocketMQ

四、RocketMQ配置示例

4.1 依赖配置

4.2 配置文件

4.3.初始化生产者配置类

4.4 MQ消费者配置

4.5 初始化消费者配置类

4.6 监听类处理消息类配置

4.7.测试消息发送接收

设置延时发送消息

定时发送消息

4.8.rocketmq控制台查看

--------------------

## 一、[RocketMQ][]？是个啥 ##

#### `官方说明:` ####

`随着使用越来越多的队列和虚拟主题，ActiveMQ IO模块遇到了瓶颈。我们尽力通过节流，断路器或降级来解决此问题，但效果不佳。因此，我们那时开始关注流行的消息传递解决方案Kafka。不幸的是，Kafka不能满足我们的要求，特别是在低延迟和高可靠性方面。`

`看到这里可以很清楚的知道RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。`

#### 具有以下特性： ####

> 1.  支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型
> 2.  能够保证严格的消息顺序，在一个队列中可靠的先进先出（FIFO）和严格的顺序传递
> 3.  提供丰富的消息拉取模式，支持拉（pull）和推（push）两种消息模式
> 4.  单一队列百万消息的堆积能力，亿级消息堆积能力
> 5.  支持多种消息协议，如 JMS、MQTT 等
> 6.  分布式高可用的部署架构,满足至少一次消息传递语义

## **二、RocketMQ消息队列的特点、优势** ##

*  首先RocketMQ是阿里巴巴自研出来的，也已开源。其性能和稳定性从双11就能看出来，借用阿里的一句官方介绍：历年双 11 购物狂欢节零点千万级 TPS、万亿级数据洪峰，创造了全球最大的业务消息并发以及流转纪录（日志类消息除外）；
 *  在始终保证高性能前提下，支持亿级消息堆积，不影响集群的正常服务，在削峰填谷（蓄洪）、微服务解耦的场景下尤为重要；这，就能说明RocketMQ的强大。

#### **RocketMQ的特点和优势（可跳过看三的整合代码）** ####

1、削峰填谷（主要解决诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲，或因没做相应的保护而导致系统超负荷甚至崩溃，或因限制太过导致请求大量失败而影响用户体验，海量消息堆积能力强）

![format_png][]

2、异步解耦（高可用松耦合架构设计，对高依赖的项目之间进行解耦，当下游系统出现宕机，不会影响上游系统的正常运行，或者雪崩）

![ce5442a365ece401ee83c0081b4858ee.png][]

![format_png 1][]

3、顺序消息（顺序消息即保证消息的先进先出，比如证券交易过程时间优先原则，交易系统中的订单创建、支付、退款等流程，航班中的旅客登机消息处理等）

![format_png 2][]

4、分布式事务消息（确保数据的最终一致性，大量引入 MQ 的分布式事务，既可以实现系统之间的解耦，又可以保证最终的数据一致性，减少系统间的交互）

## ![format_png 3][] 三、RocketMQ环境安装 ##

参考文档 [windows本地安装部署RocketMQ][windows_RocketMQ]

#### SpringBoot环境中使用RocketMQ ####

项目基于之前搭建 [SpringCloud搭建Nacos项目][SpringCloud_Nacos] 增加RocketMQ功能，

项目搭建参考：

[springcloud集成nacos的配置中心，注册中心\_lockie的博客-CSDN博客][springcloud_nacos_lockie_-CSDN]

## 四、RocketMQ配置示例 ##

### 4.1 依赖配置 ###

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.7.0</version>
    </dependency>

### 4.2 配置文件 ###

spring.application.name=boot-order-service
    server.port=8802
     
    # nacos配置地址
    nacos.config.server-addr=127.0.0.1:8848
    # nacos注册地址
    nacos.discovery.server-addr=127.0.0.1:8848
     
    spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
    spring.jackson.time-zone=GMT+8
     
    # 是否开启自动配置
    rocketmq.producer.isOnOff=on
    # 发送同一类消息设置为同一个group，保证唯一默认不需要设置，rocketmq会使用ip@pid（pid代表jvm名字）作为唯一标识
    rocketmq.producer.groupName=${spring.application.name}
    # mq的nameserver地址
    rocketmq.producer.namesrvAddr=127.0.0.1:9876
    # 消息最大长度 默认 1024 * 4 (4M)
    rocketmq.producer.maxMessageSize = 4096
    # 发送消息超时时间，默认 3000
    rocketmq.producer.sendMsgTimeOut=3000
    # 发送消息失败重试次数，默认2
    rocketmq.producer.retryTimesWhenSendFailed=2

#### 4.3.初始化生产者配置类 ####

package com.cloudorder.rocketmq;
     
    import lombok.Getter;
    import lombok.Setter;
    import lombok.ToString;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     
    /**
     * @author: htl
     * @Date: 2023/1/17 10:28
     * @Description: mq生产者配置
     */
    @Getter
    @Setter
    @ToString
    @Configuration
    @ConfigurationProperties(prefix = "rocketmq.producer")
    public class MQProducerConfigure {
        public static final Logger LOGGER = LoggerFactory.getLogger(MQProducerConfigure.class);
     
        private String groupName;
        private String namesrvAddr;
        // 消息最大值
        private Integer maxMessageSize;
        // 消息发送超时时间
        private Integer sendMsgTimeOut;
        // 发送失败重试次数
        private Integer retryTimesWhenSendFailed;
     
        /**
         * mq 生成者配置
         * @return
         * @throws MQClientException
         */
        @Bean
        @ConditionalOnProperty(prefix = "rocketmq.producer", value = "isOnOff", havingValue = "on")
        public DefaultMQProducer defaultProducer() throws MQClientException {
            LOGGER.info("defaultProducer 正在创建---------------------------------------");
            DefaultMQProducer producer = new DefaultMQProducer(groupName);
            producer.setNamesrvAddr(namesrvAddr);
            producer.setVipChannelEnabled(false);
            producer.setMaxMessageSize(maxMessageSize);
            producer.setSendMsgTimeout(sendMsgTimeOut);
            producer.setRetryTimesWhenSendAsyncFailed(retryTimesWhenSendFailed);
            producer.start();
            LOGGER.info("rocketmq producer server 开启成功----------------------------------");
            return producer;
        }
    }

### 4.4 MQ消费者配置 ###

spring.application.name=boot-user-service
    server.port=8801
     
    # nacos配置地址
    nacos.config.server-addr=127.0.0.1:8848
    # nacos注册地址
    nacos.discovery.server-addr=127.0.0.1:8848
     
    spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
    spring.jackson.time-zone=GMT+8
     
    # 是否开启自动配置
    rocketmq.consumer.isOnOff=on
    # 发送同一类消息设置为同一个group，保证唯一默认不需要设置，rocketmq会使用ip@pid（pid代表jvm名字）作为唯一标识
    rocketmq.consumer.groupName=${spring.application.name}
    # mq的nameserver地址
    rocketmq.consumer.namesrvAddr=127.0.0.1:9876
    # 消费者订阅的主题topic和tags（*标识订阅该主题下所有的tags），格式: topic~tag1||tag2||tags3;
    rocketmq.consumer.topics=TestTopic~TestTag;TestTopic~HelloTag;HelloTopic~HelloTag;MyTopic~*
    # 消费者线程数据量
    rocketmq.consumer.consumeThreadMin=5
    rocketmq.consumer.consumeThreadMax=32
    # 设置一次消费信心的条数，默认1
    rocketmq.consumer.consumeMessageBatchMaxSize=1

#### 4.5 初始化消费者配置类 ####

package com.bootuser.rocketmq;
     
    import lombok.Getter;
    import lombok.Setter;
    import lombok.ToString;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
     
    /**
     * @author: htl
     * @Date: 2023/1/17 10:28
     * @Description: mq消费者配置
     */
    @Getter
    @Setter
    @ToString
    @Configuration
    @ConfigurationProperties(prefix = "rocketmq.consumer")
    public class MQConsumerConfigure {
        public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfigure.class);
     
        private String groupName;
        private String namesrvAddr;
        private String topics;
        // 消费者线程数据量
        private Integer consumeThreadMin;
        private Integer consumeThreadMax;
        private Integer consumeMessageBatchMaxSize;
     
        @Autowired
        private MQConsumeMsgListenerProcessor consumeMsgListenerProcessor;
        /**
         * mq 消费者配置
         * @return
         * @throws MQClientException
         */
        @Bean
        @ConditionalOnProperty(prefix = "rocketmq.consumer", value = "isOnOff", havingValue = "on")
        public DefaultMQPushConsumer defaultConsumer() throws MQClientException {
            LOGGER.info("defaultConsumer 正在创建---------------------------------------");
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
            consumer.setNamesrvAddr(namesrvAddr);
            consumer.setConsumeThreadMin(consumeThreadMin);
            consumer.setConsumeThreadMax(consumeThreadMax);
            consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
            // 设置监听
            consumer.registerMessageListener(consumeMsgListenerProcessor);
     
            /**
             * 设置consumer第一次启动是从队列头部开始还是队列尾部开始
             * 如果不是第一次启动，那么按照上次消费的位置继续消费
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            /**
             * 设置消费模型，集群还是广播，默认为集群
             */
    //        consumer.setMessageModel(MessageModel.CLUSTERING);
     
            try {
                // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*,
                String[] topicArr = topics.split(";");
                for (String topic : topicArr) {
                    String[] tagArr = topic.split("~");
                    consumer.subscribe(tagArr[0], tagArr[1]);
                }
                consumer.start();
                LOGGER.info("consumer 创建成功 groupName={}, topics={}, namesrvAddr={}",groupName,topics,namesrvAddr);
            } catch (MQClientException e) {
                LOGGER.error("consumer 创建失败!");
            }
            return consumer;
        }
    }

这个只是初始化操作，实际对消费者对消息处理放在 consumer.registerMessageListener(consumeMsgListenerProcessor); 这个监听类里面了，实际接收消息，处理消息都放在监听类里

Apache RocketMQ 支持 PushConsumer 、 SimpleConsumer 以及 PullConsumer 这三种类型的消费者

### 4.6 监听类处理消息类配置 ###

package com.bootuser.rocketmq;
     
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Component;
    import org.springframework.util.CollectionUtils;
     
    import java.util.List;
     
    /**
     * @author: htl
     * @Date: 2023/1/17 11:05
     * @Description: 消费者监听
     */
    @Component
    public class MQConsumeMsgListenerProcessor implements MessageListenerConcurrently {
        public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumeMsgListenerProcessor.class);
     
     
        /**
         * 默认msg里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
         * 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
         * @param msgList
         * @param consumeConcurrentlyContext
         * @return
         */
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
            if (CollectionUtils.isEmpty(msgList)) {
                LOGGER.info("MQ接收消息为空，直接返回成功");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            MessageExt messageExt = msgList.get(0);
            LOGGER.info("MQ接收到的消息为：" + messageExt.toString());
            try {
                String topic = messageExt.getTopic();
                String tags = messageExt.getTags();
                String body = new String(messageExt.getBody(), "utf-8");
     
                LOGGER.info("MQ消息topic={}, tags={}, 消息内容={}", topic,tags,body);
            } catch (Exception e) {
                LOGGER.error("获取MQ消息内容异常{}",e);
            }
            // TODO 处理业务逻辑
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

### 4.7.测试消息发送接收 ###

生产者 boot-order-service 新建一个controller，再新建一个send方法，发送消息

#### 设置延时发送消息 ####

#### 定时发送消息 ####

package com.cloudorder.rocketmq;
     
    import com.lockie.cloudorder.model.Results;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.client.exception.MQBrokerException;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.exception.RemotingException;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
     
    /**
     * @author: htl
     * @Date: 2023/1/17 11:17
     * @Description:
     */
    @RestController
    @RequestMapping("/mqProducer")
    public class MQProducerController {
        public static final Logger LOGGER = LoggerFactory.getLogger(MQProducerController.class);
     
        @Autowired
        DefaultMQProducer defaultMQProducer;
     
        /**
         * 发送简单的MQ消息
         * @param msg
         * @return
         */
        @GetMapping("/send")
        public Results send(String msg) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
            if (StringUtils.isEmpty(msg)) {
                return new Results().succeed();
            }
            LOGGER.info("发送MQ消息内容：" + msg);
            Message sendMsg = new Message("TestTopic", "TestTag", msg.getBytes());
            
            // 延时级别 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
            // 设置消息延迟级别为3，也就是延迟10s。
            msg.setDelayTimeLevel(3);
            
            // 定时发送消息，5秒之后发送
            msg.setDeliveryTimestamp(System.currentTimeMillis() + Duration.ofSeconds(5).toMillis());
            
            // 默认3秒超时
            SendResult sendResult = defaultMQProducer.send(sendMsg);
            LOGGER.info("消息发送响应：" + sendResult.toString());
            return new Results().succeed(sendResult);
        }
     
    }

测试URL：[http://127.0.0.1:8802/mqProducer/send?msg=hello][http_127.0.0.1_8802_mqProducer_send_msg_hello]

### 4.8.rocketmq控制台查看 ###

127.0.0.1:8088/\#/toplc

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4bDY0NjgwMTkyNA_size_16_color_FFFFFF_t_70][]

项目框架学习示例项目地址：

[GitHub - LockieZou/springcloud-nacos-demo][SpringCloud_Nacos]

##  ##

[RocketMQ]: https://so.csdn.net/so/search?q=RocketMQ&spm=1001.2101.3001.7020
[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/0b7f77e3eebd4925a8f5eea7827bbc03.png
[ce5442a365ece401ee83c0081b4858ee.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/9ebdc893f35946518fe9f779bee8eaa7.png
[format_png 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/37ed497feabb4705a4f80208507506cb.png
[format_png 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/59eb9cad002c469985cef8c5d9f1c8ee.png
[format_png 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/a022a7c6352e44fb86b1759689deaac3.png
[windows_RocketMQ]: https://blog.csdn.net/zxl646801924/article/details/105637278
[SpringCloud_Nacos]: https://github.com/LockieZou/springcloud-nacos-demo
[springcloud_nacos_lockie_-CSDN]: https://blog.csdn.net/zxl646801924/article/details/103984191
[http_127.0.0.1_8802_mqProducer_send_msg_hello]: http://127.0.0.1:8802/mqProducer/send?msg=hello
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p4bDY0NjgwMTkyNA_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/ecbaf6119c454e92b9d9897fca9627e0.png