linux安装RocketMQ

简介

是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式特点。 Producer、Consumer队列都可以分布式。

Producer向一些队列轮流发送消息，队列集合称为 Topic，Consumer 如果做广播消费，则一个consumer实例消费这个Topic 对应的所有队列，如果做集群消费，则多个Consumer 实例平均消费这个topic对应的队列集合。（默认是集群消费） 能够保证严格的消息顺序（因为性能原因，不能保证消息不重复，因为总有网络不可达的情况发生，需业务端保证）。

提供丰富的消息拉取模式

高效的订阅者水平扩展能力

实时的消息订阅机制

亿级消息堆积能力

较少的依赖

支持集群部署，保证了高可用，数据不会丢失。

学习资料

Apache 上开源官方地址：https://rocketmq.apache.org/ GitHub 托管地址：https://github.com/apache/rocketmq 阿里官方的介绍文档：http://jm.taobao.org/2017/01/12/rocketmq-quick-start-in-10-minutes/ Apache 官方提供的 4.3.0 版本的 “快速入门” 文档：https://rocketmq.apache.org/docs/quick-start/

下载

Apache 官方下载网址：https://www.apache.org/dyn/closer.cgi?path=rocketmq/4.3.0/rocketmq-all-4.3.0-source-release.zip 也可以使用git下载 git clone https://github.com/apache/rocketmq.git

安装篇

[root@localhost hbk]# ll /root/hbk/rocketmq-all-4.3.0-source-release.zip -rw-r--r-- 1 root root 1605734 10月 31 10:20 /root/hbk/rocketmq-all-4.3.0-source-release.zip

解压

[root@localhost hbk]# unzip rocketmq-all-4.3.0-source-release.zip [root@localhost hbk]# cd rocketmq-all-4.3.0/ [root@localhost rocketmq-all-4.3.0]# ll 总用量 56 drwxr-xr-x 3 root root 32 7月 26 2018 broker -rw-r--r-- 1 root root 997 7月 26 2018 BUILDING drwxr-xr-x 3 root root 32 7月 26 2018 client drwxr-xr-x 3 root root 32 7月 26 2018 common -rw-r--r-- 1 root root 1997 7月 26 2018 CONTRIBUTING.md -rw-r--r-- 1 root root 271 7月 26 2018 DEPENDENCIES drwxr-xr-x 2 root root 34 7月 26 2018 dev drwxr-xr-x 5 root root 143 7月 26 2018 distribution drwxr-xr-x 3 root root 32 7月 26 2018 example drwxr-xr-x 3 root root 32 7月 26 2018 filter -rw-r--r-- 1 root root 11365 7月 26 2018 LICENSE drwxr-xr-x 3 root root 32 7月 26 2018 logappender drwxr-xr-x 3 root root 32 7月 26 2018 logging drwxr-xr-x 3 root root 32 7月 26 2018 namesrv -rw-r--r-- 1 root root 168 7月 26 2018 NOTICE drwxr-xr-x 3 root root 32 7月 26 2018 openmessaging -rw-r--r-- 1 root root 23729 7月 26 2018 pom.xml -rw-r--r-- 1 root root 2426 7月 26 2018 README.md drwxr-xr-x 3 root root 32 7月 26 2018 remoting drwxr-xr-x 3 root root 32 7月 26 2018 srvutil drwxr-xr-x 3 root root 32 7月 26 2018 store drwxr-xr-x 3 root root 74 7月 26 2018 style drwxr-xr-x 3 root root 32 7月 26 2018 test drwxr-xr-x 3 root root 32 7月 26 2018 tools

使用maven来编译整个项目

[root@localhost rocketmq-all-4.3.0]# mvn -Prelease-all -DskipTests clean install -U

接着就会从 Maven 中央仓库下载项目依赖的 jar 包编译部署，这可能需要几分钟的时间

显示build success说明安装好了

[INFO] Reactor Summary: [INFO] [INFO] Apache RocketMQ 4.3.0 4.3.0 ........................ SUCCESS [ 40.841 s] [INFO] rocketmq-logging 4.3.0 ............................. SUCCESS [ 27.985 s] [INFO] rocketmq-remoting 4.3.0 ............................ SUCCESS [ 17.864 s] [INFO] rocketmq-common 4.3.0 .............................. SUCCESS [ 27.601 s] [INFO] rocketmq-client 4.3.0 .............................. SUCCESS [ 34.164 s] [INFO] rocketmq-store 4.3.0 ............................... SUCCESS [ 20.021 s] [INFO] rocketmq-srvutil 4.3.0 ............................. SUCCESS [ 4.925 s] [INFO] rocketmq-filter 4.3.0 .............................. SUCCESS [ 9.858 s] [INFO] rocketmq-broker 4.3.0 .............................. SUCCESS [ 21.414 s] [INFO] rocketmq-tools 4.3.0 ............................... SUCCESS [ 16.750 s] [INFO] rocketmq-namesrv 4.3.0 ............................. SUCCESS [ 7.363 s] [INFO] rocketmq-logappender 4.3.0 ......................... SUCCESS [ 5.737 s] [INFO] rocketmq-openmessaging 4.3.0 ....................... SUCCESS [ 4.843 s] [INFO] rocketmq-example 4.3.0 ............................. SUCCESS [ 5.347 s] [INFO] rocketmq-test 4.3.0 ................................ SUCCESS [ 23.286 s] [INFO] rocketmq-distribution 4.3.0 4.3.0 .................. SUCCESS [04:46 min] [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO] ------------------------------------------------------------------------ [INFO] Total time: 09:18 min [INFO] Finished at: 2019-10-31T11:59:16+08:00 [INFO] ------------------------------------------------------------------------

