Rocketmq

官网地址：地址
github地址：地址
优秀学习博客:地址

1. 安装rocketmq

1.1下载

下载地址：地址

1.2 解压

# 本文安装到/opt目录下
unzip rocketmq-all-4.9.4-bin-release.zip 
cd rocketmq-all-4.9.4-bin-release
1
2
3

1.3 启动nameserver

nohup sh bin/mqnamesrv &
## 用jps命令查看nameserver是否启动成功
jps
1
2
3

1.4 启动broker

 nohup sh bin/mqbroker -n localhost:9876 &
# 同样用jps查看是否运行成功
jps
1
2
3

遇到的坑

启动broker 启动不起来，原因是runbroker.sh 文件中，设置的内存过大，大于我虚拟机内存，实际生产环境要进行调优，这边我就改成了1G,在重启上一步命令即可

vim runbroker.sh
1

2.rockmq集群搭建(2主2从)

环境 4台虚拟机，node01~04 centos7

2.1 创建文件路径

#其中{}表示的是创建多个文件夹
mkdir -p  /var/rocketmq/{logs,store/{commitlog,index,consumequeue}}
1
2

2.2 配置config

cd /opt/rocketmq-all-4.9.4-bin-release/conf/
cp -r 2m-2s-sync/  22sync/
#配置4个文件
cd 22sync/
1
2
3
4

2.3 配置以下四个文件

broker-b.properties

brokerClusterName=DefaultCluster
#这个要主从一致
brokerName=broker-b
#master是0，slave是>0的
brokerId=0
deleteWhen=04
fileReservedTime=48
#SYNC_MASTER，ASYNC_MASTER，SLAVE
brokerRole=SYNC_MASTER
#降低机器的磁盘IO瓶颈
flushDiskType=ASYNC_FLUSH
namesrvAddr=192.168.37.129:9876;192.168.37.130:9876;192.168.37.131:9876
storePathRootDir=/var/rocketmq/store
storePathCommitLog=/var/rocketmq/store/commitlog
storePathIndex=/var/rocketmq/store/index
storePathConsumeQueue=/var/rocketmq/store/consumequeue
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

broker-b-s.properties

brokerClusterName=DefaultCluster
brokerName=broker-b
brokerId=1
deleteWhen=04
fileReservedTime=48
brokerRole=SLAVE
flushDiskType=ASYNC_FLUSH
namesrvAddr=192.168.37.129:9876;192.168.37.130:9876;192.168.37.131:9876
storePathRootDir=/var/rocketmq/store
storePathCommitLog=/var/rocketmq/store/commitlog
storePathIndex=/var/rocketmq/store/index
storePathConsumeQueue=/var/rocketmq/store/consumequeue
1
2
3
4
5
6
7
8
9
10
11
12

参数说明

brokerRole=SYNC_MASTER  |  ASYNC_MASTER
message.setWaitStoreMsgOK(true);
SYNC_MASTER: producer send 到master后，master同步给slave，然后给producer返回OK ack=-1
ASYNC_MASTER：producer send 到master后，master异步给slave同步，并给producer返回OK  ack=1
message.setWaitStoreMsgOK(false);  ack=0
集群也是该sync，或者async的同步形式，只不过ok不需要等了
1
2
3
4
5
6

broker-a以此类推

2.4 修改logs文件路径

cd /opt/rocketmq-all-4.9.4-bin-release/conf/
#表示的是查找conf目录下的所有xml文件中查找{user.home}替换成自己配置的var/rocketmq
 sed -i 's#${user.home}#/var/rocketmq#g'  *.xml
1
2
3

2.5 修改runbroker.sh

vi runbroker.sh  :
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g"
1
2

2.6 复制到其他主机

 scp -r ./rocketmq-all-4.9.4-bin-release/  node02:/opt
  scp -r ./rocketmq-all-4.9.4-bin-release/  node03:/opt
   scp -r ./rocketmq-all-4.9.4-bin-release/  node04:/opt
1
2
3

2.7 启动

2.7.1启动nameserver

node01~node03

 nohup sh  mqnamesrv &
1

2.7.2 各个节点启动

node01

mqbroker -c /opt/rocketmq-all-4.9.4-bin-release/conf/22sync/broker-a.properties 
1

node02

mqbroker -c /opt/rocketmq-all-4.9.4-bin-release/conf/22sync/broker-a-s.properties 
1

node03

 mqbroker -c /opt/rocketmq-all-4.9.4-bin-release/conf/22sync/broker-b.properties
1

node04

mqbroker -c /opt/rocketmq-all-4.9.4-bin-release/conf/22sync/broker-b-s.properties 
1

2.8 测试启动

    @Test
    public void admin() throws Exception {

        DefaultMQAdminExt admin = new DefaultMQAdminExt();
        admin.setNamesrvAddr("192.168.37.129:9876;192.168.37.130:9876;192.168.37.131:9876");
        admin.start();

        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
        HashMap<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
        Set<Map.Entry<String, BrokerData>> entries = brokerAddrTable.entrySet();
        Iterator<Map.Entry<String, BrokerData>> iter = entries.iterator();
        while(iter.hasNext()){
            Map.Entry<String, BrokerData> next = iter.next();
            System.out.println(next.getKey()+ " "+ next.getValue());
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

输出