快捷導(dǎo)航

使用Java代碼實(shí)現(xiàn)RocketMQ的生產(chǎn)與消費(fèi)消息

更新時(shí)間：2024年07月31日 08:43:23 作者：小威要向諸佬學(xué)習(xí)呀

這篇文章介紹一下其他的小組件以及使用Java代碼實(shí)現(xiàn)生產(chǎn)者對(duì)消息的生成,消費(fèi)者消費(fèi)消息等知識(shí)點(diǎn),并通過代碼示例介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作有一定的幫助,需要的朋友可以參考下

RocketMQ其他組件

在RocketMQ中，除了生產(chǎn)者，消費(fèi)者，還有一些其他的小組件，接下來逐一介紹一下他們。

監(jiān)聽器（Listener）

定義：監(jiān)聽器是消費(fèi)者用于處理消息的組件。在PushConsumer（推）模式下，消費(fèi)者客戶端必須設(shè)置消費(fèi)監(jiān)聽器，以便在接收到消息時(shí)執(zhí)行相應(yīng)的處理邏輯。（比如一會(huì)兒下面的代碼）

偏移量（Offset）

定義：偏移量是指在消費(fèi)消息時(shí)，記錄消費(fèi)者已經(jīng)消費(fèi)到的消息位置的值。每個(gè)消息都有一個(gè)唯一的偏移量值，它代表了消息在消息隊(duì)列中的位置。

偏移量具有很大的作用：它能夠保證消費(fèi)者在重啟或宕機(jī)后能夠從上次消費(fèi)的位置繼續(xù)消費(fèi)消息，避免重復(fù)消費(fèi)或漏消費(fèi)。

簡而言之就是它能告訴消費(fèi)者已經(jīng)消費(fèi)到哪一條消息?。?/p>

集群模式：在集群消費(fèi)模式下，消息隊(duì)列的消費(fèi)進(jìn)度保存在Broker端。消費(fèi)者每次消費(fèi)完消息后，會(huì)將最新的消費(fèi)進(jìn)度同步到Broker，以便在消費(fèi)者重啟或者故障轉(zhuǎn)移的時(shí)候能夠從上一次消費(fèi)的位置繼續(xù)消費(fèi)。
廣播模式：在廣播消費(fèi)模式下，消息隊(duì)列的消費(fèi)進(jìn)度保存在消費(fèi)者本地。因?yàn)閺V播模式下每條消息都會(huì)被所有消費(fèi)者消費(fèi)，所以不需要在Broker端保存消費(fèi)進(jìn)度。

所以，偏移量的的實(shí)現(xiàn)方式有兩種：包括存儲(chǔ)在本地文件（OffsetStore）和存儲(chǔ)在Broker中這兩種方式。（這樣一看，清晰了吧）

命名服務(wù)器（NameServer）

定義：命名服務(wù)器是RocketMQ中的輕量級(jí)路由服務(wù)，存儲(chǔ)生產(chǎn)者和消費(fèi)者與Broker之間的路由信息。

它的作用：提供Broker的動(dòng)態(tài)注冊與發(fā)現(xiàn)服務(wù)，生產(chǎn)者和消費(fèi)者通過NameServer查詢Broker的路由信息，從而進(jìn)行消息的投遞和消費(fèi)。

消息組成

Topic：消息主題，對(duì)不同的業(yè)務(wù)消息進(jìn)行分類。
Tag：消息標(biāo)簽，進(jìn)一步區(qū)分某個(gè)Topic下的消息分類。使用Tag可以實(shí)現(xiàn)對(duì)Topic中的消息進(jìn)行過濾。消費(fèi)者可以根據(jù)Tag來訂閱自己感興趣的消息，而不是接收Topic下的所有消息。
Message Body：消息體，消息的實(shí)際內(nèi)容。
Keys：消息的鍵值，標(biāo)識(shí)消息的唯一性。在RocketMQ中，每個(gè)消息都可以設(shè)置Keys字段，以便在需要的時(shí)候根據(jù)Keys來查詢或者定位消息。
屬性：除了上面滴，RocketMQ的消息還可以包含一系列的屬性信息，比如消息的發(fā)送時(shí)間、生產(chǎn)者信息等等。這些屬性信息以鍵值對(duì)的形式存在，隨著消息一起被存儲(chǔ)和傳輸。