RocketMQ 操作

进入RocketMQ安装目录下的distribution/target/apache-rocketmq 目录 先启动namesrv

cd distribution/target/apache-rocketmq nohup sh bin/mqnamesrv & tail -f bin/nohup.out # 启动成功显示 The Name Server boot success. serializeType=JSON

然后启动broker

nohup sh bin/mqbroker -n localhost:9876 & tail -f bin/nohup.out # 启动成功显示 The broker[**, **.**.**.**:10911] boot success. serializeType=JSON and name server is localhost:9876

测试发送和接收消息

在发送和接收之前我们需要告诉客户端name servers的地址，RocketMQ提供了很多种方式来实现，为了演示方便我们使用环境变量NAMESRV_ADDR。

export NAMESRV_ADDR=localhost:9876 sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer # 可以看到有很多条消息已经发送成功了 SendResult [sendStatus=SEND_OK, msgId= ... SendResult [sendStatus=SEND_OK, msgId= ... SendResult [sendStatus=SEND_OK, msgId= ... sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer # 可以看到刚才发送的消息已经成功被消费者消费了 ConsumeMessageThread_%d Receive New Messages: [MessageExt... ConsumeMessageThread_%d Receive New Messages: [MessageExt... ConsumeMessageThread_%d Receive New Messages: [MessageExt...

关闭服务

sh bin/mqshutdown broker # 关闭成功后显示 The mqbroker(8916) is running... Send shutdown request to mqbroker(8916) OK sh bin/mqshutdown namesrv # 关闭成功后显示 The mqnamesrv(8892) is running... Send shutdown request to mqnamesrv(8892) OK

遇到的问题

启动服务时报内存不足 执行

nohup sh mqnamesrv & 时，报错

# There is insufficient memory for the Java Runtime Environment to continue. # Native memory allocation (mmap) failed to map 2147483648 bytes for committing reserved memory. # An error report file with more information is saved as: # /home/wangjun/rocketmq/distribution/bin/hs_err_pid8300.log

解决方案 修改rocketmq/distribution/target/apache-rocketmq/bin/runserver.sh和runbroker.sh。

将

JAVA_OPT="${JAVA_OPT} -server -Xms4g -Xmx4g -Xmn2g -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

改为：

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn125m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

可以看出之前的默认设置是4g内存，如果你的机器没有这么大只是自己搭建着玩就把它设置小一点就行了。

顺便将tools.sh的内存也改成256m，不然运行消息的发送和接收的demo的时候也会报错。

客户端测试遇到的问题

一、No route info of this topic 解决方案

在启动broker时加上 autoCreateTopicEnable=true，如果是用broker.conf启动，或者集群使用broker-*启动，形如：

nohup sh /oa/other/rocketmq/bin/mqbroker -c /oa/other/rocketmq/conf/2m-noslave/broker-a.properties >/dev/null 2>&1 &

broker-a.properties配置文件中加入

autoCreateTopicEnable=true

二、connect to ip:10909 failed

产生的原因：

rocketmq默认开启了vip通道

解决方案：

在客户端代码层面加入

producer.setVipChannelEnabled(false); consumer.setVipChannelEnabled(false);

附录

除了上面几个命令之外，还有如下一些较常用的命令，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

测试代码：

package com.hbk.maven_test; import java.util.List; 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.producer.DefaultMQProducer; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageExt; /** * RocketMQ测试学习 * * @author 黄宝康 2019年10月31日 上午11:33:15 */ public class RocketMQTest { public static void produceMsg() throws Exception { // 测试生产者生成消息 DefaultMQProducer producer = new DefaultMQProducer("producer_hbk");// 生产组 // producer_hbk // 指定NameServer地址 producer.setNamesrvAddr("192.168.86.140:9876");// 多个使用;分隔 // 192.168.86.134:9876;192.168.86.135:9876 // producer.setVipChannelEnabled(false); /** * Producer对象在使用之前必须要调用start初始化，初始化一次即可 注意：切记不可以在每次发送消息时，都调用start方法 */ producer.start(); for (int i = 0; i < 100; i++) { try { Message msg = new Message("TopicTest", ("测试RocketMQ" + i).getBytes("UTF-8")); SendResult sendResult = producer.send(msg); System.out.printf("%s%n", sendResult); } catch (Exception e) { e.printStackTrace(); } } producer.shutdown(); } public static void consumerMsg() throws Exception { DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_hbk");// 消费组 //指定NameServer地址，多个地址以 ;隔开 consumer.setNamesrvAddr("192.168.86.140:9876"); /** * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费 * 如果非第一次启动，那么按照上次消费的位置继续消费 */ consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); consumer.subscribe("TopicTest", "*"); consumer.registerMessageListener(new MessageListenerConcurrently() { public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { try { for(MessageExt msg :msgs){ System.out.println("MessageBody: "+new String(msg.getBody(),"UTF-8")); } } catch (Exception e) { e.printStackTrace(); return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 稍后再试 } return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 消费成功 } }); consumer.start(); System.out.println("消费开始"); } public static void main(String[] args) throws Exception { produceMsg(); consumerMsg(); } }