cent7下rocketmq集群配置

左手的ㄟ右手 2022-06-13 08:37 204阅读 0赞

关于roketmq安装请参考博文roketmq 单机版，集群主要是在不同机器上配置roketmq的broker属性文件，启动broker的时候指定相应的属性文件。

本人采用2台虚拟机，A: 192.168.19.33，B: 192.168.19.34

A 作为broker-a Master与 broker-b  Slave

B 配置broker-b  master 与 broker-a  slave

以下参考：http://binux.cn/2017/03/07/RocketMQ-Cluster-Install/

#### 创建存储路径 ####

mkdir /usr/local/rocketmq/store
    mkdir /usr/local/rocketmq/store/commitlog
    mkdir /usr/local/rocketmq/store/consumequeue
    mkdir /usr/local/rocketmq/store/index

#### 配置RocketMQ配置文件 ####

vim /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties   （A服务器）
    vim /usr/local/rocketmq/conf/2m-2s-async/broker-b.properties （B服务器）
    vim /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties  （B服务器）
    vim /usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties （A服务器）

##### 配置文件 #####

#所属集群名字 （注意不要有空格）
    brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样 192.168.1.1/192.168.1.3 填broker-a 192.168.1.2/192.168.1.4 填broker-b
    brokerName=(broker-a|broker-b)
    #0 表示 Master，>0 表示 Slave 192.168.1.1/192.168.1.2 填0 192.168.1.3/192.168.1.4 填1
    brokerId=0
    #本机IP 默认识别 多块网卡会导致识别错误
    brokerIP1=本机IP
    #nameServer地址，分号分割
    namesrvAddr=192.168.1.1:9876;192.168.1.2:9876;192.168.1.3:9876;192.168.1.4:9876
    #在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
    defaultTopicQueueNums=4
    #是否允许 Broker 自动创建Topic，建议线下开启，线上关闭
    autoCreateTopicEnable=true
    #是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
    autoCreateSubscriptionGroup=true
    #Broker 对外服务的监听端口
    listenPort=10911
    #删除文件时间点，默认凌晨 4点
    deleteWhen=04
    #文件保留时间，默认 48 小时
    fileReservedTime=120
    #commitLog每个文件的大小默认1G
    mapedFileSizeCommitLog=1073741824
    #ConsumeQueue每个文件默认存30W条，根据业务情况调整
    mapedFileSizeConsumeQueue=300000
    #destroyMapedFileIntervalForcibly=120000
    #redeleteHangedFileInterval=120000
    #检测物理文件磁盘空间
    diskMaxUsedSpaceRatio=88
    
    #存储路径
    storePathRootDir=/usr/local/rocketmq/store
    #commitLog 存储路径
    storePathCommitLog=/usr/local/rocketmq/store/commitlog
    #消费队列存储路径存储路径
    storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
    #消息索引存储路径
    storePathIndex=/usr/local/rocketmq/store/index
    #checkpoint 文件存储路径
    storeCheckpoint=/usr/local/rocketmq/store/checkpoint
    #abort 文件存储路径
    abortFile=/usr/local/rocketmq/store/abort
    #限制的消息大小
    maxMessageSize=65536
    
    #flushCommitLogLeastPages=4
    #flushConsumeQueueLeastPages=2
    #flushCommitLogThoroughInterval=10000
    #flushConsumeQueueThoroughInterval=60000
    
    #Broker 的角色 192.168.1.1/192.168.1.2 填ASYNC_MASTER 192.168.1.3/192.168.1.4 填SLAVE
    #- ASYNC_MASTER 异步复制Master
    #- SYNC_MASTER 同步双写Master
    #- SLAVE
    brokerRole=ASYNC_MASTER
    
    #刷盘方式
    #- ASYNC_FLUSH 异步刷盘
    #- SYNC_FLUSH 同步刷盘
    flushDiskType=ASYNC_FLUSH
    
    #checkTransactionMessageEnable=false
    #发消息线程池数量
    #sendMessageThreadPoolNums=128
    #拉消息线程池数量
    #pullMessageThreadPoolNums=128
    #filterNums
    #filterServerNums=1