實(shí)現(xiàn)生產(chǎn)與消費(fèi)消息

按之前的步驟搭建完成RocketMQ集群后...

首先我們創(chuàng)建一個(gè)空的maven工程，在pom.xml文件中添加RocketMQ的依賴（RocketMQ的依賴版本需要與虛擬機(jī)中的保持一致，這里選擇和之前一樣的4.7.1版本）：

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

生產(chǎn)消息

然后編寫生產(chǎn)者生產(chǎn)消息的代碼：

// 1.創(chuàng)建一個(gè)DefaultMQProducer實(shí)例，指定生產(chǎn)者組名 
DefaultMQProducer producer = new DefaultMQProducer("my-producer-group1");  
// 2.設(shè)置NameServer的地址
producer.setNamesrvAddr("192.168.220.135:9876");    
// 3.啟動(dòng)生產(chǎn)者實(shí)例  
producer.start();   
// 4.使用for循環(huán)發(fā)送10條消息  
for (int i = 0; i < 10; i++) {  
    // 創(chuàng)建一條消息，指定Topic為"MyTopic1"，Tag標(biāo)簽為"TagA"，消息體為"hello rocketmq"加上循環(huán)變量的值，同時(shí)把字符串轉(zhuǎn)換為字節(jié)數(shù)組  
    Message message = new Message("MyTopic1","TagA",("hello rocketmq"+i).getBytes(StandardCharsets.UTF_8));       
    // 5.發(fā)送消息并接收發(fā)送結(jié)果  
    SendResult sendResult = producer.send(message);  
    // 打印發(fā)送結(jié)果，包括消息ID、發(fā)送狀態(tài)等信息  
    System.out.println(sendResult);  
}  
// 6.發(fā)送完所有消息后，關(guān)閉生產(chǎn)者實(shí)例，釋放資源  
producer.shutdown();

生產(chǎn)者生產(chǎn)消息和消費(fèi)者消費(fèi)消息這塊的代碼都相對(duì)較為簡單，已經(jīng)在代碼塊中加了注釋，這里就不再贅述了。

這個(gè)時(shí)候就可以訪問虛擬機(jī)+端口號(hào)來搜索到發(fā)送的消息詳情了！

消費(fèi)消息

// 1.和生產(chǎn)者一樣，創(chuàng)建一個(gè)DefaultMQPushConsumer實(shí)例
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my-consumer-group1");  
// 2.設(shè)置NameServer的地址，消費(fèi)者通過這個(gè)地址與NameServer進(jìn)行通信，來獲取Broker的地址信息  
consumer.setNamesrvAddr("192.168.220.135:9876");  
// 訂閱一個(gè)或多個(gè)Topic，以及Tag來過濾需要消費(fèi)的消息。我們訂閱了"MyTopic1"，使用"*"來匹配此Topic下的所有Tag  
consumer.subscribe("MyTopic1", "*");  
// 3.注冊消息監(jiān)聽器，用于處理從Broker接收到的消息。使用MessageListenerConcurrently接口的實(shí)現(xiàn)，表示并行消費(fèi)  
consumer.registerMessageListener(new MessageListenerConcurrently() {  
    @Override  
    // 4.當(dāng)收到消息時(shí)，方法會(huì)被調(diào)用。
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {  
        // 5.遍歷消息列表，并打印每條消息的內(nèi)容（注意：這里直接打印msg對(duì)象不會(huì)得到預(yù)期的消息內(nèi)容字符串）  
        for (MessageExt msg : msgs) {  
            // 所以我們打印msg.getBody()的內(nèi)容，為了保留消息原樣  
            System.out.println("已收到消息" + msg);  
        }  
        // 6.返回消費(fèi)狀態(tài)，這里表示消息已成功消費(fèi)  
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
    }  
});  
// 7.啟動(dòng)消費(fèi)者實(shí)例  
consumer.start();  
// 8.打印日志消費(fèi)者已啟動(dòng)  
System.out.println("消費(fèi)者已啟動(dòng)");

這里需要注意，msg包含了消息的詳細(xì)信息，包括消息體、標(biāo)簽、屬性等等。如果想打印消息內(nèi)容，應(yīng)該使用msg.getBody()方法獲取消息體的字節(jié)數(shù)組，并且把它轉(zhuǎn)換為字符串（如果消息體是文本的話）。