Docker 安装 RocketMQ 单机实例

发表于 阅读次数： 评论数： 本文字数： 1.2k 阅读时长 ≈ 1 分钟

大纲

前言

本文将介绍如何使用 Docker 和 Docker Compose 快速安装单机版的 RocketMQ，包括 NameServer 和 Broker 组件。

版本说明

组件版本说明
RocketMQ Server4.9.6RocketMQ 服务端（包括 NameServer 和 Broker），运行时依赖 JDK 1.8+
RocketMQ Client4.9.6RocketMQ 客户端（SDK），要求跟 RocketMQ Server 的版本严格匹配

Docker 安装 RocketMQ

创建容器共享网络

  • RocketMQ 中有多个服务，需要创建多个容器，创建 Docker 网络便于容器间相互通信。
1
2
# 创建共享网络
docker network create rocketmq

启动 NameServer

1
2
3
4
5
# 启动 NameServer
docker run -d --name rmqnamesrv -p 9876:9876 --net rocketmq apache/rocketmq:4.9.6 sh mqnamesrv

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

提示

NameServer 启动后，如果看到日志信息 The Name Server boot success..，则表示 NameServer 已成功启动。

启动 Broker

  • 创建 Broker 的配置文件
1
2
# 配置 Broker 的 IP 地址
echo "brokerIP1=127.0.0.1" > broker.conf
  • 启动 Broker
1
2
3
4
5
6
7
8
9
10
11
12
# 启动 Broker
docker run -d \
    --name rmqbroker \
    --net rocketmq \
    -p 10912:10912 -p 10911:10911 -p 10909:10909 \
    -e "NAMESRV_ADDR=rmqnamesrv:9876" \
    -v ./broker.conf:/home/rocketmq/rocketmq-4.9.6/conf/broker.conf \
    apache/rocketmq:4.9.6 sh mqbroker \
    -c /home/rocketmq/rocketmq-4.9.6/conf/broker.conf

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

提示

  • Broker 启动后，如果看到日志信息 The broker boot success..，则表示 Broker 已成功启动。
  • 至此，基于 单组节点单副本模式 已经将 RocketMQ 部署起来了，后续可以利用 Shell 脚本或者 SDK 代码进行简单的消息收发测试。

Docker Compose 安装 RocketMQ

创建配置文件

  • 创建 Broker 的配置文件
1
2
# 配置 Broker 的 IP 地址
echo "brokerIP1=127.0.0.1" > broker.conf
  • 创建 Docker Compose 配置文件
1
2
# 创建配置文件（写入以下配置内容）
vim docker-compose.yml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
version: '3.5'

services:
  namesrv:
    image: apache/rocketmq:4.9.6
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    networks:
      - rocketmq
    command: sh mqnamesrv

  broker:
    image: apache/rocketmq:4.9.6
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
      - 10912:10912
    environment:
      - NAMESRV_ADDR=rmqnamesrv:9876
    volumes:
      - ./broker.conf:/home/rocketmq/rocketmq-4.9.6/conf/broker.conf
    depends_on:
      - namesrv
    networks:
      - rocketmq
    command: sh mqbroker -c /home/rocketmq/rocketmq-4.9.6/conf/broker.conf

networks:
  rocketmq:
    driver: bridge

提示

为了快速启动并运行 RocketMQ，可以使用以上模板，通过修改或添加环境变量部分中的配置来创建 docker-compose.yml 配置文件。

启动 Docker 容器

  • 启动 Namer Server 和 Broker
1
docker-compose up -d
  • 验证 NameServer 是否启动成功，如果看到日志信息 The Name Server boot success..，则表示 NameServer 已成功启动
1
docker logs -f rmqnamesrv
  • 验证 Broker 是否启动成功，如果看到日志信息 The broker boot success..，则表示 Broker 已成功启动
1
docker logs rmqbroker

基于脚本测试消息收发

消息发送测试

1
2
3
4
5
# 进入 Broker 容器
docker exec -it rmqbroker bash