#### 修改日志配置文件 ####

mkdir -p /usr/local/rocketmq/logs
    cd /usr/local/rocketmq/conf && sed -i 's#${user.home}#/usr/local/rocketmq#g' *.xml

#### 修改启动脚本参数（测试时虚拟机内存不够修改） ####

vim /usr/local/rocketmq/bin/runbroker.sh
    JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m - XX:PermSize=128m -XX:MaxPermSize=320m"
    
    vim /usr/local/rocketmq/bin/runserver.sh
    JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m - XX:PermSize=128m -XX:MaxPermSize=320m"

#### 启动NameServer ####

nohup sh /usr/local/rocketmq/bin/mqnamesrv &

##### 查看是否启动成功 #####

tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log

**开放端口号：**

可以先执行：netstat -ntlp查看正在运行的端口号，通过下面命令开启对外开放，否者一端将服务连接另外一端服务

firewall-cmd --zone=public --add-port=9876/tcp      //将9876改成其他端口号

#### 启动BrokerServer 多Master多Slave模式，异步复制 ####

> A -192.168.1.1|B-192.168.1.2|C-192.168.1.3|B-192.168.1.4

启动命令:

A:
    cd /usr/local/rocketmq/bin
    nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a.properties >/dev/null 2>&1 &
    //查看是否启动成功
    netstat -ntlp
    jps
    // 查看日志
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log
    
    B:
    cd /usr/local/rocketmq/bin
    nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b.properties >/dev/null 2>&1 &
    //查看是否启动成功
    netstat -ntlp
    jps
    // 查看日志
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log
    
    C:
    cd /usr/local/rocketmq/bin
    nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-a-s.properties >/dev/null 2>&1 &
    //查看是否启动成功
    netstat -ntlp
    jps
    // 查看日志
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log
    
    D:
    cd /usr/local/rocketmq/bin
    nohup sh /usr/local/rocketmq/bin/mqbroker -c /usr/local/rocketmq/conf/2m-2s-async/broker-b-s.properties >/dev/null 2>&1 &
    //查看是否启动成功
    netstat -ntlp
    jps
    // 查看日志
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/broker.log
    tail -f -n 500 /usr/local/rocketmq/logs/rocketmqlogs/namesrv.log

实例参考：http://blog.csdn.net/lovesomnus/article/details/51769977

生产者：（为方便，个人引入roketmq lib下所有jar）

package com.store.test;

import java.util.concurrent.TimeUnit;

import org.apache.rocketmq.client.exception.MQClientException;

import org.apache.rocketmq.client.producer.LocalTransactionExecuter;

import org.apache.rocketmq.client.producer.LocalTransactionState;

import org.apache.rocketmq.client.producer.SendResult;

import org.apache.rocketmq.client.producer.TransactionCheckListener;

import org.apache.rocketmq.client.producer.TransactionMQProducer;

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

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

