侧边栏壁纸
博主头像
不过是从头再来罢了博主等级

每日的一小步，如同水滴石穿，持之以恒，必有成效。

  • 累计撰写 62 篇文章
  • 累计创建 35 个标签
  • 累计收到 0 条评论

目 录CONTENT

文章目录
docker

Docker安装RocketMQ

zhouzz
zhouzz
2024-09-27 / 0 评论 / 0 点赞 / 13 阅读 / 13429 字
温馨提示：
本文最后更新于 2024-09-28，若内容或图片失效，请留言反馈。部分素材来自网络，若不小心影响到您的利益，请联系我们删除。

1.前言

本章节将介绍如何在Docker环境下安装RocketMQ以及RocketMQ-Console可视化界面

官网地址: https://rocketmq.apache.org/zh/docs/

2.RocketMQ安装

在安装之前请保证您的服务器已经安装好Docker环境并且安装好JDK

# 拉取rocketMQ最新版本镜像
$ docker pull apache/rocketmq
$ docker pull apache/rocketmq:4.9.6

2.1 nameserver 安装

rocketMQ 分为nameserver和broker两部分，在启动时应该先启动nameserver，因此我们现在先创建nameserver的日志和数据存放目录。

1.创建挂载目录

这个目录可由我们自己定义路径，这里我将其放到data路径下:

mkdir -p /app/docker/rocketmq/nameserver/{logs,store}

2.创建容器共享网络

RocketMQ中有多个服务，需要创建多个容器，创建 docker 网络便于容器间相互通信。

docker network create rocketmq

3.构建namesrv容器并启动

我们已经创建好了nameserver的日志和数据的存放路径，此时我们只需要挂在日志和数据路径，执行以下命令启动nameserver即可。

docker run -d --privileged=true \
--name rmqnamesrv \
-p 9876:9876 \
--net rocketmq \
-v  /app/docker/rocketmq/nameserver/logs:/root/logs \
-v /app/docker/rocketmq/nameserver/store:/root/store \
--memory=256m \
apache/rocketmq:4.9.6 sh mqnamesrv

2.2 创建broker

1.创建broker数据存储路径

在操作这步之前，我们已经将nameserver启动成功，则接下来我们将要为broker创建数据挂载目录和配置文件，以确保broker能够成功启动。

mkdir -p /app/docker/rocketmq/broker/{logs,store,conf}

说明：logs:是broker的日志目录，store:是broker的数据目录，conf是broker的配置信息目录

2.创建broker配置文件

我们刚刚已经创建了broker配置目录(/app/docker/rocketmq/broker/conf)，现在我们要在改配置目录下增加一个broker的配置文件，命名叫 broker.conf

# 所属集群名字
brokerClusterName=DefaultCluster

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

# 0 表示 Master，> 0 表示 Slave
brokerId=0

# nameServer地址，分号分割。这里已经创建了 net ,就不用配置
#namesrvAddr=192.168.254.160:9876

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

# 是否允许 Broker 自动创建 Topic，建议线下开启，线上关闭 ！！！这里仔细看是 false，false，false
autoCreateTopicEnable=true
# 是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭
autoCreateSubscriptionGroup=true
# 在发送消息时，自动创建服务器不存在的topic，默认创建的队列数
defaultTopicQueueNums=4

# 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/ztztdata/rocketmq-all-4.1.0-incubating/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

3.构建broker容器并启动

docker run -d --privileged=true \
--name rmqbroker  \
--net rocketmq \
-p 10912:10912 -p 10911:10911 -p 10909:10909 \
-e "NAMESRV_ADDR=rmqnamesrv:9876" \
-v /app/docker/rocketmq/broker/logs:/root/logs \
-v /app/docker/rocketmq/broker/store:/root/store \
-v  /app/docker/rocketmq/broker/conf/broker.conf:/home/rocketmq/rocketmq-4.9.6/conf/broker.conf \
--memory=1g \
apache/rocketmq:4.9.6 sh mqbroker \
-c /home/rocketmq/rocketmq-4.9.6/conf/broker.conf

通过docker ps可以看到nameserver和broker都已经启动成功，此时rocketmq已经安装完成。如果您需要一个可视化界面查看信息，可以再安装一个rocketmq-console。

4.开放端口

firewall-cmd --zone=public --add-port=9876/tcp --permanent
firewall-cmd --zone=public --add-port=10909/tcp --permanent
firewall-cmd --zone=public --add-port=10911/tcp --permanent
firewall-cmd --zone=public --add-port=10912/tcp --permanent
firewall-cmd --reload

2.3 创建 rocketmq-dashboard

1.拉取 rocketmq-dashboard 镜像

$ docker pull apacherocketmq/rocketmq-dashboard:latest

2.构建RocketMQ-Console容器并启动

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

提示
namesrv.addr:port 替换为 rocketmq 中配置的 nameserver 地址：端口号
默认端口配置下，需开放端口号：控制台：9999；namesrv：9876；broker：10909、10911。

  • 云服务器：设置安全组访问规则
  • 本地虚拟机：关闭防火墙，或 -add-port

3.通过访问9999端口查看RocketMQ-Console是否启动成功

3.SDK测试消息收发

工具测试完成后，我们可以尝试使用 SDK 收发消息，这里以 Java SDK 为例介绍一下消息收发过程。

在IDEA中创建一个Java工程。

在 pom.xml 文件中添加以下依赖引入Java依赖库

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

在已创建的Java工程中，创建发送普通消息程序并运行，Apache RocketMQ可用于以三种方式发送消息：同步、异步和单向传输，这里以同步模式为示例：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

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

        try {
            // 创建消息实例，指定 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();
        }
    }
}

在已创建的Java工程中，创建订阅普通消息程序并运行。Apache RocketMQ 有两种消费模式：Push和Pull，这里以Push消费为示例。

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.message.MessageExt;

import java.util.List;

public class ConsumerExample {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例，并设置消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
        // 设置 Name Server 地址，此处为示例，实际使用时请替换为真实的 Name Server 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 订阅指定的主题和标签（* 表示所有标签）
        consumer.subscribe("TestTopic", "*");

        // 注册消息监听器
        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;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("Consumer started.");
    }
}

4.rocketMQ占用内存过大

设置合理的资源限制：根据应用程序的需要，设置合理的--memory和--cpus参数。可以根据应用程序的负载情况进行调整。

$ docker run -d --memory=1g --cpus=2 xxx_app
0
docker应用
版权归属： zhouzz
本文链接： http://localhost:8090/?p=cga8eH8s
许可协议： 本文使用《署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0)》协议授权

评论区