SpringBoot3集成RocketMq場(chǎng)景分析

更新時(shí)間：2023年08月17日 09:35:56 作者：知了一笑

RocketMQ因其架構(gòu)簡(jiǎn)單、業(yè)務(wù)功能豐富、具備極強(qiáng)可擴(kuò)展性等特點(diǎn)被廣泛應(yīng)用,比如金融業(yè)務(wù)、互聯(lián)網(wǎng)、大數(shù)據(jù)、物聯(lián)網(wǎng)等領(lǐng)域的業(yè)務(wù)場(chǎng)景,這篇文章主要介紹了SpringBoot3集成RocketMq,需要的朋友可以參考下

RocketMQ因其架構(gòu)簡(jiǎn)單、業(yè)務(wù)功能豐富、具備極強(qiáng)可擴(kuò)展性等特點(diǎn)被廣泛應(yīng)用，比如金融業(yè)務(wù)、互聯(lián)網(wǎng)、大數(shù)據(jù)、物聯(lián)網(wǎng)等領(lǐng)域的業(yè)務(wù)場(chǎng)景；

標(biāo)簽：RocketMq5.Dashboard；

一、簡(jiǎn)介

二、環(huán)境部署

1、編譯打包

1、下載5.0版本源碼包
rocketmq-all-5.0.0-source-release.zip
2、解壓后進(jìn)入目錄，編譯打包
mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U

2、修改配置

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

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

3、服務(wù)啟動(dòng)

1、該目錄下
distribution/target/rocketmq-5.0.0/rocketmq-5.0.0/bin/
2、啟動(dòng)NameServer
sh mqnamesrv
輸出日志
The Name Server boot success. serializeType=JSON
3、啟動(dòng)Broker+Proxy
sh mqbroker -n localhost:9876 --enable-proxy
輸出日志
rocketmq-proxy startup successfully
4、關(guān)閉服務(wù)
sh mqshutdown namesrv
Send shutdown request to mqnamesrv(18636) OK
sh mqshutdown broker
Send shutdown request to mqbroker with proxy enable OK(18647)

4、控制臺(tái)安裝

1、下載master源碼包
rocketmq-dashboard-master
2、解壓后進(jìn)入目錄，編譯打包
mvn clean package -Dmaven.test.skip=true
3、啟動(dòng)服務(wù)
java -jar target/rocketmq-dashboard-1.0.1-SNAPSHOT.jar
4、輸出日志
INFO main - Tomcat started on port(s): 8080 (http) with context path ''
5、訪問服務(wù)：localhost:8080

三、工程搭建

1、工程結(jié)構(gòu)

2、依賴管理

在 rocketmq-starter 組件中，實(shí)際上依賴的是 rocketmq-client 組件的 5.0 版本，由于兩個(gè)新版框架間的兼容問題，需要添加相關(guān)配置解決該問題；

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>${rocketmq-starter.version}</version>
</dependency>

3、配置文件

配置RocketMq服務(wù)地址，消息生產(chǎn)者和消費(fèi)者；

rocketmq:
  name-server: 127.0.0.1:9876
  # 生產(chǎn)者
  producer:
    group: boot_group_1
    # 消息發(fā)送超時(shí)時(shí)間
    send-message-timeout: 3000
    # 消息最大長度4M
    max-message-size: 4096
    # 消息發(fā)送失敗重試次數(shù)
    retry-times-when-send-failed: 3
    # 異步消息發(fā)送失敗重試次數(shù)
    retry-times-when-send-async-failed: 2
  # 消費(fèi)者
  consumer:
    group: boot_group_1
    # 每次提取的最大消息數(shù)
    pull-batch-size: 5

4、配置類

在配置類中主要定義兩個(gè)Bean的加載，即 RocketMQTemplate 和 DefaultMQProducer ，主要是提供消息發(fā)送的能力，即生產(chǎn)消息；

@Configuration
public class RocketMqConfig {
    @Value("${rocketmq.name-server}")
    private String nameServer;
    @Value("${rocketmq.producer.group}")
    private String producerGroup;
    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer sendMsgTimeout;
    @Value("${rocketmq.producer.max-message-size}")
    private Integer maxMessageSize;
    @Value("${rocketmq.producer.retry-times-when-send-failed}")
    private Integer retryTimesWhenSendFailed ;
    @Value("${rocketmq.producer.retry-times-when-send-async-failed}")
    private Integer retryTimesWhenSendAsyncFailed ;
    @Bean
    public RocketMQTemplate rocketMqTemplate(){
        RocketMQTemplate rocketMqTemplate = new RocketMQTemplate();
        rocketMqTemplate.setProducer(defaultMqProducer());
        return rocketMqTemplate;
    }
    @Bean
    public DefaultMQProducer defaultMqProducer() {
        DefaultMQProducer producer = new DefaultMQProducer();
        producer.setNamesrvAddr(this.nameServer);
        producer.setProducerGroup(this.producerGroup);
        producer.setSendMsgTimeout(this.sendMsgTimeout);
        producer.setMaxMessageSize(this.maxMessageSize);
        producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
        producer.setRetryTimesWhenSendAsyncFailed(this.retryTimesWhenSendAsyncFailed);
        return producer;
    }
}

四、基礎(chǔ)用法

1、消息生產(chǎn)

編寫一個(gè)生產(chǎn)者接口類，分別使用 RocketMQTemplate 和 DefaultMQProducer 實(shí)現(xiàn)消息發(fā)送的功能，然后可以通過 Dashboard 控制面板查看消息詳情；

@RestController
public class ProducerWeb {
    private static final Logger log = LoggerFactory.getLogger(ProducerWeb.class);
    @Autowired
    private RocketMQTemplate rocketMqTemplate;
    @GetMapping("/send/msg1")
    public String sendMsg1 (){
        try {
            // 構(gòu)建消息主體
            JsonMapper jsonMapper = new JsonMapper();
            String msgBody = jsonMapper.writeValueAsString(new MqMsg(1,"boot_mq_msg"));
            // 發(fā)送消息
            rocketMqTemplate.convertAndSend("boot-mq-topic",msgBody);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "OK" ;
    }
    @Autowired
    private DefaultMQProducer defaultMqProducer ;
    @GetMapping("/send/msg2")
    public String sendMsg2 (){
        try {
            // 構(gòu)建消息主體
            JsonMapper jsonMapper = new JsonMapper();
            String msgBody = jsonMapper.writeValueAsString(new MqMsg(2,"boot_mq_msg"));
            // 構(gòu)建消息對(duì)象
            Message message = new Message();
            message.setTopic("boot-mq-topic");
            message.setTags("boot-mq-tag");
            message.setKeys("boot-mq-key");
            message.setBody(msgBody.getBytes());
            // 發(fā)送消息，打印日志
            SendResult sendResult = defaultMqProducer.send(message);
            log.info("msgId:{},sendStatus:{}",sendResult.getMsgId(),sendResult.getSendStatus());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "OK" ;
    }
}

2、消息消費(fèi)

編寫消息監(jiān)聽類，實(shí)現(xiàn) RocketMQListener 接口，通過 RocketMQMessageListener 注解控制監(jiān)聽的具體信息；

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