Quick Start 快速部署

2024年7月26日

前言

本部分内容主要介绍如何快速拉起 RocketMQ 集群，我们提供了如下一些快速拉起的指南：

通过 Docker 部署 RocketMQ 集群
通过 Docker Compose 迅速拉起 RocketMQ 集群
迅速部署具备自动切换能力的 RocketMQ 集群

通过 Docker 部署 RocketMQ 集群

这一节介绍如何使用 Docker 快速部署一个单节点单副本 RocketMQ 服务，并完成简单的消息收发。

系统要求：

64 位操作系统
64 位 JDK 1.8+

1. 拉取 RocketMQ 镜像

这里以dockerhub上 RocketMQ 5.3.0 版本的镜像为例，介绍部署过程。

docker pull apache/rocketmq:5.3.0

2. 创建容器共享网络

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

docker network create rocketmq

3. 启动 NameServer

# 启动 NameServer
docker run -d --name rmqnamesrv -p 9876:9876 --network rocketmq apache/rocketmq:5.3.0 sh mqnamesrv

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

如果我们看到 ‘The Name Server boot success..’， 那表示 NameServer 已成功启动。

4. 启动 Broker + Proxy

NameServer 成功启动后，我们启动 Broker 和 Proxy。

Linux
Windows

# 配置 Broker 的IP地址
echo "brokerIP1=127.0.0.1" > broker.conf

# 启动 Broker 和 Proxy
docker run -d \
--name rmqbroker \
--network rocketmq \
-p 10912:10912 -p 10911:10911 -p 10909:10909 \
-p 8080:8080 -p 8081:8081 \
-e "NAMESRV_ADDR=rmqnamesrv:9876" \
-v ./broker.conf:/home/rocketmq/rocketmq-5.3.0/conf/broker.conf \
apache/rocketmq:5.3.0 sh mqbroker --enable-proxy \
-c /home/rocketmq/rocketmq-5.3.0/conf/broker.conf

# 验证 Broker 是否启动成功
docker exec -it rmqbroker bash -c "tail -n 10 /home/rocketmq/logs/rocketmqlogs/proxy.log"

如果我们可以看到 ‘The broker boot success..’， 表示 Broker 已成功启动。

至此，一个单节点副本的 RocketMQ 集群已经部署起来了，我们可以利用脚本进行简单的消息收发。

5. SDK 测试消息收发

工具测试完成后，我们可以尝试使用 SDK 收发消息。这里以 Java SDK 为例介绍一下消息收发过程，可以从 rocketmq-clients 中参阅更多细节。

在 IDEA 中创建一个 Java 工程。
在 pom.xml 文件中添加以下依赖引入 Java 依赖库，将 rocketmq-client-java-version 替换成最新的版本.

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>${rocketmq-client-java-version}</version>
</dependency>

进入 broker 容器，通过 mqadmin 创建 Topic。

$ docker exec -it rmqbroker bash
$ sh mqadmin updatetopic -t TestTopic -c DefaultCluster

在已创建的 Java 工程中，创建发送普通消息程序并运行，示例代码如下：

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProducerExample {
    private static final Logger logger = LoggerFactory.getLogger(ProducerExample.class);

    public static void main(String[] args) throws ClientException {
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
        // 此处为示例，实际使用时请替换为真实的 Proxy 地址和端口
        String endpoint = "localhost:8081";
        // 消息发送的目标Topic名称，需要提前创建。
        String topic = "TestTopic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();
        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        Producer producer = provider.newProducerBuilder()
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();
        // 普通消息发送。
        Message message = provider.newMessageBuilder()
            .setTopic(topic)
            // 设置消息索引键，可根据关键字精确查找某条消息。
            .setKeys("messageKey")
            // 设置消息Tag，用于消费端根据指定Tag过滤消息。
            .setTag("messageTag")
            // 消息体。
            .setBody("messageBody".getBytes())
            .build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (ClientException e) {
            logger.error("Failed to send message", e);
        }
        // producer.close();
    }
}

在已创建的 Java 工程中，创建订阅普通消息程序并运行。Apache RocketMQ 支持SimpleConsumer和PushConsumer两种消费者类型，您可以选择以下任意一种方式订阅消息。

import java.io.IOException;
import java.util.Collections;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PushConsumerExample {
    private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
        // 此处为示例，实际使用时请替换为真实的 Proxy 地址和端口
        String endpoints = "localhost:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // 为消费者指定所属的消费者分组，Group需要提前创建。
        String consumerGroup = "YourConsumerGroup";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        String topic = "TestTopic";
        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // 设置消费者分组。
            .setConsumerGroup(consumerGroup)
            // 设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // 设置消费监听器。
            .setMessageListener(messageView -> {
                // 处理消息并返回消费结果。
                logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
                return ConsumeResult.SUCCESS;
            })
            .build();
        Thread.sleep(Long.MAX_VALUE);
        // 如果不需要再使用 PushConsumer，可关闭该实例。
        // pushConsumer.close();
    }
}

