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

时间 2019-12-12

标签 rocketmq 安装部署过程 4.0.0 版本繁體版

原文原文链接

准备工做

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

安装步骤

操做过程

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

二、解压缩安装包git

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

解压缩以后以下：apache

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

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

移动以后路径以下：post

四、修改配置文件测试

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

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

《broker-a-m.properties》

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
storePathCommitLog=/home/hadmin/data/rocketmq/commitlog
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
flushDiskType=ASYNC_FLUSH

《broker-c-s.properties》

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
storePathCommitLog=/home/hadmin/data/rocketmq/commitlog
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
flushDiskType=ASYNC_FLUSH

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

《broker-a-s.properties》

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
storePathCommitLog=/home/hadmin/data/rocketmq/commitlog
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
flushDiskType=ASYNC_FLUSH

《broker-b-m.properties》

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
storePathCommitLog=/home/hadmin/data/rocketmq/commitlog
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
flushDiskType=ASYNC_FLUSH

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

《broker-b-s.properties》

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
storePathCommitLog=/home/hadmin/data/rocketmq/commitlog
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
flushDiskType=ASYNC_FLUSH

《broker-c-m.properties》

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
storePathCommitLog=/home/hadmin/data/rocketmq/commitlog
defaultTopicQueueNums=4
autoCreateTopicEnable=true
autoCreateSubscriptionGroup=true
flushDiskType=ASYNC_FLUSH

五、启动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

■NameServer = 节点2

六、启动Broker-a（Master位于节点一、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 &

■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 &

■验证broker-a：

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

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

七、启动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 &

■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 &

■验证broker-b

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

八、启动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 &

■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 &

■验证broker-c

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

问题1：

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

解决方法：因为我的电脑配置不够，没法为虚拟机申请更大的内存。因此，采用修改broker启动内存的方式解决了。

修改文件路径：{ROCKET_HOME}/bin/runbroker.sh

修改前：JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"

修改后：JAVA_OPT="${JAVA_OPT} -server -Xms128m -Xmx256m -Xmn256m -XX:PermSize=128m -XX:MaxPermSize=320m"

如图所示：

从新启动broker，日志中没有错误，jps进程中也能够看到broker正常启动。

问题2：

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

问题缘由：

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

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

解决办法：

修改rocketmq的配置文件，增长listenPort配置。配置以后以下所示：

→同一台机器的配置文件

→broker-a-m.properties

→broker-c-s.properties

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

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

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

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

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

使用程序测试

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

package post;

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.common.message.Message;

import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.UUID;

public class ProducerTest {

private static DefaultMQProducer producer = null;

public static void main(String[] args) {

System.out.print("[----------]Start");

boolean result = false;

try {

ProducerStart();

SendMessage("qch_20170706", "hello rocketmq!");

}finally {

producer.shutdown();

}

System.out.print("[----------]Succeed");

}

private static boolean ProducerStart() {

producer = new DefaultMQProducer("pro_qch_test");

producer.setNamesrvAddr("192.168.6.3:9876;192.168.6.4:9876");

producer.setInstanceName(UUID.randomUUID().toString());

producer.setVipChannelEnabled(false);

try {

producer.start();

} catch (MQClientException e) {

e.printStackTrace();

return false;

}

return true;

}

private static boolean SendMessage(String topic, String str) {

Message msg = new Message(topic, str.getBytes());

try {

producer.send(msg);

} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {

e.printStackTrace();

return false;

}

return true;

}

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

问题缘由：

由于broker部署在虚拟机，而且虚拟双网卡，client没法正常链接服务端。

解决方法：

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

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

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

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

package post;

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.common.consumer.ConsumeFromWhere;

import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

import java.util.UUID;

public class ConsumerTest {

public static void main(String[] args) {

DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("con_qch_test");

consumer.setInstanceName(UUID.randomUUID().toString());

consumer.setConsumeMessageBatchMaxSize(32);

consumer.setNamesrvAddr("192.168.6.3:9876;192.168.6.4:9876");

consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

consumer.registerMessageListener(new MessageListenerConcurrently() {

@Override

public ConsumeConcurrentlyStatus consumeMessage(

List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

for(MessageExt me : list) {

System.out.print(new String(me.getBody()));

}

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

}

});

try {

consumer.subscribe("qch_20170706", "*");

consumer.start();

} catch (Exception e) {

e.printStackTrace();

}

样例代码上传到了git上了，有须要的能够去参考。

https://github.com/quchunhui/rocketmq_sample