# 启动消息生产实例（运行输出结果如下）
sh tools.sh org.apache.rocketmq.example.quickstart.Producer
1
2
3
4
5
SendResult [sendStatus=SEND_OK, msgId=7F000001018627C170F001C87FFF03E4, offsetMsgId=7F00000100002A9F000000000002EB4E, messageQueue=MessageQueue [topic=TopicTest, brokerName=3fb968547e0f, queueId=0], queueOffset=249]
SendResult [sendStatus=SEND_OK, msgId=7F000001018627C170F001C8800003E5, offsetMsgId=7F00000100002A9F000000000002EC0E, messageQueue=MessageQueue [topic=TopicTest, brokerName=3fb968547e0f, queueId=1], queueOffset=249]
SendResult [sendStatus=SEND_OK, msgId=7F000001018627C170F001C8800103E6, offsetMsgId=7F00000100002A9F000000000002ECCE, messageQueue=MessageQueue [topic=TopicTest, brokerName=3fb968547e0f, queueId=2], queueOffset=249]
SendResult [sendStatus=SEND_OK, msgId=7F000001018627C170F001C8800203E7, offsetMsgId=7F00000100002A9F000000000002ED8E, messageQueue=MessageQueue [topic=TopicTest, brokerName=3fb968547e0f, queueId=3], queueOffset=249]
......

消息消费测试

1
2
3
4
5
# 进入 Broker 容器
docker exec -it rmqbroker bash

# 启动消息消费实例（运行输出结果如下）
sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
1
2
3
ConsumeMessageThread_please_rename_unique_group_name_4_8 Receive New Messages: [MessageExt [brokerName=3fb968547e0f, queueId=0, storeSize=192, queueOffset=226, sysFlag=0, bornTimestamp=1780301917082, bornHost=/127.0.0.1:52616, storeTimestamp=1780301917083, storeHost=/127.0.0.1:10911, msgId=7F00000100002A9F000000000002A64E, commitLogOffset=173646, bodyCRC=159021295, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1780301981633, UNIQ_KEY=7F000001018627C170F001C87F9A0388, CLUSTER=DefaultCluster, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 48, 52], transactionId='null'}]] 
ConsumeMessageThread_please_rename_unique_group_name_4_3 Receive New Messages: [MessageExt [brokerName=3fb968547e0f, queueId=0, storeSize=192, queueOffset=224, sysFlag=0, bornTimestamp=1780301917073, bornHost=/127.0.0.1:52616, storeTimestamp=1780301917073, storeHost=/127.0.0.1:10911, msgId=7F00000100002A9F000000000002A04E, commitLogOffset=172110, bodyCRC=930400445, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='TopicTest', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=250, CONSUME_START_TIME=1780301981633, UNIQ_KEY=7F000001018627C170F001C87F910380, CLUSTER=DefaultCluster, TAGS=TagA}, body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 56, 57, 54], transactionId='null'}]]
......

基于 SDK 测试消息收发

创建 Maven 工程

  • 在 IDEA 中创建一个 Maven 工程，往 pom.xml 文件中添加以下依赖引入 Java 依赖库:
1
2
3
4
5
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.9.6</version>
</dependency>

创建消息生产者

  • 在已创建的 Maven 工程中，创建发送普通消息的程序并运行。RocketMQ 可以使用三种方式发送消息：同步、异步和单向传输，这里以同步发送模式为示例：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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("producer_group");
        // 设置 NameServer 地址，此处为示例，实际使用时请替换为真实的 NameServer 地址
        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();
        }
    }

}
  • 启动发送普通消息的程序，程序运行输出的结果如下：
1
2
Message sent: Hello RocketMQ
Send result: SendResult [sendStatus=SEND_OK, msgId=ACAF0701967D512DDF170375668C0000, offsetMsgId=7F00000100002A9F0000000000000000, messageQueue=MessageQueue [topic=TestTopic, brokerName=3fb968547e0f, queueId=1], queueOffset=0]

创建消息消费者

  • 在已创建的 Maven 工程中，创建消费普通消息的程序并运行。RocketMQ 有两种消费模式：Push 和 Pull，这里以 Push 消费模式为示例：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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("consumer_group");
        // 设置 NameServer 地址，此处为示例，实际使用时请替换为真实的 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 设置初始消费位置（该配置只有在消费者组第一次启动时才生效）
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 订阅指定的 Topic 和 Tag（* 表示所有标签）
        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.");
    }

}
  • 启动消费普通消息的程序，程序运行输出的结果如下：
1
2
Consumer started.
Received message: Hello RocketMQ

参考博客资料