当前位置：首页 > news >正文

RocketMQ Learning(一)

news 2026/2/7 20:47:34

一、RocketMQ

0、RocketMQ的产品发展

1、RocketMQ安装

1.1、windows下的安装

注意事项

1.2、Linux下的安装

1.3、源码的安装

1.4、控制台

2、消息发送方式

2.1、发送同步消息

2.2、发送异步消息

2.3、单向发送

3、消息消费方式

3.1、负载均衡模式（集群消费）

3.2、广播消费

一、RocketMQ

0、RocketMQ的产品发展

        MetaQ：2011年，阿里基于Kafka的设计使用Java完全重写并推出了MetaQ 1.0版本。
        2012年，阿里对MetaQ的存储进行了改进，推出MetaQ 2.0，同年阿里把Meta2.0从阿里内部开源出来，取名RocketMQ，为了命名上的规范以及版本上的延续，对外称为RocketMQ3.0。
        2016年，阿里宣布将开源分布式消息中间件RocketMQ捐赠给Apache，同时RocketMQ3也升级为RocketMQ4，现在RocketMQ主要维护的是4.x的版本，也是大家使用得最多的版本。
        2021年，RocketMQ在github上发布5.0预览版。RocketMQ 5.0定义为云原生的消息、事件、流的超融合平台。

RocketMQ源码链接

RocketMQ官网下载地址

1、RocketMQ安装

1.1、windows下的安装

1.确保安装好了JDK1.8&64位系统

2.解压运行版本(Binary)

3.配置环境变量

变量名：ROCKETMQ_HOME
变量值：MQ解压路径\MQ文件夹名

4.启动

在RocketMQ的架构中，都是需要先启动NameServer再启动Broker的。所以先启动NameServer。
启动NameServer
使用cmd命令框执行进入至‘MQ文件夹\bin’下，然后执行‘start mqnamesrv.cmd’，启动NameServer。成功后会弹出提示框，此框勿关闭。

启动Broker
使用cmd命令框执行进入至‘MQ文件夹\bin’下，然后执行‘start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true’，启动Broker。成功后会弹出提示框，此框勿关闭。

注意事项

弹出提示框‘错误: 找不到或无法加载主类 Files\Java\jdk1.8.0_202\lib\tools.jar;C:\Program’的处理
打开‘MQ文件夹\bin’下的runbroker.cmd，然后将‘%CLASSPATH%’加上英文双引号。保存并重新执行start语句。

再次启动

1.2、Linux下的安装

...

1.3、源码的安装

1.解压源码版本(Source)

2.导入idea中，maven编译通过

3.创建目录D盘创建文件夹RocketMQ

1.把源码下的distribution下的conf文件夹拷贝到RocketMQ下面

2.再创建logs和store文件夹

4、启动RocketMQ源码

4.1、启动NameServer

namesrv工程下NamesrvStartup启动类，启动前需要配置环境变量

ROCKETMQ_HOME=D:\RocketMQ

4.2、启动Broker
在broker模块找到broker模块，同时找到启动类BrokerStartup.java

需要修改配置文件broker.conf

#配置如下：
#nameServer
namesrvAddr=127.0.0.1:9876
autoCreateTopicEnable = true
storePathRootDir = D:\\RocketMQ\\store
#commitLog存储路径
storePathCommitLog = D:\\RocketMQ\\store\\commitlog
#消费队列存储路径
storePathConsumeQueue =D:\\RocketMQ\\store\\consumequeue
#消息索引存储路径
storePathindex = D:\\RocketMQ\\store\\index
#checkpoint文件存储路径
storeCheckpoint = D:\\RocketMQ\\store\\checkpoint
#abort文件存储路径
abortFile = D:\\RocketMQ\\store\\abort

配置环境变量

ROCKETMQ_HOME=D:\RocketMQ

配置参数

-c D:\RocketMQ\conf\broker.conf

启动成功，检查下数据文件

1.4、控制台