5. 停止容器

完成实验后，我们可以通过以下方式停止容器。

# 停止 NameServer 容器
docker stop rmqnamesrv

# 停止 Broker 容器
docker stop rmqbroker

通过 Docker Compose 迅速拉起 RocketMQ 集群

这一节介绍如何使用 Docker-compose 快速部署一个单节点单副本 RocketMQ 服务，并完成简单的消息收发。

系统要求：

64 位操作系统
64 位 JDK 1.8+

具体可参照前置准备工作指南进行配置。

1. 编写 docker-compose

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

version: '3.8'
services:
  namesrv:
    image: apache/rocketmq:5.3.0
    container_name: rmqnamesrv
    ports:
      - 9876:9876
    networks:
      - rocketmq
    command: sh mqnamesrv
  broker:
    image: apache/rocketmq:5.3.0
    container_name: rmqbroker
    ports:
      - 10909:10909
      - 10911:10911
      - 10912:10912
    environment:
      - NAMESRV_ADDR=rmqnamesrv:9876
    depends_on:
      - namesrv
    networks:
      - rocketmq
    command: sh mqbroker
  proxy:
    image: apache/rocketmq:5.3.0
    container_name: rmqproxy
    networks:
      - rocketmq
    depends_on:
      - broker
      - namesrv
    ports:
      - 8080:8080
      - 8081:8081
    restart: on-failure
    environment:
      - NAMESRV_ADDR=rmqnamesrv:9876
    command: sh mqproxy
networks:
  rocketmq:
    driver: bridge

2. 启动 RocketMQ 集群

根据 docker-compose.yml 文件启动所有定义的服务。

Linux
Windows

docker-compose up -d

3. SDK 测试消息收发

工具测试完成后，我们可以尝试使用 SDK 收发消息。这里以 Java SDK 为例介绍一下消息收发过程，可以从 rocketmq-clients 中参阅更多细节。

在 IDEA 中创建一个 Java 工程。
在 pom.xml 文件中添加以下依赖引入 Java 依赖库，将 rocketmq-client-java-version 替换成最新的版本.

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>${rocketmq-client-java-version}</version>
</dependency>

进入 broker 容器，通过 mqadmin 创建 Topic。

$ docker exec -it rmqbroker bash
$ sh mqadmin updatetopic -t TestTopic -c DefaultCluster

在已创建的 Java 工程中，创建发送普通消息程序并运行，示例代码如下：

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ProducerExample {
    private static final Logger logger = LoggerFactory.getLogger(ProducerExample.class);

    public static void main(String[] args) throws ClientException {
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
        String endpoint = "localhost:8081";
        // 消息发送的目标Topic名称，需要提前创建。
        String topic = "TestTopic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();
        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        Producer producer = provider.newProducerBuilder()
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();
        // 普通消息发送。
        Message message = provider.newMessageBuilder()
            .setTopic(topic)
            // 设置消息索引键，可根据关键字精确查找某条消息。
            .setKeys("messageKey")
            // 设置消息Tag，用于消费端根据指定Tag过滤消息。
            .setTag("messageTag")
            // 消息体。
            .setBody("messageBody".getBytes())
            .build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (ClientException e) {
            logger.error("Failed to send message", e);
        }
        // producer.close();
    }
}

在已创建的 Java 工程中，创建订阅普通消息程序并运行。Apache RocketMQ 支持SimpleConsumer和PushConsumer两种消费者类型，您可以选择以下任意一种方式订阅消息。

import java.io.IOException;
import java.util.Collections;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PushConsumerExample {
    private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
        String endpoints = "localhost:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // 为消费者指定所属的消费者分组，Group需要提前创建。
        String consumerGroup = "YourConsumerGroup";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        String topic = "TestTopic";
        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // 设置消费者分组。
            .setConsumerGroup(consumerGroup)
            // 设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // 设置消费监听器。
            .setMessageListener(messageView -> {
                // 处理消息并返回消费结果。
                logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
                return ConsumeResult.SUCCESS;
            })
            .build();
        Thread.sleep(Long.MAX_VALUE);
        // 如果不需要再使用 PushConsumer，可关闭该实例。
        // pushConsumer.close();
    }
}

4. 停止所有服务

docker-compose down

迅速部署具备自动切换能力的 RocketMQ 集群

该文档主要介绍如何快速构建自动主从切换的 RocketMQ 集群，其架构如上图所示，主要增加支持自动主从切换的 Controller 组件，其可以独立部署也可以内嵌在 NameServer 中。

1. 前置准备工作

你需要准备好运行 RocketMQ 的环境，以及可运行文件。具体可以参考前置准备工作指南。

2. 快速部署

在你的 RocketMQ 目录下，运行如下脚本：

$ sh bin/controller/fast-try.sh start

如果上面的步骤执行成功，可以通过运维命令查看 Controller 状态。

$ sh bin/mqadmin getControllerMetaData -a localhost:9878

