rocketmq安装部署过程（4.0.0版本）

曾经终败给现在 2022-05-09 03:40 120阅读 0赞

*  **准备工作**

![915691-20170727092707343-1474359845.png][]

*  **3个虚拟机节点的构成如下** **：**

![915691-20170706155413253-1226867687.png][]

*  **安装步骤**

![915691-20170727092812203-808976318.png][]

*  **操作过程**

1、安装包已经上传至其中1个节点。

![915691-20170727093212984-406200655.png][]

2、解压缩安装包

命令：unzip rocketmq-all-4.0.0-incubating-bin-release.zip

解压缩之后如下：

![915691-20170727093505750-38487954.png][]

3、 我这里将解压缩之后的文件夹移动了位置，并修改了名字，以便后续操作。

命令： mv /home/hadmin/software/apache-rocketmq-all/ /home/hadmin/rocketmq

移动之后路径如下：

![915691-20170727093839015-965948135.png][]

4、修改配置文件

我这里已经将配置文件提前准备好了，只呈现以下配置文件的结果。

默认配置可以参考源码：

[https://github.com/apache/rocketmq/blob/master/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java][https_github.com_apache_rocketmq_blob_master_store_src_main_java_org_apache_rocketmq_store_config_MessageStoreConfig.java]

Tips1：

autoCreateTopicEnable，建议线下开启，线上关闭。

Tips2：

同一个机器上有多个Broker时，各个Broker的文件路径要独立。

■节点1（192.168.6.3）配置文件：

《broker-a-m.properties》

![复制代码][copycode.gif]

brokerClusterName=post
    brokerName=broker-a
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=0
    listenPort=10911
    brokerIP1=192.168.6.3
    deleteWhen=04
    fileReservedTime=72
    brokerRole=ASYNC_MASTER
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-a-m
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-a-m
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

![复制代码][copycode.gif]

《broker-c-s.properties》

![复制代码][copycode.gif]

brokerClusterName=post
    brokerName=broker-c
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=1
    listenPort=10920
    brokerIP1=192.168.6.3
    deleteWhen=04
    fileReservedTime=72
    brokerRole=SLAVE
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-c-s
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-c-s
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

![复制代码][copycode.gif]

■节点2（192.168.6.4）配置文件：

《broker-a-s.properties》

![复制代码][copycode.gif]

brokerClusterName=post
    brokerName=broker-a
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=1
    listenPort=10920
    deleteWhen=04
    brokerIP1=192.168.6.4
    fileReservedTime=72
    brokerRole=SLAVE
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-a-s
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-a-s
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

![复制代码][copycode.gif]

《broker-b-m.properties》

![复制代码][copycode.gif]

brokerClusterName=post
    brokerName=broker-b
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=0
    listenPort=10911
    brokerIP1=192.168.6.4
    deleteWhen=04
    fileReservedTime=72
    brokerRole=ASYNC_MASTER
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-b-m
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-b-m
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

![复制代码][copycode.gif]

■节点3（192.168.6.5）配置文件：

《broker-b-s.properties》

![复制代码][copycode.gif]

brokerClusterName=post
    brokerName=broker-b
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=1
    listenPort=10920
    brokerIP1=192.168.6.5
    deleteWhen=04
    fileReservedTime=72
    brokerRole=SLAVE
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-b-s
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-b-s
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

![复制代码][copycode.gif]

《broker-c-m.properties》

![复制代码][copycode.gif]

brokerClusterName=post
    brokerName=broker-c
    namesrvAddr=192.168.6.3:9876;192.168.6.4:9876
    brokerId=0
    listenPort=10911
    brokerIP1=192.168.6.5
    deleteWhen=04
    fileReservedTime=72
    brokerRole=ASYNC_MASTER
    storePathRootDir=/home/hadmin/data/rocketmq/rootdir-c-m
    storePathCommitLog=/home/hadmin/data/rocketmq/commitlog-c-m
    defaultTopicQueueNums=4
    autoCreateTopicEnable=false
    autoCreateSubscriptionGroup=true
    flushDiskType=ASYNC_FLUSH

![复制代码][copycode.gif]

5、启动nameserver

启动命令：nohup sh /home/hadmin/rocketmq/bin/mqnamesrv >/home/hadmin/rocketmq/logs/mqnamesrv.log 2>&1 &

注意：我这里将启动日志重定义到了logs路径下，需要提前手动创建logs文件夹，以便于统一管理日志，方便查看。

创建文件夹命令：mkdir /home/hadmin/rocketmq/logs

下图中展示了启动命令，日志中可以看到NameServer成功启动的日志。

■NameServer - 节点1

![915691-20170727095000437-1970796133.png][]

■NameServer = 节点2

![915691-20170727095304937-124024529.png][]

6、启动Broker-a（Master位于节点1、Slave位于节点2）

broker-a分为Master和Slave，分别位于节点1和节点2上，需要分别启动。

注意：需要根据启动角色，为broker指定一个配置文件。

■broker-a的master - 节点1

命令： nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-a-m.properties>/home/hadmin/rocketmq/logs/broker-a-m.log 2>&1 &

![915691-20170727095558312-1317840367.png][]

■broker-a的slave - 节点2

命令：nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-a-s.properties>/home/hadmin/rocketmq/logs/broker-a-s.log 2>&1 &

![915691-20170727095905234-2601642.png][]

■验证broker-a：

broker-a启动结束，这时候可以使用命令查看一下rocketmq集群状态。

命令：sh /home/hadmin/rocketmq/bin/mqadmin clusterList -n 192.168.6.3:9876

![915691-20170727100041218-383690385.png][]

7、启动broker-b（Master位于节点2,、Slave位于节点3）

■broker-b的Master - 节点2

命令：nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-b-m.properties >/home/hadmin/rocketmq/logs/broker-b-m.log 2>&1 &

![915691-20170727100320203-50626500.png][]

■broker-b的Slave - 节点3

命令：nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-b-s.properties >/home/hadmin/rocketmq/logs/broker-b-s.log 2>&1 &

![915691-20170727100513984-2054874809.png][]

■验证broker-b

命令：sh /home/hadmin/rocketmq/bin/mqadmin clusterList -n 192.168.6.3:9876

![915691-20170727100701218-1415347556.png][]

8、 启动broker-b（Master位于节点3,、Slave位于节点1）

■broker-c的Master - 节点3

命令：nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-c-m.properties >/home/hadmin/rocketmq/logs/broker-c-m.log 2>&1 &

![915691-20170727101141203-248271416.png][]

■broker-c的Slave - 节点1

命令：nohup sh /home/hadmin/rocketmq/bin/mqbroker -c /home/hadmin/rocketmq/conf/broker-c-s.properties >/home/hadmin/rocketmq/logs/broker-c-s.log 2>&1 &

![915691-20170727101247718-554647041.png][]

■验证broker-c

命令：sh /home/hadmin/rocketmq/bin/mqadmin clusterList -n 192.168.6.3:9876

![915691-20170727101335750-1093080270.png][]

*  **问题1：**

启动broker的时候提示内存不够的错误。

![915691-20170706155359409-887169032.png][]

解决方法：由于个人电脑配置不够，无法为虚拟机申请更大的内存。所以，采用修改broker启动内存的方式解决了。

修改文件路径：\{ROCKET\_HOME\}/bin/runbroker.sh

修改前：JAVA\_OPT="$\{JAVA\_OPT\} -server -Xms8g -Xmx8g -Xmn4g"

修改后：JAVA\_OPT="$\{JAVA\_OPT\} -server -Xms1024m -Xmx1536m -Xmn1546m"

如图所示：

![915691-20180207105703263-167452212.png][]

重新启动broker，日志中没有错误，jps进程中也可以看到broker正常启动。

![915691-20170706160537862-246052289.png][]

![915691-20170706160549894-1038724701.png][]

*  **问题2：**

在同一台机器上启动多个broker的时候提示如下错误。

![915691-20170706161231878-733200846.png][]

问题原因：

按照本文最开始的额图片所示，同一台机器上会存在两个Broker，如果不进行特殊指定，broker的默认端口是10911。

所以一台机器上启动两个broker时，第二个broker就会出现端口被占用的错误。

解决办法：

修改rocketmq的配置文件，增加listenPort配置。配置之后如下所示：

→同一台机器的配置文件

![915691-20170706163514269-1842780138.png][]

→broker-a-m.properties

![915691-20170706163404909-482356288.png][]

→broker-c-s.properties

![915691-20170706163437815-971512136.png][]

使用jps查看一下进程是否有问题

![915691-20170706163603315-2142588543.png][]

在查看一下启动日志是否有问题

![915691-20170706163633659-764745035.png][]

最后使用clusterList命令来验证一下集群健康状态。

命令：sh bin/mqadmin clusterList -n 192.168.6.3:9876

![915691-20170706163737456-960015278.png][]

成功实现了，如本文最开始图片所示的安装部署。

*  使用程序测试

编写了java程序代码尝试向集群中生产消息，程序代码如下：

需要注意的是，如果autoCreateTopicEnable=false的时候，需要自己手动创建Topic。

命令样例：sh $MQ\_HOME/bin/mqadmin updateTopic -c post -n "192.168.1.80:9876;192.168.1.81:9876" -t QCH2 -r 4 -w 4

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 </td> 
 <td> <code>package</code>&nbsp;<code>post;</code> &nbsp; <code>import</code>&nbsp;<code>org.apache.rocketmq.client.exception.MQBrokerException;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.client.exception.MQClientException;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.client.producer.DefaultMQProducer;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.common.message.Message;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.remoting.exception.RemotingException;</code> &nbsp; <code>import</code>&nbsp;<code>java.util.UUID;</code> &nbsp; <code>public</code>&nbsp;<code>class</code>&nbsp;<code>ProducerTest { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>static</code>&nbsp;<code>DefaultMQProducer producer =&nbsp;</code><code>null</code><code>;</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>static</code>&nbsp;<code>void</code>&nbsp;<code>main(String[] args) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>System.out.print(</code><code>"[----------]Start"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>boolean</code>&nbsp;<code>result =&nbsp;</code><code>false</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>ProducerStart();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>SendMessage(</code><code>"qch_20170706"</code><code>,&nbsp;</code><code>"hello rocketmq!"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code><code>finally</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer.shutdown();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>System.out.print(</code><code>"[----------]Succeed"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>static</code>&nbsp;<code>boolean</code>&nbsp;<code>ProducerStart() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer =&nbsp;</code><code>new</code>&nbsp;<code>DefaultMQProducer(</code><code>"pro_qch_test"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer.setNamesrvAddr(</code><code>"192.168.6.3:9876;192.168.6.4:9876"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer.setInstanceName(UUID.randomUUID().toString());</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer.setVipChannelEnabled(</code><code>false</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer.start();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(MQClientException e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>e.printStackTrace();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>false</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>true</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> &nbsp; <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>private</code>&nbsp;<code>static</code>&nbsp;<code>boolean</code>&nbsp;<code>SendMessage(String topic, String str) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>Message msg =&nbsp;</code><code>new</code>&nbsp;<code>Message(topic, str.getBytes());</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>producer.send(msg);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(MQClientException | RemotingException | MQBrokerException | InterruptedException e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>e.printStackTrace();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>false</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>true</code><code>;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

运行之后，日志中提示下面的错误：

![915691-20170706182536034-1662762351.png][]

问题原因：

因为broker部署在虚拟机，并且虚拟双网卡，client无法正常连接服务端。

解决方法：

可以在broker的配置文件中配置brokerIP1（本机IP）属性。

修改后配置文件如下图所示：

![915691-20170706191352315-1111822596.png][]

修改之后，重新启动rocketmq集群，运行生产者程序，确认正常结束。

![915691-20170706192000597-398857394.png][]

然后，有尝试这编写消费者代码，验证是否可以正常消费。结果正常，这里贴一下代码及结果日志。

<table> 
 <tbody> 
 <tr> 
 <td> 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 </td> 
 <td> <code>package</code>&nbsp;<code>post;</code> &nbsp; <code>import</code>&nbsp;<code>org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.common.consumer.ConsumeFromWhere;</code> <code>import</code>&nbsp;<code>org.apache.rocketmq.common.message.MessageExt;</code> &nbsp; <code>import</code>&nbsp;<code>java.util.List;</code> <code>import</code>&nbsp;<code>java.util.UUID;</code> &nbsp; <code>public</code>&nbsp;<code>class</code>&nbsp;<code>ConsumerTest { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>static</code>&nbsp;<code>void</code>&nbsp;<code>main(String[] args) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>DefaultMQPushConsumer consumer =&nbsp;</code><code>new</code>&nbsp;<code>DefaultMQPushConsumer(</code><code>"con_qch_test"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.setInstanceName(UUID.randomUUID().toString());</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.setConsumeMessageBatchMaxSize(</code><code>32</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.setNamesrvAddr(</code><code>"192.168.6.3:9876;192.168.6.4:9876"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.registerMessageListener(</code><code>new</code>&nbsp;<code>MessageListenerConcurrently() { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>@Override</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>public</code>&nbsp;<code>ConsumeConcurrentlyStatus consumeMessage(</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>List&lt;MessageExt&gt; list, ConsumeConcurrentlyContext consumeConcurrentlyContext) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>for</code><code>(MessageExt me : list) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>System.out.print(</code><code>new</code>&nbsp;<code>String(me.getBody()));</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>return</code>&nbsp;<code>ConsumeConcurrentlyStatus.CONSUME_SUCCESS;</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>});</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>try</code>&nbsp;<code>{ </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.subscribe(</code><code>"qch_20170706"</code><code>,&nbsp;</code><code>"*"</code><code>);</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>consumer.start();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}&nbsp;</code><code>catch</code>&nbsp;<code>(Exception e) { </code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>e.printStackTrace();</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>&nbsp;&nbsp;&nbsp;&nbsp;</code><code>}</code> <code>}</code> </td> 
 </tr> 
 </tbody> 
</table>

![915691-20170706193415050-1188552277.png][]

样例代码上传到了git上了，有需要的可以去参考。

[https://github.com/quchunhui/rocketmq\_sample][https_github.com_quchunhui_rocketmq_sample]

\--END--

转载于：[https://www.cnblogs.com/quchunhui/p/7126868.html][https_www.cnblogs.com_quchunhui_p_7126868.html]

[915691-20170727092707343-1474359845.png]: /images/20220509/0a310c9452244ea19b10cea5ec494be7.png
[915691-20170706155413253-1226867687.png]: /images/20220509/5215f189a7d54d7badbc8257e765ad90.png
[915691-20170727092812203-808976318.png]: /images/20220509/0be0f05895c943999064f80c9643cfcb.png
[915691-20170727093212984-406200655.png]: /images/20220509/6587b831870c4c158e87ea5b9e7bcc57.png
[915691-20170727093505750-38487954.png]: /images/20220509/6f1582004908495eb047b2fe8bd76571.png
[915691-20170727093839015-965948135.png]: /images/20220509/22a038280247426d8914155b72d82e97.png
[https_github.com_apache_rocketmq_blob_master_store_src_main_java_org_apache_rocketmq_store_config_MessageStoreConfig.java]: https://github.com/apache/rocketmq/blob/master/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
[copycode.gif]: /images/20220509/b05bc3e445c2446ab7ad285d5f7347ad.png
[915691-20170727095000437-1970796133.png]: /images/20220509/33d00f1b5d0d41c38fba231f6805dfa5.png
[915691-20170727095304937-124024529.png]: /images/20220509/eed0a0ecdce54b1482ffd351fe1dd435.png
[915691-20170727095558312-1317840367.png]: /images/20220509/4d5ed473eb194e44bf6120fea631f3e6.png
[915691-20170727095905234-2601642.png]: /images/20220509/f2ebbdbfd0124e8292a11ea548d7155f.png
[915691-20170727100041218-383690385.png]: /images/20220509/bdadee553d1b409f9bd1fb43f64a7cd2.png
[915691-20170727100320203-50626500.png]: /images/20220509/6c948968bb9b4c03b2a3a08ee670fae8.png
[915691-20170727100513984-2054874809.png]: /images/20220509/9c05b3092a1542cfabcf546fab0dd4fd.png
[915691-20170727100701218-1415347556.png]: /images/20220509/f4070c6a726a45e992d01a481b1b5d40.png
[915691-20170727101141203-248271416.png]: /images/20220509/2c8d014f8b344aecaabbbf5bc26b44b0.png
[915691-20170727101247718-554647041.png]: /images/20220509/fb030453bd05452c86c4a2f69115b352.png
[915691-20170727101335750-1093080270.png]: /images/20220509/ab1f3c42df4845f38bfdb49d5bdcfa80.png
[915691-20170706155359409-887169032.png]: /images/20220509/32dc0a3211884a46b088caedc736ae67.png
[915691-20180207105703263-167452212.png]: /images/20220509/026f63a730a64b699d0221eda04c3065.png
[915691-20170706160537862-246052289.png]: /images/20220509/11babc577a554dc69c432c139fb9cfa4.png
[915691-20170706160549894-1038724701.png]: /images/20220509/ca0128204eb84509b65c9bc044507fed.png
[915691-20170706161231878-733200846.png]: /images/20220509/3e89ecd3e07a42c3a84c652a05300c88.png
[915691-20170706163514269-1842780138.png]: /images/20220509/7c60a4ca5b0a4091a0814b9b5968e7f8.png
[915691-20170706163404909-482356288.png]: /images/20220509/16106053582c48bf9ca8b052f438b431.png
[915691-20170706163437815-971512136.png]: /images/20220509/f1462af19e754a228ba61a9f4e4b426e.png
[915691-20170706163603315-2142588543.png]: /images/20220509/6d869479200a4088944fd534681a33e4.png
[915691-20170706163633659-764745035.png]: /images/20220509/5c73eae9b0ae41268cd07a12f89ecfd2.png
[915691-20170706163737456-960015278.png]: /images/20220509/00cadda8120c4f5e8d54fd55f909599d.png
[915691-20170706182536034-1662762351.png]: /images/20220509/4970cdc6a664456e97117855b478794c.png
[915691-20170706191352315-1111822596.png]: /images/20220509/4ce4965cdfd2452388bb903be09b3f69.png
[915691-20170706192000597-398857394.png]: /images/20220509/dee875d63bd04e01ba1b4a8ce0f4d091.png
[915691-20170706193415050-1188552277.png]: /images/20220509/e2ffde99de3c45788213f6fbf91db4d0.png
[https_github.com_quchunhui_rocketmq_sample]: https://github.com/quchunhui/rocketmq_sample
[https_www.cnblogs.com_quchunhui_p_7126868.html]: https://www.cnblogs.com/quchunhui/p/7126868.html