Rocketmq老版本下载

Rocketmq新版本下载

这里下载的新版本rocketmq-dashboard启动项目下面如下

浏览器中输入‘http://localhost:8080’，成功后即可进行管理端查看。

运维页面
你可以修改这个服务使用的namesrv的地址
你可以修改这个服务是否使用VIPChannel(如果你的mq server版本小于3.5.8，请设置不使用)

驾驶舱
查看broker的消息量（总量/5分钟图）
查看单一主题的消息量（总量/趋势图）

集群
查看集群的分布情况：cluster与broker关系、broker
查看broker具体信息/运行信息
查看broker配置信息

主题页面
展示所有的主题，可以通过搜索框进行过滤
筛选普通/重试/死信主题
添加/更新主题
        clusterName 创建在哪几个cluster上
        brokerName 创建在哪几个broker上
        topicName 主题名
        writeQueueNums 写队列数量
        readQueueNums 读队列数量
        perm //2是写 4是读 6是读写
状态查询消息投递状态（投递到哪些broker/哪些queue/多少量等）
路由查看消息的路由（现在你发这个主题的消息会发往哪些broker，对应broker的queue信息）
CONSUMER管理（这个topic都被哪些group消费了，消费情况何如）
topic配置（查看变更当前的配置）
发送消息（向这个主题发送一个测试消息）
重置消费位点(分为在线和不在线两种情况，不过都需要检查重置是否成功)
删除主题（会删除掉所有broker以及namesrv上的主题配置和路由信息）

消费者页面
展示所有的消费组，可以通过搜索框进行过滤
刷新页面/每隔五秒定时刷新页面
按照订阅组/数量/TPS/延迟进行排序
添加/更新消费组
        clusterName 创建在哪几个集群上
        brokerName 创建在哪几个broker上
        groupName 消费组名字
        consumeEnable //是否可以消费 FALSE的话将无法进行消费
        consumeBroadcastEnable //是否可以广播消费
        retryQueueNums //重试队列的大小
        brokerId //正常情况从哪消费
        whichBrokerWhenConsumeSlowly//出问题了从哪消费
终端在线的消费客户端查看，包括版本订阅信息和消费模式
消费详情对应消费组的消费明细查看，这个消费组订阅的所有Topic的消费情况，每个queue对应的消费client查看（包括Retry消息）
配置查看变更消费组的配置
删除在指定的broker上删除消费组

生产者页面
通过Topic和Group查询在线的消息生产者客户端
信息包含客户端主机版本

消息查询页面
根据Topic和时间区间查询由于数据量大最多只会展示2000条，多的会被忽略
根据Topic和Key进行查询
最多只会展示64条
根据消息主题和消息Id进行消息的查询
消息详情可以展示这条消息的详细信息，查看消息对应到具体消费组的消费情况（如果异常，可以查看具体的异常信息）。可以向指定的消费组重发消息

2、消息发送方式

2.1、发送同步消息

同步发送是指消息发送方发出数据后，同步等待，直到收到接收方发回响应之后才发下一个请求。这种可靠性同步地发送方式使用的比较广泛，比如：重要的消息通知，短信通知。

RocketMQ源码中的example模块的org.apache.rocketmq.example.quickstart.Producer

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;public class Producer {public static final int MESSAGE_COUNT = 1000;public static final String PRODUCER_GROUP = "please_rename_unique_group_name";public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";public static final String TOPIC = "TopicTest";public static final String TAG = "TagA";public static void main(String[] args) throws MQClientException, InterruptedException {DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);producer.setNamesrvAddr(DEFAULT_NAMESRVADDR);producer.start();for (int i = 0; i < MESSAGE_COUNT; i++) {try {Message msg = new Message(TOPIC /* Topic */,TAG /* Tag */,("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);SendResult sendResult = producer.send(msg);System.out.printf("%s%n", sendResult);} catch (Exception e) {e.printStackTrace();Thread.sleep(1000);}}producer.shutdown();}
}

