从0开始搭建微服务架构第九章docker部署RocketMQ

方通网络

247人浏览 · 2025-03-08 22:37:46

方通网络 · 2025-03-08 22:37:46 发布

前言：RocketMq逻辑结构图
在这里插入图片描述

注意：RocketMQ和Sentinel区别，Sentinel是外部对内部服务的流量控制，RocketMQ是服务间的流量控制。
1、拉取rocket mq

docker pull apache/rocketmq

2、创建容器共享网络（可选）
RocketMQ中有多个服务，需要创建多个容器，创建 docker 网络便于容器间相互通信。

docker network create rocketmq-network

查看所有docker网络

docker network ls

删除docker网络

docker network rm rocketmq-network

3、启动NameServer

docker run -d \
  --name rmqnamesrv \
  -p 9876:9876 \
  -v /mydata/rocketmq/namesrv:/home/rocketmq/data \
  --network rocketmq-network \
  apache/rocketmq \
  sh mqnamesrv

参数说明：

-p 9876:9876：映射NameServer 默认端口。
-v：挂载数据卷避免容器重启丢失数据。

# 验证NameServer是否启动成功
docker logs -f rmqnamesrv

4、启动Broker
这里注意的是虚拟机启动broker如果内存在3G以下，会频繁卡死。所以，设置多点内存给虚拟机。
建议–restart=always不要设置自动重启，因为，容易内容溢出导致系统卡死。
注意：broker.conf要提前建好空文件

docker run -d \
  --name rmqbroker \
  -p 10909:10909 -p 10911:10911 -p 10912:10912 \
  -v  /mydata/rocketmq/broker/conf:/home/rocketmq/conf \
  -v  /mydata/rocketmq/broker/store:/home/rocketmq/store \
  -e "NAMESRV_ADDR=rmqnamesrv:9876" \
  --network rocketmq-network \
   apache/rocketmq\
  nohup sh mqbroker -c /home/rocketmq/conf/broker.conf

这里的rmqnamesrv:9876对应NameServer的名称。

# 验证 Broker 是否启动成功
docker logs rmqbroker

如果返回 killed 大概率就是内存不够。
可以通过

docker stats

查看分配内存资源占用率
配置文件broker.conf

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.


# 所属集群名字
brokerClusterName=DefaultCluster

# broker 名字，注意此处不同的配置文件填写的不一样，如果在 broker-a.properties 使用: broker-a,
# 在 broker-b.properties 使用: broker-b
brokerName=broker-a

# 0 表示 Master，&gt; 0 表示 Slave
brokerId=0

# nameServer地址，分号分割
# namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876

# 启动IP,如果 docker 报 com.alibaba.rocketmq.remoting.exception.RemotingConnectException: connect to &lt;192.168.0.120:10909&gt; failed
# 解决方式1 加上一句 producer.setVipChannelEnabled(false);，解决方式2 brokerIP1 设置宿主机IP，不要使用docker 内部IP
brokerIP1=192.168.0.5

# 在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4

# 是否允许 Broker 自动创建 Topic，建议线下开启，线上关闭 ！！！这里仔细看是 false，false，false
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=/home/rocketmq/store
# commitLog 存储路径
# storePathCommitLog=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/commitlog
# 消费队列存储
# storePathConsumeQueue=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/consumequeue
# 消息索引存储路径
# storePathIndex=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/index
# checkpoint 文件存储路径
# storeCheckpoint=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/checkpoint
# abort 文件存储路径
# abortFile=/home/ztztdata/rocketmq-all-4.1.0-incubating/store/abort
# 限制的消息大小
maxMessageSize=65536

# flushCommitLogLeastPages=4
# flushConsumeQueueLeastPages=2
# flushCommitLogThoroughInterval=10000
# flushConsumeQueueThoroughInterval=60000

# Broker 的角色
# - ASYNC_MASTER 异步复制Master
# - SYNC_MASTER 同步双写Master
# - SLAVE
brokerRole=ASYNC_MASTER

# 刷盘方式
# - ASYNC_FLUSH 异步刷盘
# - SYNC_FLUSH 同步刷盘
flushDiskType=ASYNC_FLUSH

# 发消息线程池数量
# sendMessageThreadPoolNums=128
# 拉消息线程池数量
# pullMessageThreadPoolNums=128

5.SDK测试消息收发
发送和接收的服务都要加依赖。

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>5.3.1</version>
</dependency>

6、发消息的服务方法

 public String Send() {
        // 创建生产者实例，并设置生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // 设置 Name Server 地址，此处为示例，实际使用时请替换为真实的 Name Server 地址
        producer.setNamesrvAddr("localhost:9876");


        try {
            producer.start();
            // 创建消息实例，指定 topic、Tag和消息体
            Message msg = new Message("TestTopic", "TagA", ("Hello RocketMQ").getBytes());
            // 发送消息并获取发送结果
            SendResult sendResult = producer.send(msg);
            System.out.println("Message sent: " + new String(msg.getBody()));
            System.out.println("Send result: " + sendResult);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("Message sending failed.");
        } finally {
            // 关闭生产者
            producer.shutdown();
        }
        return "";
    }

6、接收消息的服务方法

 public String Reserver() {
        // 创建消费者实例，并设置消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
        // 设置 Name Server 地址，此处为示例，实际使用时请替换为真实的 Name Server 地址
        consumer.setNamesrvAddr("localhost:9876");
        try {
            // 订阅指定的主题和标签（* 表示所有标签）
            consumer.subscribe("TestTopic", "*");
        } catch (Exception e) {
        }
        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println("Received message: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        try {
            // 启动消费者
            consumer.start();
        } catch (Exception e) {
        }
        System.out.println("Consumer started.");
        return "收到";
    }

7、RocketMQ可视化管理工具RocketMQ Dashboard
拉取

docker pull apacherocketmq/rocketmq-dashboard:latest

运行

docker run -d --name rocketmq-dashboard -e "JAVA_OPTS=-Drocketmq.namesrv.addr=192.168.0.5:9876" -p 8080:8080 -t apacherocketmq/rocketmq-dashboard:latest

注意：如果无法启动，使用命令查看日志docker logs rocketmq-dashboard
错误如下：

library initialization failed - unable to allocate file descriptor table - out of memoryAborted (core dumped)

这类错误通常意味着系统中的文件描述符（file descriptors）资源已经耗尽
可以通过修改系统配置来增加可用的文件描述符数量。这可以通过修改/etc/security/limits.conf文件来实现。例如，要为所有用户增加文件描述符的数量，可以添加或修改以下行：

* soft nofile 65536
* hard nofile 65536

或者直接这样运行

docker run -d --name rocketmq-dashboard -e "JAVA_OPTS=-Drocketmq.namesrv.addr=192.168.0.5:9876" -p 8080:8080 --ulimit nofile=65535:65535 --ulimit nproc=65535:65535 -t apacherocketmq/rocketmq-dashboard:latest

ulimit -n查看默认值。

或者

docker run -d \
  --name rocketmq-console \
  -p 8080:8080 \
  -e "JAVA_OPTS=-Drocketmq.namesrv.addr=rmqnamesrv:9876" \
  --network rocketmq-network \
  apacherocketmq/rocketmq-dashboard

这里端口8080要注意了，如果docker logs rocketmq-console ，查看日志发现tomcat的端口是8082，就得把8080:8080,改成8082:8082。