1 环境概述

192.168.38.50 node1 broker-master nameserver rocketmq-console
192.168.38.51 node2 broker-slave  nameserver
192.168.38.52 node3 broker-master
192.168.38.53 node4 broker-slave

2 nameserver搭建

#node1、node2
docker pull foxiswho/rocketmq:server-4.3.2

docker run -p 9876:9876 --name rmqserver \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
-e "JAVA_OPTS=-Duser.home=/opt" \
-v /usr/local/rocketmq/rmqserver/logs:/opt/logs \
-v /usr/local/rocketmq/rmqserver/store:/opt/store \
-d foxiswho/rocketmq:server-4.3.2

#查看
docker ps

3 广控台搭建

node1 :

#拉取镜像
docker pull styletang/rocketmq-console-ng:1.0.0
#启动
docker run --name rocketmq-console \
-e "JAVA_OPTS=-Drocketmq.namesrv.addr=192.168.38.50:9876;192.168.38.51:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" \
-p 8082:8080 -t -d styletang/rocketmq-console-ng:1.0.0

4 broker安装

4.1 拉取镜像

#4台
docker pull foxiswho/rocketmq:broker-4.3.2

4.2 node1

mkdir -p /usr/local/rocketmq/rmqbroker/conf
vi /usr/local/rocketmq/rmqbroker/conf/broker.conf

#nameserver地址
namesrvAddr=192.168.38.50:9876;192.168.38.51:9876
#集群名称
brokerClusterName=rosh-cluster
#broker 名称
brokerName=rosh-master1
#0是主 1是从
brokerId=0
#每天凌晨4点清理过期消息
deleteWhen=04
#消息保存时间48小时
fileReservedTime=48
#同步双写
brokerRole=ASYNC_FLUSH
#异步刷盘
flushDiskType=ASYNC_FLUSH
#brokerIP1: 本机IP。brokerIp2：主从通信。
brokerIP1=192.168.38.50
brokerIP2=192.168.38.50
#集群通信端口
listenPort=10911

#启动
docker run -p 10911:10911 -p 10909:10909  -p 10912:10912 --name rmqbroker \
-e "JAVA_OPTS=-Duser.home=/opt" \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
-v /usr/local/rocketmq/rmqbroker/conf/broker.conf:/etc/rocketmq/broker.conf \
-v /usr/local/rocketmq/rmqbroker/logs:/opt/logs \
-v /usr/local/rocketmq/rmqbroker/store:/opt/store \
-d foxiswho/rocketmq:broker-4.3.2

4.3 node2

mkdir -p /usr/local/rocketmq/rmqbroker/conf
vi /usr/local/rocketmq/rmqbroker/conf/broker.conf

#nameserver地址
namesrvAddr=192.168.38.50:9876;192.168.38.51:9876
#集群名称
brokerClusterName=rosh-cluster
#broker 名称，通过brokerName 找主
brokerName=rosh-master1
#0是主 1是从
brokerId=1
#每天凌晨4点清理过期消息
deleteWhen=04
#消息保存时间48小时
fileReservedTime=48
#同步双写
brokerRole=SLAVE
#异步刷盘
flushDiskType=ASYNC_FLUSH
#brokerIP1找主节点
brokerIP1=192.168.38.50
#集群通信端口
listenPort=10911


# 10911 生产者、消费者通信端口 。10909 主从通信端口。10912集群通信端口
docker run -p 10911:10911 -p 10909:10909  -p 10912:10912 --name rmqbroker \
-e "JAVA_OPTS=-Duser.home=/opt" \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
-v /usr/local/rocketmq/rmqbroker/conf/broker.conf:/etc/rocketmq/broker.conf \
-v /usr/local/rocketmq/rmqbroker/logs:/opt/logs \
-v /usr/local/rocketmq/rmqbroker/store:/opt/store \
-d foxiswho/rocketmq:broker-4.3.2

4.4 node3

#创建配置文件
mkdir -p /usr/local/rocketmq/rmqbroker/conf
vi /usr/local/rocketmq/rmqbroker/conf/broker.conf



#nameserver地址
namesrvAddr=192.168.38.50:9876;192.168.38.51:9876
#集群名称
brokerClusterName=rosh-cluster
#broker 名称
brokerName=rosh-master2
#0是主 1是从
brokerId=0
#每天凌晨4点清理过期消息
deleteWhen=04
#消息保存时间48小时
fileReservedTime=48
#同步双写
brokerRole=ASYNC_FLUSH
#异步刷盘
flushDiskType=ASYNC_FLUSH
#brokerIP1: 本机IP。brokerIp2：主从通信。
brokerIP1=192.168.38.52
brokerIP2=192.168.38.52
#集群通信端口
listenPort=10911


# 10911 生产者、消费者通信端口 。10909 主从通信端口。10912集群通信端口
docker run -p 10911:10911 -p 10909:10909  -p 10912:10912 --name rmqbroker \
-e "JAVA_OPTS=-Duser.home=/opt" \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
-v /usr/local/rocketmq/rmqbroker/conf/broker.conf:/etc/rocketmq/broker.conf \
-v /usr/local/rocketmq/rmqbroker/logs:/opt/logs \
-v /usr/local/rocketmq/rmqbroker/store:/opt/store \
-d foxiswho/rocketmq:broker-4.3.2

4.5 node4

#创建配置文件
mkdir -p /usr/local/rocketmq/rmqbroker/conf
vi /usr/local/rocketmq/rmqbroker/conf/broker.conf



#nameserver地址
namesrvAddr=192.168.38.50:9876;192.168.38.51:9876
#集群名称
brokerClusterName=rosh-cluster
#broker 名称，找主节点
brokerName=rosh-master2
#0是主 1是从
brokerId=1
#每天凌晨4点清理过期消息
deleteWhen=04
#消息保存时间48小时
fileReservedTime=48
#同步双写
brokerRole=SLAVE
#异步刷盘
flushDiskType=ASYNC_FLUSH
#brokerIP 找主节点
brokerIP1=192.168.38.52
#集群通信端口
listenPort=10911


# 10911 生产者、消费者通信端口 。10909 主从通信端口。10912集群通信端口
docker run -p 10911:10911 -p 10909:10909  -p 10912:10912 --name rmqbroker \
-e "JAVA_OPTS=-Duser.home=/opt" \
-e "JAVA_OPT_EXT=-server -Xms128m -Xmx128m -Xmn128m" \
-v /usr/local/rocketmq/rmqbroker/conf/broker.conf:/etc/rocketmq/broker.conf \
-v /usr/local/rocketmq/rmqbroker/logs:/opt/logs \
-v /usr/local/rocketmq/rmqbroker/store:/opt/store \
-d foxiswho/rocketmq:broker-4.3.2

5 测试

5.1 生产者

public class RocketClient {

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

        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        producer.setNamesrvAddr("192.168.38.50:9876;192.168.38.51:9876");
        producer.start();
        for (int i = 0; i < 100; i++) {
            Message msg = new Message("rosh-topic1", "TagA", ("Hello RocketMQ " + i).getBytes(StandardCharsets.UTF_8));
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        producer.shutdown();

    }

}

5.2 消费者

public class RocketMqCustomer {

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

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rosh-group");
        consumer.setNamesrvAddr("192.168.38.50:9876;192.168.38.51:9876");
        consumer.subscribe("rosh-topic1","*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg:msgs){
                System.out.println(new String(msg.getBody(), StandardCharsets.UTF_8));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }

}

5.3 结果