-a 代表集群中任意一个 Controller 的地址

至此，启动成功，现在可以向集群收发消息，并进行切换测试了。

如果需要关闭快速集群，可以执行：

$ sh bin/controller/fast-try.sh stop

对于快速部署，默认配置在 conf/controller/quick-start 里面，默认的存储路径在 /tmp/rmqstore，且会开启一个 Controller (嵌入在 Namesrv) 和两个 Broker。

2.1 查看 SyncStateSet

可以通过运维工具查看 SyncStateSet：

$ sh bin/mqadmin getSyncStateSet -a localhost:9878 -b broker-a

-a 代表的是任意一个 Controller 的地址

如果顺利的话，可以看到以下内容：

2.2 查看 BrokerEpoch

可以通过运维工具查看 BrokerEpochEntry：

$ sh bin/mqadmin getBrokerEpoch -n localhost:9876 -b broker-a

-n 代表的是任意一个 Namesrv 的地址

如果顺利的话，可以看到以下内容：

3. 切换

部署成功后，现在尝试进行 Master 切换。

首先，kill 掉原 Master 的进程，在上文的例子中，就是使用端口 30911 的进程：

#查找端口:
$ ps -ef|grep java|grep BrokerStartup|grep ./conf/controller/quick-start/broker-n0.conf|grep -v grep|awk '{print $2}'
#杀掉 master:
$ kill -9 PID

接着，用 SyncStateSet admin 脚本查看：

$ sh bin/mqadmin getSyncStateSet -a localhost:9878 -b broker-a

可以发现 Master 已经发生了切换。

4. Controller 内嵌 Namesvr 集群部署

Controller 以插件方式内嵌 Namesvr 集群(3 个 Node 组成)部署，快速启动：

$ sh bin/controller/fast-try-namesrv-plugin.sh start

或者通过命令单独启动：

$ nohup bin/mqnamesrv -c ./conf/controller/cluster-3n-namesrv-plugin/namesrv-n0.conf &
$ nohup bin/mqnamesrv -c ./conf/controller/cluster-3n-namesrv-plugin/namesrv-n1.conf &
$ nohup bin/mqnamesrv -c ./conf/controller/cluster-3n-namesrv-plugin/namesrv-n2.conf &

如果上面的步骤执行成功，可以通过运维命令查看 Controller 集群状态。

$ sh bin/mqadmin getControllerMetaData -a localhost:9878

-a代表的是任意一个 Controller 的地址

如果 controller 启动成功可以看到以下内容：

#ControllerGroup        group1
#ControllerLeaderId     n0
#ControllerLeaderAddress        127.0.0.1:9878
#Peer:  n0:127.0.0.1:9878
#Peer:  n1:127.0.0.1:9868
#Peer:  n2:127.0.0.1:9858

启动成功后 Broker Controller 模式部署就能使用 Controller 集群。

如果需要快速停止集群：

$ sh bin/controller/fast-try-namesrv-plugin.sh stop

使用 fast-try-namesrv-plugin.sh 脚本快速部署，默认配置在 conf/controller/cluster-3n-namesrv-plugin 里面并且会启动 3 个 Namesvr 和 3 个 Controller (内嵌 Namesrv)。

5. Controller 独立集群部署

Controller 独立集群(3 个 Node 组成)部署，快速启动：

$ sh bin/controller/fast-try-independent-deployment.sh start

或者通过命令单独启动：

$ nohup bin/mqcontroller -c ./conf/controller/cluster-3n-independent/controller-n0.conf &
$ nohup bin/mqcontroller -c ./conf/controller/cluster-3n-independent/controller-n1.conf &
$ nohup bin/mqcontroller -c ./conf/controller/cluster-3n-independent/controller-n2.conf &

如果上面的步骤执行成功，可以通过运维命令查看 Controller 集群状态。

$ sh bin/mqadmin getControllerMetaData -a localhost:9878

-a 代表的是任意一个 Controller 的地址

如果 Controller 启动成功可以看到以下内容：

#ControllerGroup        group1
#ControllerLeaderId     n1
#ControllerLeaderAddress        127.0.0.1:9868
#Peer:  n0:127.0.0.1:9878
#Peer:  n1:127.0.0.1:9868
#Peer:  n2:127.0.0.1:9858

启动成功后 Broker Controller 模式部署就能使用 Controller 集群。

如果需要快速停止集群：

$ sh bin/controller/fast-try-independent-deployment.sh stop

使用 fast-try-independent-deployment.sh 脚本快速部署，默认配置在 conf/controller/cluster-3n-independent 里面并且会启动 3 个 Controller (独立部署)组成一个集群。

6. 注意说明

若需要保证 Controller 具备容错能力，Controller 部署需要三副本及以上（遵循 Raft 的多数派协议）
Controller 部署配置文件中配置参数controllerDLegerPeers 中的 IP 地址配置成其他节点能够访问的 IP，在多机器部署的时候尤为重要。例子仅供参考需要根据实际情况进行修改调整。