public class Producer \{

public staticvoid main(String\[\]args)throws MQClientException, InterruptedException \{

/\*\*

\* 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例

\* 注意：ProducerGroupName需要由应用来保证唯一,一类Producer集合的名称，这类Producer通常发送一类消息，

\* 且发送逻辑一致

\* ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，

\* 因为服务器会回查这个Group下的任意一个Producer

\*/

final TransactionMQProducerproducer =new TransactionMQProducer("ProducerGroupName");

// nameserver服务

producer.setNamesrvAddr("192.168.19.33:9876;192.168.19.34:9876");

producer.setInstanceName("Producer");

/\*\*

\* Producer对象在使用之前必须要调用start初始化，初始化一次即可

\* 注意：切记不可以在每次发送消息时，都调用start方法

\*/

producer.start();

// 服务器回调Producer，检查本地事务分支成功还是失败

producer.setTransactionCheckListener(new TransactionCheckListener() \{

public LocalTransactionState checkLocalTransactionState(

MessageExt msg) \{

System.out.println("checkLocalTransactionState --" \+new String(msg.getBody()));

return LocalTransactionState.COMMIT\_MESSAGE;

\});

/\*\*

\* 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。

\* 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，

\* 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，

\* 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。

\*/

for (inti = 0;i < 10;i\++) \{

try \{

Message msg =new Message("TopicTest1",// topic

"TagA",                        // tag

"OrderID001",                  // key消息关键词，多个Key用KEY\_SEPARATOR隔开（查询消息使用）

("Hello MetaQA").getBytes());  // body

SendResult sendResult =producer.sendMessageInTransaction(

msg,new LocalTransactionExecuter()\{

public LocalTransactionState executeLocalTransactionBranch(Messagemsg, Objectarg) \{

System.out.println("executeLocalTransactionBranch--msg=" \+new String(msg.getBody()));

System.out.println("executeLocalTransactionBranch--arg=" \+arg);

return LocalTransactionState.COMMIT\_MESSAGE;

\},

"$$$");

System.out.println(sendResult);

Message msg =new Message("TopicTest2",// topic

"TagB",                        // tag

"OrderID0034",                 // key 消息关键词，多个Key用KEY\_SEPARATOR隔开（查询消息使用）

("Hello MetaQB").getBytes());  // body

SendResult sendResult =producer.sendMessageInTransaction(

msg,new LocalTransactionExecuter()\{

public LocalTransactionState executeLocalTransactionBranch(Messagemsg, Objectarg) \{

System.out.println("executeLocalTransactionBranch--msg=" \+new String(msg.getBody()));

System.out.println("executeLocalTransactionBranch--arg=" \+arg);

return LocalTransactionState.COMMIT\_MESSAGE;

\},

"$$$");

System.out.println(sendResult);

Message msg =new Message("TopicTest3",// topic

"TagC",                        // tag

"OrderID061",                  // key

("Hello MetaQC").getBytes());  // body

SendResult sendResult =producer.sendMessageInTransaction(

msg,new LocalTransactionExecuter()\{

public LocalTransactionState executeLocalTransactionBranch(Messagemsg, Objectarg) \{

System.out.println("executeLocalTransactionBranch--msg=" \+new String(msg.getBody()));

System.out.println("executeLocalTransactionBranch--arg=" \+arg);

return LocalTransactionState.COMMIT\_MESSAGE;

\},

"$$$");

System.out.println(sendResult);

\} catch (Exceptione) \{

e.printStackTrace();

TimeUnit.MILLISECONDS.sleep(1000);

/\*\*

\* 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己

\* 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法

\*/

// producer.shutdown();

Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() \{

public  void run() \{

producer.shutdown();

\}));

System.exit(0);

\} // 执行本地事务，由客户端回调

运行效果：

SendResult \[sendStatus=SEND\_OK, msgId=A9FECC8B0BFD73D16E930C8F8C9F0019, offsetMsgId=null, messageQueue=MessageQueue \[topic=TopicTest2, brokerName=broker-a, queueId=0\], queueOffset=0\]

executeLocalTransactionBranch--msg=Hello MetaQC

executeLocalTransactionBranch--arg=$$$

SendResult \[sendStatus=SEND\_OK, msgId=A9FECC8B0BFD73D16E930C8F8CA2001A, offsetMsgId=null, messageQueue=MessageQueue \[topic=TopicTest3, brokerName=broker-a, queueId=3\], queueOffset=0\]

executeLocalTransactionBranch--msg=Hello MetaQA

executeLocalTransactionBranch--arg=$$$

SendResult \[sendStatus=SEND\_OK, msgId=A9FECC8B0BFD73D16E930C8F908F001B, offsetMsgId=null, messageQueue=MessageQueue \[topic=TopicTest1, brokerName=broker-a, queueId=2\], queueOffset=0\]

executeLocalTransactionBranch--msg=Hello MetaQB

executeLocalTransactionBranch--arg=$$$

SendResult \[sendStatus=SEND\_OK, msgId=A9FECC8B0BFD73D16E930C8F9091001C, offsetMsgId=null, messageQueue=MessageQueue \[topic=TopicTest2, brokerName=broker-a, queueId=1\], queueOffset=0\]

executeLocalTransactionBranch--msg=Hello MetaQC

executeLocalTransactionBranch--arg=$$$

SendResult \[sendStatus=SEND\_OK, msgId=A9FECC8B0BFD73D16E930C8F90B7001D, offsetMsgId=null, messageQueue=MessageQueue \[topic=TopicTest3, brokerName=broker-a, queueId=0\], queueOffset=0\]

消费者：

package com.store.test;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.Set;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;

import org.apache.rocketmq.client.consumer.PullResult;

import org.apache.rocketmq.client.exception.MQClientException;

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

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

public class Consumer \{

// Java缓存

private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();

/\*\*

\* 主动拉取方式消费

\* @throws MQClientException

\*/

public static  void main(String\[\] args) throws MQClientException \{

/\*\*

\* 一个应用创建一个Consumer，由应用来维护此对象，可以设置为全局对象或者单例

\* 注意：ConsumerGroupName需要由应用来保证唯一 ,最好使用服务的包名区分同一服务,一类Consumer集合的名称，

\* 这类Consumer通常消费一类消息，且消费逻辑一致

\* PullConsumer：Consumer的一种，应用通常主动调用Consumer的拉取消息方法从Broker拉消息，主动权由应用控制

\*/

DefaultMQPullConsumer consumer =  new DefaultMQPullConsumer("ConsumerGroupName");

// //nameserver服务

consumer.setNamesrvAddr("192.168.19.33:9876;192.168.19.34:9876");

consumer.setInstanceName("Consumber");

consumer.start();

// 拉取订阅主题的队列，默认队列大小是4

Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");

for (MessageQueue  mq : mqs) \{

System.out.println("Consume from the queue: " \+mq);

SINGLE\_MQ: while (true) \{

try \{

PullResult pullResult =consumer.pullBlockIfNotFound(mq,null, getMessageQueueOffset(mq), 32);

List<MessageExt> list =pullResult.getMsgFoundList();

if (list !=null &&list.size() < 100) \{

for (MessageExt  msg : list) \{

System.out.println(new String(msg.getBody()));

System.out.println(pullResult.getNextBeginOffset());

putMessageQueueOffset(mq,pullResult.getNextBeginOffset());

switch (pullResult.getPullStatus()) \{

case  FOUND:

break;

case  NO\_MATCHED\_MSG:

break;

case  NO\_NEW\_MSG:

break SINGLE\_MQ;

case  OFFSET\_ILLEGAL:

break;

default:

break;

\} catch (Exception  e) \{

e.printStackTrace();

consumer.shutdown();

private static  void putMessageQueueOffset(MessageQueue mq,  long offset) \{

offseTable.put(mq,offset);

private static  long getMessageQueueOffset(MessageQueue mq) \{

Long offset = offseTable.get(mq);

if (offset !=null) \{

System.out.println(offset);

return offset;

return 0;

运行效果：

Consume from the queue: MessageQueue \[topic=TopicTest1, brokerName=broker-a, queueId=3\]

Hello MetaQA

Consume from the queue: MessageQueue \[topic=TopicTest1, brokerName=broker-a, queueId=2\]

Hello MetaQA

参考文章：

1、http://binux.cn/2017/03/07/RocketMQ-Cluster-Install/

2、http://blog.csdn.net/lovesomnus/article/details/51769977

cent7下rocketmq集群配置

发表评论取消回复

还没有评论，来说两句吧...

相关阅读

相关 Linux环境下安装RocketMQ（单机、集群）

相关 Centos7搭建RocketMq集群

相关 RocketMQ集群详解

相关 centos7安装rocketmq集群4.7.1

相关 cent7下rocketmq集群配置

相关 CentOS7 集群配置

相关 centos7下redis哨兵集群配置

相关 RocketMQ集群部署

相关 RocketMQ本地集群配置

相关 cent os 7——安装配置手记