RocketMQ快速开始

前置推荐阅读：RocketMQ简介-CSDN博客

本地部署 RocketMQ

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

系统要求

64位操作系统，推荐 Linux/Unix/macOS
64位 JDK 1.8+

1.下载安装Apache RocketMQ

RocketMQ下载

RocketMQ 的安装包分为两种，二进制包和源码包。点击这里下载 Apache RocketMQ 5.3.1的源码包。你也可以从这里下载到二进制包。二进制包是已经编译完成后可以直接运行的，源码包是需要编译后运行的。

这里以在Linux环境下利用社区5.3.1的源码包为例，介绍RocketMQ安装过程。

解压5.3.1的源码包并编译构建二进制可执行文件

$ unzip rocketmq-all-5.3.1-source-release.zip
$ cd rocketmq-all-5.3.1-source-release/
$ mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
$ cd distribution/target/rocketmq-5.3.1/rocketmq-5.3.1

2. 启动NameServer

安装完RocketMQ包后，我们启动NameServer

### 启动namesrv
$ nohup sh bin/mqnamesrv &### 验证namesrv是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

信息

我们可以在namesrv.log 中看到 'The Name Server boot success..'， 表示NameServer 已成功启动。

3. 启动Broker+Proxy

NameServer成功启动后，我们启动Broker和Proxy。这里我们使用 Local 模式部署，即 Broker 和 Proxy 同进程部署。5.x 版本也支持 Broker 和 Proxy 分离部署以实现更灵活的集群能力。详情参考部署教程。

### 先启动broker
$ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &### 验证broker是否启动成功, 比如, broker的ip是192.168.1.2 然后名字是broker-a
$ tail -f ~/logs/rocketmqlogs/proxy.log 
The broker[broker-a,192.169.1.2:10911] boot success...

信息

我们可以在 proxy.log 中看到“The broker[brokerName,ip:port] boot success..”，这表明 broker 已成功启动。

备注

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

4. 工具测试消息收发

在进行工具测试消息收发之前，我们需要告诉客户端NameServer的地址，RocketMQ有多种方式在客户端中设置NameServer地址，这里我们利用环境变量NAMESRV_ADDR

$ export NAMESRV_ADDR=localhost:9876
$ sh bin/tools.sh org.apache.rocketmq.example.quickstart.ProducerSendResult [sendStatus=SEND_OK, msgId= ...$ sh bin/tools.sh org.apache.rocketmq.example.quickstart.ConsumerConsumeMessageThread_%d Receive New Messages: [MessageExt...

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>

通过mqadmin创建 Topic。

$ sh bin/mqadmin updatetopic -n localhost:9876 -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:8081;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();}
}

6. 关闭服务器

完成实验后，我们可以通过以下方式关闭服务

$ sh bin/mqshutdown broker
The mqbroker(36695) is running...
Send shutdown request to mqbroker with proxy enable OK(36695)$ sh bin/mqshutdown namesrv
The mqnamesrv(36664) is running...
Send shutdown request to mqnamesrv(36664) OK

Docker 部署 RocketMQ

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

系统要求

64位操作系统
64位 JDK 1.8+
docker参考：windows10 vagrant+VirtualBox安装docker_vagrant 6.0下载-CSDN博客

1.拉取RocketMQ镜像

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

docker pull apache/rocketmq:5.3.1

2.创建容器共享网络

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

docker network create rocketmq

3.启动NameServer

# 启动 NameServer
docker run -d --name rmqnamesrv -p 9876:9876 --network rocketmq apache/rocketmq:5.3.1 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.1/conf/broker.conf \
apache/rocketmq:5.3.1 sh mqbroker --enable-proxy \
-c /home/rocketmq/rocketmq-5.3.1/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();}
}

6. 停止容器

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

# 停止 NameServer 容器
docker stop rmqnamesrv# 停止 Broker 容器
docker stop rmqbroker

RocketMQ Dashboard

RocketMQ Dashboard 是 RocketMQ 的管控利器，为用户提供客户端和应用程序的各种事件、性能的统计信息，支持以可视化工具代替 Topic 配置、Broker 管理等命令行操作。

介绍

功能概览

面板	功能
运维	修改nameserver 地址; 选用 `VIPChannel`
驾驶舱	查看 broker, topic 消息量
集群	集群分布，broker 配置、运行信息
主题	搜索、筛选、删除、更新/新增主题，消息路由，发送消息，重置消费位点
消费者	搜索、删除、新增/更新消费者组，终端，消费详情，配置
消息	消息记录，私信消息，消息轨迹等消息详情

操作面板：

快速开始

系统要求：

Linux/Unix/Mac
64bit JDK 1.8+
Maven 3.2.x
启动 RocketMQ

网络配置：

云服务器可远程访问或本地虚拟机可 PING 通外网
rocketmq 配置文件 broker.conf / broker-x.properties 设置 nameserver 地址和端口号
用配置文件启动 broker

1. docker 镜像安装

① 安装docker，拉取 rocketmq-dashboard 镜像

$ docker pull apacherocketmq/rocketmq-dashboard:latest

② docker 容器中运行 rocketmq-dashboard

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

提示

namesrv.addr:port 替换为 rocketmq 中配置的 nameserver 地址：端口号

开放端口号：8080，9876，10911，11011 端口

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

2. 源码安装

源码地址：apache/rocketmq-dashboard

下载并解压，切换至源码目录 rocketmq-dashboard-master/

① 编译 rocketmq-dashboard

$ mvn clean package -Dmaven.test.skip=true

② 运行 rocketmq-dashboard

$ java -jar target/rocketmq-dashboard-1.0.1-SNAPSHOT.jar

提示：Started App in x.xxx seconds (JVM running for x.xxx) 启动成功

浏览器页面访问：namesrv.addr:8080

关闭 rocketmq-dashboard : ctrl + c

再次启动：执行 ②

tips：下载后的源码需要上传到 Linux 系统上编译，本地编译可能会报错。

使用教程

1. 创建主题 Topic

主题 > 新增/更新

2. 创建消费者组 consumer

消费者 > 新增/更新

3. 重置消费位点

主题 > 重置消费位点

tips:

集群消费支持重置消费位点, 广播模式不支持.
消费者不在线不能重置消费位点

4. 扩容 Topic 队列

主题 > TOPIC配置

5. 扩容 Broker

安装部署一个新的 broker, nameserver 地址和当前集群一样

更新 Topic 的BROKER_NAME

主题 > 新增/更新 > BROKER_NAME

6. 发送消息

向指定 Topic 发送消息

主题 > 发送消息

发送结果