SpringBoot3集成RocketMq

标签：RocketMq5.Dashboard；

一、简介

RocketMQ因其架构简单、业务功能丰富、具备极强可扩展性等特点被广泛应用，比如金融业务、互联网、大数据、物联网等领域的业务场景；
二、环境部署

1、编译打包

1. 1、下载5.0版本源码包
2. rocketmq-all-5.0.0-source-release.zip
4. 2、解压后进入目录，编译打包
5. mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U

复制代码

2、修改配置

1. 在distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/runserver.sh

复制代码

1. distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/runbroker.sh

复制代码

3、服务启动

1. 1、该目录下
2. distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/
4. 2、启动NameServer
5. sh mqnamesrv
7. 输出日志
8. The Name Server boot success. serializeType=JSON
10. 3、启动Broker+Proxy
11. sh mqbroker -n localhost:9876 --enable-proxy
13. 输出日志
14. rocketmq-proxy startup successfully
16. 4、关闭服务
17. sh mqshutdown namesrv
18. Send shutdown request to mqnamesrv(18636) OK
20. sh mqshutdown broker
21. Send shutdown request to mqbroker with proxy enable OK(18647)

复制代码

4、控制台安装

1. 1、下载master源码包
2. rocketmq-dashboard-master
4. 2、解压后进入目录，编译打包
5. mvn clean package -Dmaven.test.skip=true
7. 3、启动服务
8. java -jar target/rocketmq-dashboard-1.0.1-SNAPSHOT.jar
10. 4、输出日志
11. INFO main - Tomcat started on port(s): 8080 (http) with context path ''
13. 5、访问服务：localhost:8080

复制代码

三、工程搭建

1、工程结构


2、依赖管理

在rocketmq-starter组件中，实际上依赖的是rocketmq-client组件的5.0版本，由于两个新版框架间的兼容问题，需要添加相关配置解决该问题；

1. <dependency>
2.     <groupId>org.apache.rocketmq</groupId>
3.     <artifactId>rocketmq-spring-boot-starter</artifactId>
4.     <version>${rocketmq-starter.version}</version>
5. </dependency>

复制代码

3、配置文件

配置RocketMq服务地址，消息生产者和消费者；

1. rocketmq:
2.   name-server: 127.0.0.1:9876
3.   # 生产者
4.   producer:
5.     group: boot_group_1
6.     # 消息发送超时时间
7.     send-message-timeout: 3000
8.     # 消息最大长度4M
9.     max-message-size: 4096
10.     # 消息发送失败重试次数
11.     retry-times-when-send-failed: 3
12.     # 异步消息发送失败重试次数
13.     retry-times-when-send-async-failed: 2
14.   # 消费者
15.   consumer:
16.     group: boot_group_1
17.     # 每次提取的最大消息数
18.     pull-batch-size: 5

复制代码

4、配置类

在配置类中主要定义两个Bean的加载，即RocketMQTemplate和DefaultMQProducer，主要是提供消息发送的能力，即生产消息；

1. @Configuration
2. public class RocketMqConfig {
4.     @Value("${rocketmq.name-server}")
5.     private String nameServer;
7.     @Value("${rocketmq.producer.group}")
8.     private String producerGroup;
10.     @Value("${rocketmq.producer.send-message-timeout}")
11.     private Integer sendMsgTimeout;
13.     @Value("${rocketmq.producer.max-message-size}")
14.     private Integer maxMessageSize;
16.     @Value("${rocketmq.producer.retry-times-when-send-failed}")
17.     private Integer retryTimesWhenSendFailed ;
19.     @Value("${rocketmq.producer.retry-times-when-send-async-failed}")
20.     private Integer retryTimesWhenSendAsyncFailed ;
22.     @Bean
23.     public RocketMQTemplate rocketMqTemplate(){
24.         RocketMQTemplate rocketMqTemplate = new RocketMQTemplate();
25.         rocketMqTemplate.setProducer(defaultMqProducer());
26.         return rocketMqTemplate;
27.     }
29.     @Bean
30.     public DefaultMQProducer defaultMqProducer() {
31.         DefaultMQProducer producer = new DefaultMQProducer();
32.         producer.setNamesrvAddr(this.nameServer);
33.         producer.setProducerGroup(this.producerGroup);
34.         producer.setSendMsgTimeout(this.sendMsgTimeout);
35.         producer.setMaxMessageSize(this.maxMessageSize);
36.         producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
37.         producer.setRetryTimesWhenSendAsyncFailed(this.retryTimesWhenSendAsyncFailed);
38.         return producer;
39.     }
40. }

复制代码

四、基础用法

1、消息生产

编写一个生产者接口类，分别使用RocketMQTemplate和DefaultMQProducer实现消息发送的功能，然后可以通过Dashboard控制面板查看消息详情；

1. @RestController
2. public class ProducerWeb {
3.     private static final Logger log = LoggerFactory.getLogger(ProducerWeb.class);
5.     @Autowired
6.     private RocketMQTemplate rocketMqTemplate;
8.     @GetMapping("/send/msg1")
9.     public String sendMsg1 (){
10.         try {
11.             // 构建消息主体
12.             JsonMapper jsonMapper = new JsonMapper();
13.             String msgBody = jsonMapper.writeValueAsString(new MqMsg(1,"boot_mq_msg"));
14.             // 发送消息
15.             rocketMqTemplate.convertAndSend("boot-mq-topic",msgBody);
16.         } catch (Exception e) {
17.             e.printStackTrace();
18.         }
19.         return "OK" ;
20.     }
22.     @Autowired
23.     private DefaultMQProducer defaultMqProducer ;
25.     @GetMapping("/send/msg2")
26.     public String sendMsg2 (){
27.         try {
28.             // 构建消息主体
29.             JsonMapper jsonMapper = new JsonMapper();
30.             String msgBody = jsonMapper.writeValueAsString(new MqMsg(2,"boot_mq_msg"));
31.             // 构建消息对象
32.             Message message = new Message();
33.             message.setTopic("boot-mq-topic");
34.             message.setTags("boot-mq-tag");
35.             message.setKeys("boot-mq-key");
36.             message.setBody(msgBody.getBytes());
37.             // 发送消息，打印日志
38.             SendResult sendResult = defaultMqProducer.send(message);
39.             log.info("msgId:{},sendStatus:{}",sendResult.getMsgId(),sendResult.getSendStatus());
40.         } catch (Exception e) {
41.             e.printStackTrace();
42.         }
43.         return "OK" ;
44.     }
45. }

复制代码

2、消息消费

编写消息监听类，实现RocketMQListener接口，通过RocketMQMessageListener注解控制监听的具体信息；

1. @Service
2. @RocketMQMessageListener(consumerGroup = "boot_group_1",topic = "boot-mq-topic")
3. public class ConsumerListener implements RocketMQListener<String> {
5.     private static final Logger log = LoggerFactory.getLogger(ConsumerListener.class);
7.     @Override
8.     public void onMessage(String message) {
9.         log.info("\n=====\n message：{} \n=====\n",message);
10.     }
11. }

复制代码

五、参考源码

1. 文档仓库：
2. https://gitee.com/cicadasmile/butte-java-note
4. 源码仓库：
5. https://gitee.com/cicadasmile/butte-spring-parent

复制代码