发送结果分析

SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFB90000, offsetMsgId=C0A8380100002A9F0000000000046F14, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=3], queueOffset=275]
SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFCD0001, offsetMsgId=C0A8380100002A9F0000000000047003, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=0], queueOffset=275]
SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFD10002, offsetMsgId=C0A8380100002A9F00000000000470F2, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=1], queueOffset=275]
SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFD30003, offsetMsgId=C0A8380100002A9F00000000000471E1, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=2], queueOffset=275]
SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFD50004, offsetMsgId=C0A8380100002A9F00000000000472D0, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=3], queueOffset=276]
SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFD60005, offsetMsgId=C0A8380100002A9F00000000000473BF, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=0], queueOffset=276]
SendResult [sendStatus=SEND_OK, msgId=C0A800AB34BC18B4AAC228E6CFD90006, offsetMsgId=C0A8380100002A9F00000000000474AE, messageQueue=MessageQueue [topic=TopicTest, brokerName=MS-TGOOFNKABBOB, queueId=1], queueOffset=276
....

msgId
消息的全局唯一标识（RocketMQ的ID生成是使用机器IP和消息偏移量的组成），由消息队列 MQ 系统自动生成，唯一标识某条消息。
sendStatus
发送的标识：成功，失败等
queueId
queueId是Topic的分区；Producer发送具体一条消息的时，对应选择的该Topic下的某一个Queue的标识ID。
queueOffset
Message queue是无限长的数组。一条消息进来下标就会涨1,而这个数组的下标就是queueOffset，queueOffset是从0开始递增。

2.2、发送异步消息

异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。消息发送方在发送了一条消息后，不等接收方发回响应，接着进行第二条消息发送。发送方通过回调接口的方式接收服务器响应，并对响应结果进行处理。

RocketMQ源码中的example模块的org.apache.rocketmq.example.quickstart.Producer

package org.apache.rocketmq.example.simple;import java.io.UnsupportedEncodingException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.client.exception.MQClientException;
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;public class AsyncProducer {public static void main(String[] args) throws MQClientException, InterruptedException, UnsupportedEncodingException {DefaultMQProducer producer = new DefaultMQProducer("Jodie_Daily_test");producer.setNamesrvAddr("127.0.0.1:9876");producer.start();// suggest to on enableBackpressureForAsyncMode in heavy traffic, default is falseproducer.setEnableBackpressureForAsyncMode(true);producer.setRetryTimesWhenSendAsyncFailed(0);int messageCount = 100;final CountDownLatch countDownLatch = new CountDownLatch(messageCount);for (int i = 0; i < messageCount; i++) {try {final int index = i;Message msg = new Message("Jodie_topic_1023","TagA","OrderID188","Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));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();}});} catch (Exception e) {e.printStackTrace();}}countDownLatch.await(5, TimeUnit.SECONDS);producer.shutdown();}
}

2.3、单向发送

这种方式主要用在不特别关心发送结果的场景，例如日志发送。单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。此方式发送消息的过程耗时非常短，一般在微秒级别。

package org.apache.rocketmq.example.simple;import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;import java.nio.charset.StandardCharsets;public class OnewayProducer {public static void main(String[] args) throws Exception {//Instantiate with a producer group name.DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");// Specify name server addresses.producer.setNamesrvAddr("localhost:9876");//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("TopicTest" /* Topic */,"TagA" /* Tag */,("Hello RocketMQ " +i).getBytes(StandardCharsets.UTF_8) /* Message body */);//Call send message to deliver message to one of brokers.producer.sendOneway(msg);}//Wait for sending to completeThread.sleep(5000);producer.shutdown();}
}

3、消息消费方式

3.1、负载均衡模式（集群消费）

消费者采用负载均衡方式消费消息，一个分组(Group)下的多个消费者共同消费队列消息，每个消费者处理的消息不同。一个Consumer Group中的各个Consumer实例分摊去消费消息，即一条消息只会投递到一个Consumer Group下面的一个实例。例如某个Topic有3个队列，其中一个Consumer Group 有 3 个实例，那么每个实例只消费其中的1个队列。集群消费模式是消费者默认的消费方式。

package org.apache.rocketmq.example.quickstart;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
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.consumer.ConsumeFromWhere;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;public class Consumer {public static final String CONSUMER_GROUP = "please_rename_unique_group_name_4";public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";public static final String TOPIC = "TopicTest";public static void main(String[] args) throws MQClientException {//实例化消息生产者，指定组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);//指定Namesrv地址信息consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);//订阅Topicconsumer.subscribe(TOPIC, "*");//负载均衡模式消费(可以不设置，默认就是负载均衡模式)consumer.setMessageModel(MessageModel.CLUSTERING);//注册回调函数，处理消息consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});consumer.start();//启动消费者System.out.printf("Consumer Started.%n");}
}

3.2、广播消费

广播消费模式中消息将对一个Consumer Group下的各个Consumer实例都投递一遍。即使这些 Consumer属于同一个Consumer Group，消息也会被Consumer Group 中的每个Consumer都消费一次。实际上，是一个消费组下的每个消费者实例都获取到了topic下面的每个Message Queue去拉取消费。所以消息会投递到每个消费者实例。

package org.apache.rocketmq.example.quickstart;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
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.consumer.ConsumeFromWhere;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;public class Consumer {public static final String CONSUMER_GROUP = "please_rename_unique_group_name_4";public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";public static final String TOPIC = "TopicTest";public static void main(String[] args) throws MQClientException {//实例化消息生产者，指定组名DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);//指定Namesrv地址信息consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);//订阅Topicconsumer.subscribe(TOPIC, "*");//广播消费模式consumer.setMessageModel(MessageModel.BROADCASTING);//注册回调函数，处理消息consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;});consumer.start();//启动消费者System.out.printf("Consumer Started.%n");}
}

消息消费时的权衡
负载均衡模式：适用场景&注意事项
        消费端集群化部署，每条消息只需要被处理一次。
        由于消费进度在服务端维护，可靠性更高。
        集群消费模式下，每一条消息都只会被分发到一台机器上处理。如果需要被集群下的每一台机器都处理，请使用广播模式。
        集群消费模式下，不保证每一次失败重投的消息路由到同一台机器上，因此处理消息时不应该做任何确定性假设。
广播模式：适用场景&注意事项
        每条消息都需要被相同逻辑的多台机器处理。
        消费进度在客户端维护，出现重复的概率稍大于集群模式。
        广播模式下，消息队列 RocketMQ 保证每条消息至少被每台客户端消费一次，但是并不会对消费失败的消息进行失败重投，因此业务方需要关注消费失败的情况。
        广播模式下，客户端每一次重启都会从最新消息消费。客户端在被停止期间发送至服务端的消息将会被自动跳过，请谨慎选择。
        广播模式下，每条消息都会被大量的客户端重复处理，因此推荐尽可能使用集群模式。
目前仅 Java 客户端支持广播模式。
        广播消费模式下不支持顺序消息。
        广播消费模式下不支持重置消费位点。
        广播模式下服务端不维护消费进度，所以消息队列 RocketMQ 控制台不支持消息堆积查询、消息堆积报警和订阅关系查询功能。

不是你觉的悟到的东西给了你，你也接不住！

干我们这行，啥时候懈怠，就意味着长进的停止，长进的停止就意味着被淘汰，只能往前冲，直到凤凰涅槃的一天！

一、RocketMQ

0、RocketMQ的产品发展

1、RocketMQ安装

1.1、windows下的安装

注意事项

1.2、Linux下的安装

1.3、源码的安装

1.4、控制台

2、消息发送方式

2.1、发送同步消息

2.2、发送异步消息

2.3、单向发送

3、消息消费方式

3.1、负载均衡模式（集群消费）

3.2、广播消费

相关文章：