官方指导命令:

 > export NAMESRV_ADDR=localhost:9876
 > sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
 SendResult [sendStatus=SEND_OK, msgId= ...

如下所示官方提供这个例子属于生产者,用于发送消息,运行之后会发送大量的消息,之后就会退出

[root@localhost apache-rocketmq]# export NAMESRV_ADDR=localhost:9876
[root@localhost apache-rocketmq]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0
03:05:13.152 [main] DEBUG i.n.u.i.l.InternalLoggerFactory - Using SLF4J as the default logging framework
SendResult [sendStatus=SEND_OK, msgId=C0A83A8120207D4991AD2F02D2280000, offsetMsgId=C0A83A8100002A9F00000000000AFAC8, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=0], queueOffset=1000]
SendResult [sendStatus=SEND_OK, msgId=C0A83A8120207D4991AD2F02D2740001, offsetMsgId=C0A83A8100002A9F00000000000AFB7A, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=1], queueOffset=1000]
SendResult [sendStatus=SEND_OK, msgId=C0A83A8120207D4991AD2F02D27A0002, offsetMsgId=C0A83A8100002A9F00000000000AFC2C, messageQueue=MessageQueue [topic=TopicTest, brokerName=localhost.localdomain, queueId=2], queueOffset=1000]

.........................

接收消息
官方指导命令:

> sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
 ConsumeMessageThread_%d Receive New Messages: [MessageExt...
上面的消息发送完毕之后就会退出,在同一窗口中可以使用消费者类来进行接收消息,如下所示,显然是多线程的
[root@localhost apache-rocketmq]# sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
ConsumeMessageThread_3 Receive New Messages: [MessageExt [queueId=0, storeSize=180, queueOffset=1031, sysFlag=0, bornTimestamp=1533841514766, bornHost=/192.168.58.129:46816, storeTimestamp=1533841514767, storeHost=/192.168.58.129:10911, msgId=C0A83A8100002A9F00000000000B518A, commitLogOffset=741770, bodyCRC=895437781, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1250, CONSUME_START_TIME=1533841719674, UNIQ_KEY=C0A83A8120207D4991AD2F02D50E007C, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 49, 50, 52], transactionId='null'}]] 
ConsumeMessageThread_2 Receive New Messages: [MessageExt [queueId=0, storeSize=180, queueOffset=1030, sysFlag=0, bornTimestamp=1533841514746, bornHost=/192.168.58.129:46816, storeTimestamp=1533841514747, storeHost=/192.168.58.129:10911, msgId=C0A83A8100002A9F00000000000B4EBA, commitLogOffset=741050, bodyCRC=842174412, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1250, CONSUME_START_TIME=1533841719669, UNIQ_KEY=C0A83A8120207D4991AD2F02D4FA0078, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 49, 50, 48], transactionId='null'}]] 
ConsumeMessageThread_20 Receive New Messages: [MessageExt [queueId=0, storeSize=179, queueOffset=1019, sysFlag=0, bornTimestamp=1533841514511, bornHost=/192.168.58.129:46816, storeTimestamp=1533841514515, storeHost=/192.168.58.129:10911, msgId=C0A83A8100002A9F00000000000B2FE2, commitLogOffset=733154, bodyCRC=884882597, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=1250, CONSUME_START_TIME=1533841719680, UNIQ_KEY=C0A83A8120207D4991AD2F02D40F004C, WAIT=true, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 55, 54], transactionId='null'}]] 
...............
关闭服务器

常用命令
除了上面几个命令之外,还有如下一些较常用的命令,ip请以实际为准:

查看集群情况: ./mqadmin clusterList -n 127.0.0.1:9876
查看 broker 状态: ./mqadmin brokerStatus -n 127.0.0.1:9876 -b 172.20.1.138:10911
查看 topic 列表: ./mqadmin topicList -n 127.0.0.1:9876
查看 topic 状态: ./mqadmin topicStatus -n 127.0.0.1:9876 -t MyTopic (换成想查询的 topic)
查看 topic 路由: ./mqadmin topicRoute -n 127.0.0.1:9876 -t MyTopic

文档更新时间: 2020-02-04 09:31   作者:月影鹏鹏