SpringBoot整合RocketMQ的方法詳解

更新時間：2022年08月12日 14:38:39 作者：羈客%

這篇文章主要為大家詳細(xì)介紹了SpringBoot整合RocketMQ的方法，文中的示例代碼講解詳細(xì)，對我們學(xué)習(xí)有一定幫助，感興趣的小伙伴可以了解一下

一：Ubuntu安裝RocketMQ

1.下載（在下面地址選擇自己需要的版本的rocketmq）

http://rocketmq.apache.org/release_notes/

2.解壓，更改配置

將下載的zip文件解壓到自己需要安裝的位置

在unbuntu系統(tǒng)下需要修改安裝跟目錄下的兩個文件

bin/runserver.sh 和 bin/runbroker.sh 文件

(1).都修改兩個文件的 java_home=/java安裝HOME路徑/

(2).都修改兩個文件的 java_home=/java安裝HOME路徑/jre/

(3).修改runserver.sh內(nèi)存

JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"
 
# 原來是：JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn4g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

(4).修改runbroker.sh內(nèi)存

JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g   
# 原來是：JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g

(5).創(chuàng)建日志目錄

mkdir logs

3.后臺掛載啟動 mqnamesrv(在安裝根目錄下執(zhí)行)

nohup sh bin/mqnamesrv > logs/namesrv.log 2>&1 &

5.查看mqnamesrv是否啟動成功

查看 logs/namesrv.log 文件出現(xiàn)以下提示表示啟動成功！

The Name Server boot success. serializeType=JSON

6.安裝 mqbroker

Broker是 RocketMQ 的核心,大部分'重量級'工作都是由 Broker完成的,

包括接收 Producer 發(fā)過來的消息、處理 Consumer 的消費消息請求、消息的持久化存儲、消息的 HA 機(jī)制以及服務(wù)端過濾功能等。

(1).更改broker配置，在安裝根目錄下conf/broker.conf文件添加一行

brokerIP1 = 127.0.0.0
# IP根據(jù)自己電腦服務(wù)器修改一下，最好不要使用這個,ifconfig就可以查看到自己服務(wù)器的IP

(2).掛載啟動 mqbroker（在安裝根目錄下）

nohup sh bin/mqbroker -c conf/broker.conf -n localhost:9876 > logs/broker.log 2>&1 &

(3).查看 mqbroker 是否啟動成功

查看 logs/broker.log 文件出現(xiàn)以下提示表示啟動成功！

The broker[broker-a, 127.0.0.0:10911] boot success. serializeType=JSON and name server is localhost:9876

7.關(guān)閉broker和nameserver(在安裝根目錄下執(zhí)行)

./bin/mqshutdown broker
./bin/mqshutdown namesrv

8.創(chuàng)建一個topic

./bin/mqadmin updateTopic -n '127.0.0.0:9876' -c MyCluster -t MyTopic

9.查看集群狀態(tài)

./bin/mqadmin clusterList -n localhost:9876

二：添加RocketMQ依賴

<!--rocket mq 依賴-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
        </exclusion>
        <exclusion>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
        </exclusion>
    </exclusions>
</dependency>

三：在application中添加RocketMQ配置

#rocketmq 配置
rocketmq:
  name-server: 127.0.0.0:9876 #IP根據(jù)自己電腦服務(wù)器配置的 mqbroker->brokerIP1 修改一下，最好不要使用這個
  producer:
    group: rocketmq_group
 
myrocketmq-config:
  my-topic: rocketmq_topic
  my-consumer-group: rocketmq_group_consumer

四：編寫消費者，消息生產(chǎn)者，消息實體類(自定義)

//Rocket消費者
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "${myrocketmq-config.my-consumer-group}", topic = "${myrocketmq-config.my-topic}")
public class RocketConsumer implements RocketMQListener<RocketmqVo> {
 
    @Override
    public void onMessage(RocketmqVo rocketmqVo) {
        log.info("收到RocketMQ消息:{}",rocketmqVo);
    }
}

//Rocket生產(chǎn)者
@Slf4j
@Component
public class RocketProduce {
 
    @Value("${myrocketmq-config.my-topic}")
    private String mqTopic;
    @Value("${myrocketmq-config.my-consumer-group}")
    private String mqConsumerGroup;
    @Autowired
    private RocketMQTemplate mqTemplate;
 
    /**
     * 同步發(fā)送
     *
     * @param title 發(fā)送消息
     * @param data  消息內(nèi)容
     */
    public void sync(String title, Object data) {
        String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date());
        RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data);
        SendResult sendResult = mqTemplate.syncSend(mqTopic, rocketmqVo);
        log.info("同步發(fā)送:{}", rocketmqVo);
        log.info("同步發(fā)送消息結(jié)果:{}", sendResult);
    }
 
    /**
     * 異步發(fā)送
     *
     * @param title 發(fā)送消息
     * @param data  消息內(nèi)容
     */
    public void async(String title, Object data) {
        String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date());
        RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data);
        log.info("異步發(fā)送:{}", rocketmqVo);
        mqTemplate.asyncSend(mqTopic, rocketmqVo, new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {
                log.info("異步發(fā)送成功:{}", var1);
            }
 
            @Override
            public void onException(Throwable var1) {
                log.info("異步發(fā)送失敗:{}", var1.getMessage());
            }
        });
    }
 
    /**
     * 單向發(fā)送
     *
     * @param title 發(fā)送消息
     */
    public void oneway(String title, Object data) {
        String time = new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date());
        RocketmqVo rocketmqVo = new RocketmqVo(mqConsumerGroup, mqTopic, title, time, data);
        mqTemplate.sendOneWay(mqTopic, rocketmqVo);
        log.info("單向發(fā)送:{}", rocketmqVo);
    }
}
//Rocket消息實體
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class RocketmqVo {
 
    @Value("${myrocketmq-config.my-consumer-group}")
    private String group;
 
    @Value("${myrocketmq-config.my-topic}")
    private String topic;
 
    private String title;
 
    private String date;
 
    private Object data;
 
 
    public RocketmqVo(String title, String date, Object data) {
        this.title = title;
        this.date = date;
        this.data = data;
    }
}

五：測試Controller

@RestController
@RequestMapping("/mqtest")
public class RocketController {
 
    @Resource
    private RocketProduce producer;
 
    @RequestMapping("/sendMessageSync/{msg}")
    public String sendMessageSync(@PathVariable("msg") String message) {
        producer.sync(message, null);
        return "消息發(fā)送完成";
    }
 
    @RequestMapping("/sendMessageAsync/{msg}")
    public String sendMessageAsync(@PathVariable("msg") String message) {
        producer.async(message, null);
        return "消息發(fā)送完成";
    }
 
    @RequestMapping("/sendMessageOneway/{msg}")
    public String sendMessageOneway(@PathVariable("msg") String message) {
        producer.oneway(message, null);
        return "消息發(fā)送完成";
    }
}

以上就是SpringBoot整合RocketMQ的方法詳解的詳細(xì)內(nèi)容，更多關(guān)于SpringBoot整合RocketMQ的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: