當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

RocketMQ-初体验RocketMQ(06)-使用API操作RocketMQ ,理解RocketMQ的存储结构

發(fā)布時(shí)間：2025/3/21 编程问答 21 豆豆

生活随笔收集整理的這篇文章主要介紹了 RocketMQ-初体验RocketMQ(06)-使用API操作RocketMQ ,理解RocketMQ的存储结构小編覺得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

文章目錄

RocketMQ集群基本信息
目標(biāo)
知識(shí)預(yù)習(xí)
- 發(fā)送方式
- 發(fā)送結(jié)果
環(huán)境搭建
使用Java API操作RocketMQ—Simple Message
- Step1. pom.xml增加依賴和bulid 插件
- Step2.日志文件 logback.xml
- 發(fā)送同步消息
- 發(fā)送異步消息
- 發(fā)送one way 消息
- 消費(fèi)消息
- - push模式
  - pull模式
結(jié)構(gòu)變化
代碼
更多示例

RocketMQ集群基本信息

右側(cè)的部署模式，雙機(jī)互為主備

如何搭建的，請(qǐng)移步： RocketMQ-初體驗(yàn)RocketMQ(03)_RocketMQ多機(jī)集群部署

為了更好的觀察RocketMQ在我們發(fā)送和消費(fèi)消息的過程中，給我們產(chǎn)生了什么樣的文件，我們把 storePathRootDir 和 storePathCommitLog 自定義到一個(gè)新的目錄下

130 節(jié)點(diǎn)的 broker-m.conf 和 broker-s.conf中的 storePathRootDir 和 storePathCommitlog配置

一個(gè)broker節(jié)點(diǎn)對(duì)應(yīng)一個(gè)commitlog, 所以130主機(jī)的 master broker 和 slave broker 會(huì)對(duì)應(yīng)兩個(gè)存儲(chǔ)路徑和兩個(gè)commitlog . 注意看存儲(chǔ)路徑

broker-m.conf

broker-s.conf

啟動(dòng) 130的 namesrv 和 master broker 、 slave broker 節(jié)點(diǎn)，觀察我們配置的
storePathRootDir 和 storePathCommitlog

同樣的也把131 修改下

目標(biāo)

1. 使用RocketMQ 發(fā)送3種類型的消息： reliable synchronous、 reliable asynchronous、one-way transmission

2. 使用RocketMQ消費(fèi)消息

知識(shí)預(yù)習(xí)

發(fā)送方式

在Producer端

org.apache.rocketmq.client.impl.CommunicationMode 枚舉類中定義了3種發(fā)送方式

public enum CommunicationMode {SYNC,ASYNC,ONEWAY, }

Sync：同步的發(fā)送方式，會(huì)等待發(fā)送結(jié)果后才返回
Async：異步的發(fā)送方式，發(fā)送完后，立刻返回。Client 在拿到 Broker 的響應(yīng)結(jié)果后，會(huì)回調(diào)指定的 callback. 可以指定 Timeout。默認(rèn)的 3000ms.
Oneway：發(fā)出去后，什么都不管直接返回

發(fā)送結(jié)果

org.apache.rocketmq.client.produce.SendStatus 枚舉類中定義了如下4種發(fā)送結(jié)果

package org.apache.rocketmq.client.producer;public enum SendStatus {SEND_OK,FLUSH_DISK_TIMEOUT,FLUSH_SLAVE_TIMEOUT,SLAVE_NOT_AVAILABLE, }

1. SEND_OK ：消息發(fā)送成功。要注意的是消息發(fā)送成功也不意味著它是可靠的。要確保不會(huì)丟失任何消息，還應(yīng)啟用同步Master服務(wù)器或同步刷盤，即SYNC_MASTER或 SYNC_FLUSH。

2. FLUSH_DISK_TIMEOUT：消息發(fā)送成功但是服務(wù)器刷盤超時(shí)。此時(shí)消息已經(jīng)進(jìn)入服務(wù)器隊(duì)列（內(nèi)存），只有服務(wù)器宕機(jī)，消息才會(huì)丟失。消息存儲(chǔ)配置參數(shù)中可以設(shè)置刷盤方式和同步刷盤時(shí)間長度，如果Broker服務(wù)器設(shè)置了刷盤方式為同步刷盤，即FlushDiskType=SYNC_FLUSH（默認(rèn)為異步刷盤方式），當(dāng)Broker服務(wù)器未在同步刷盤時(shí)間內(nèi)（默認(rèn)為5s）完成刷盤，則將返回該狀態(tài)——刷盤超時(shí)。

3. FLUSH_SLAVE_TIMEOUT ：消息發(fā)送成功，但是服務(wù)器同步到Slave時(shí)超時(shí)。此時(shí)消息已經(jīng)進(jìn)入服務(wù)器隊(duì)列，只有服務(wù)器宕機(jī)，消息才會(huì)丟失。如果Broker服務(wù)器的角色是同步Master，即SYNC_MASTER（默認(rèn)是異步Master即ASYNC_MASTER），并且從Broker服務(wù)器未在同步刷盤時(shí)間（默認(rèn)為5秒）內(nèi)完成與主服務(wù)器的同步，則將返回該狀態(tài)——數(shù)據(jù)同步到Slave服務(wù)器超時(shí)

4. SLAVE_NOT_AVAILABLE：消息發(fā)送成功，但是此時(shí)Slave不可用。如果Broker服務(wù)器的角色是同步Master，即SYNC_MASTER（默認(rèn)是異步Master服務(wù)器即ASYNC_MASTER），但沒有配置slaveBroker服務(wù)器，則將返回該狀態(tài)——無Slave服務(wù)器可用。

環(huán)境搭建

移步： IDEA-使用IDEA創(chuàng)建maven多模塊父子工程

使用Java API操作RocketMQ—Simple Message

官方指導(dǎo)：戳這里

Step1. pom.xml增加依賴和bulid 插件

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>rocketmqMaster</artifactId><groupId>com.artisan</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><artifactId>rocketmq_base</artifactId><dependencies><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>4.3.2</version></dependency></dependencies><build><pluginManagement><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>2.3.2</version><configuration><source>1.8</source><target>1.8</target><encoding>utf8</encoding><compilerArgument>-XDignore.symbol.file=true -Xlint</compilerArgument><testCompilerArgument>-XDignore.symbol.file=true -Xlint</testCompilerArgument></configuration></plugin></plugins></pluginManagement></build></project>

Step2.日志文件 logback.xml

<configuration><property name="APP_NAME" value="rocketmq_base" /><property name="LOG_HOME" value="${log.dir:-logs}/${APP_NAME}" /><property name="ENCODER_PATTERN"value="%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{80} - %msg%n" /><contextName>${APP_NAME}</contextName><appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"><encoder><pattern>${ENCODER_PATTERN}</pattern></encoder></appender><appender name="FILE"class="ch.qos.logback.core.rolling.RollingFileAppender"><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><fileNamePattern>${LOG_HOME}/output.%d{yyyy-MM-dd}.log</fileNamePattern><maxHistory>7</maxHistory></rollingPolicy><encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"><pattern>${ENCODER_PATTERN}</pattern></encoder></appender><appender name="ERROR_FILE"class="ch.qos.logback.core.rolling.RollingFileAppender"><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><fileNamePattern>${LOG_HOME}/error.%d{yyyy-MM-dd}.log</fileNamePattern><maxHistory>7</maxHistory></rollingPolicy><encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"><pattern>${ENCODER_PATTERN}</pattern></encoder><filter class="ch.qos.logback.classic.filter.ThresholdFilter"><level>WARN</level></filter></appender><appender name="SYNC_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"><rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"><fileNamePattern>${LOG_HOME}/sync.%d{yyyy-MM-dd}.log</fileNamePattern><maxHistory>7</maxHistory></rollingPolicy><encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder"><pattern>${ENCODER_PATTERN}</pattern></encoder></appender><logger name="rocketmq_base" level="ERROR" addtivity="true"><appender-ref ref="SYNC_FILE" /></logger><root level="ERROR"><appender-ref ref="STDOUT" /><appender-ref ref="FILE" /><appender-ref ref="ERROR_FILE" /></root> </configuration>

發(fā)送同步消息

package com.artisan.rocketmq.simple.producer;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.common.RemotingHelper; import org.apache.rocketmq.remoting.exception.RemotingException;import java.io.UnsupportedEncodingException;/*** @author 小工匠* @version v1.0* @create 2019-11-10 1:46* @motto show me the code ,change the word* @blog https://artisan.blog.csdn.net/* @description 同步發(fā)送消息**/public class SyncProducer {public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException {//Instantiate with a producer group name.DefaultMQProducer producer = newDefaultMQProducer("Artisan_ProducerGroup");// Specify name server addresses.producer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");// 設(shè)置超時(shí)時(shí)間，默認(rèn)3秒producer.setSendMsgTimeout(10_000);//Launch the instance.producer.start(); // for (int i = 0; i < 100; i++) { // //Create a message instance, specifying topic, tag and message body. // Message msg = new Message("TopicArtisan" /* Topic */, // "TagArtisan" /* Tag */, // ("Artisan:Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */ // ); // //Call send message to deliver message to one of brokers. // SendResult sendResult = producer.send(msg); // System.out.printf("%s%n", sendResult); // }//Create a message instance, specifying topic, tag and message body.Message msg = new Message("TopicArtisan" /* Topic */,"TagArtisan" /* Tag */,("Artisan:Hello RocketMQ ").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);//Call send message to deliver message to one of brokers.SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);//Shut down once the producer instance is not longer in use.producer.shutdown();} }

如果超時(shí)的話，設(shè)置下超時(shí)時(shí)間 producer.setSendMsgTimeout(10_000);

發(fā)送一條消息

SendResult [sendStatus=SEND_OK, msgId=C0A81F891D2418B4AAC230A647AD0000, offsetMsgId=C0A8128300002A9F00000000000025B0, messageQueue=MessageQueue [topic=TopicArtisan, brokerName=brokera€?b, queueId=3], queueOffset=13]

可知發(fā)送到了集群中的節(jié)點(diǎn) 的queueId=3 第四個(gè)隊(duì)列里去了

到控制臺(tái)根據(jù)msgId查看一下

發(fā)送異步消息

package com.artisan.rocketmq.simple.producer;import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendCallback; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper;import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit;/*** @author 小工匠* @version v1.0* @create 2019-11-10 12:18* @motto show me the code ,change the word* @blog https://artisan.blog.csdn.net/* @description 異步消息**/public class AsyncProducer {public static void main(String[] args) throws Exception {DefaultMQProducer producer = new DefaultMQProducer("Artisan_ProducerGroup");producer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");//設(shè)置發(fā)送失敗重試機(jī)制producer.setRetryTimesWhenSendAsyncFailed(5);producer.start();int messageCount = 1;final CountDownLatch countDownLatch = new CountDownLatch(messageCount);for (int i = 0; i < messageCount; i++) {final int index = i;Message msg = new Message("TopicAsyn","TagAsyn","OrderID188","I m sending msg content xxx".getBytes(RemotingHelper.DEFAULT_CHARSET));//消息發(fā)送成功后，執(zhí)行回調(diào)函數(shù)producer.send(msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {countDownLatch.countDown();System.out.printf("%-10d OK %s %n", index,sendResult.getMsgId());}@Overridepublic void onException(Throwable e) {countDownLatch.countDown();System.out.printf("%-10d Exception %s %n", index, e);e.printStackTrace();}});}countDownLatch.await(5, TimeUnit.SECONDS);producer.shutdown();} }

日志：

控制臺(tái)查詢

發(fā)送one way 消息

package com.artisan.rocketmq.simple.producer;import org.apache.rocketmq.client.producer.DefaultMQProducer; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.remoting.common.RemotingHelper;/*** @author 小工匠* @version v1.0* @create 2019-11-10 12:45* @motto show me the code ,change the word* @blog https://artisan.blog.csdn.net/* @description**/public class OnewayProducer {public static void main(String[] args) throws Exception{DefaultMQProducer producer = new DefaultMQProducer("tl_message_group");// Specify name server addresses.producer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");producer.setSendMsgTimeout(10000);producer.start();for (int i = 0; i < 1; i++) {Message msg = new Message("TopicOneWay" /* Topic */,"TagSendOne" /* Tag */,"OrderID198",("Hello RocketMQ test i " +i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);producer.sendOneway(msg);}//Shut down once the producer instance is not longer in use.producer.shutdown();} }

消費(fèi)消息

push模式

package com.artisan.rocketmq.simple.consumer;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; 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.client.exception.MQClientException; import org.apache.rocketmq.common.message.MessageExt;import java.util.List;/*** @author 小工匠* @version v1.0* @create 2019-11-10 12:49* @motto show me the code ,change the word* @blog https://artisan.blog.csdn.net/* @description**/public class Consumer {public static void main(String[] args) throws InterruptedException, MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Artisan_ProducerGroup");consumer.setNamesrvAddr("192.168.18.130:9876;192.168.18.131:9876");// Subscribe one more more topics to consume.consumer.subscribe("TopicAsyn", "*");// Register callback to execute on arrival of messages fetched from brokers.consumer.registerMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {for (MessageExt msg : msgs){System.out.println(new String(msg.getBody()));}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.printf("Consumer Started.%n");} }

pull模式

package com.artisan.rocketmq.simple.consumer;import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer; import org.apache.rocketmq.client.consumer.PullResult; import org.apache.rocketmq.client.exception.MQClientException; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageQueue;import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set;/*** @author 小工匠* @version v1.0* @create 2019-11-10 13:51* @motto show me the code ,change the word* @blog https://artisan.blog.csdn.net/* @description**/public class PullConsumer {private static final Map<MessageQueue, Long> offsetTable = new HashMap<MessageQueue, Long>();public static void main(String[] args) throws MQClientException {DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("tl_message_group");consumer.setNamesrvAddr("192.168.241.198:9876");consumer.start();Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicStudent");for (MessageQueue mq : mqs) {System.err.println("Consume from the queue: " + mq);SINGLE_MQ:while (true) try {PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);System.out.println(pullResult);putMessageQueueOffset(mq, pullResult.getNextBeginOffset());switch (pullResult.getPullStatus()) {case FOUND:List<MessageExt> messageExtList = pullResult.getMsgFoundList();for (MessageExt m : messageExtList) {System.out.println(new String(m.getBody()));}break;case NO_MATCHED_MSG:break;case NO_NEW_MSG:break SINGLE_MQ;case OFFSET_ILLEGAL:break;default:break;}} catch (Exception e) {e.printStackTrace();}}consumer.shutdown();}private static void putMessageQueueOffset(MessageQueue mq, long offset) {offsetTable.put(mq, offset);}private static long getMessageQueueOffset(MessageQueue mq) {Long offset = offsetTable.get(mq);if (offset != null)return offset;return 0;} }

結(jié)構(gòu)變化

自動(dòng)創(chuàng)建了commitlog目錄，其中目錄下的文件固定為 1024M

自動(dòng)創(chuàng)建了 index 索引目錄

自動(dòng)創(chuàng)建consumequeue目錄

代碼

請(qǐng)移步：https://github.com/yangshangwei/rocketmqMaster