Docker 部署 RocketMQ

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

Docker 部署 RocketMQ - 图1系统要求

  1. 64位操作系统
  2. 64位 JDK 1.8+

1.拉取RocketMQ镜像

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

  1. docker pull apache/rocketmq:5.2.0

2.创建容器共享网络

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

  1. docker network create rocketmq

3.启动NameServer

  1. # 启动 NameServer
  2. docker run -d --name rmqnamesrv -p 9876:9876 --network rocketmq apache/rocketmq:5.2.0 sh mqnamesrv
  4. # 验证 NameServer 是否启动成功
  5. docker logs -f rmqnamesrv

Docker 部署 RocketMQ - 图2信息

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

4.启动 Broker+Proxy

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

  • Linux
  • Windows
  1. # 配置 Broker 的IP地址
  2. echo "brokerIP1=127.0.0.1" > broker.conf
  4. # 启动 Broker 和 Proxy
  5. docker run -d \
  6. --name rmqbroker \
  7. --network rocketmq \
  8. -p 10912:10912 -p 10911:10911 -p 10909:10909 \
  9. -p 8080:8080 -p 8081:8081 \
  10. -e "NAMESRV_ADDR=rmqnamesrv:9876" \
  11. -v ./broker.conf:/home/rocketmq/rocketmq-5.2.0/conf/broker.conf \
  12. apache/rocketmq:5.2.0 sh mqbroker --enable-proxy \
  13. -c /home/rocketmq/rocketmq-5.2.0/conf/broker.conf
  15. # 验证 Broker 是否启动成功
  16. docker exec -it rmqbroker bash -c "tail -n 10 /home/rocketmq/logs/rocketmqlogs/proxy.log"
  1. # 配置 Broker 的 IP 地址
  2. echo "brokerIP1=127.0.0.1" > broker.conf
  4. # 启动 Broker 和 Proxy
  5. docker run -d ^
  6. --name rmqbroker ^
  7. --net rocketmq ^
  8. -p 10912:10912 -p 10911:10911 -p 10909:10909 ^
  9. -p 8080:8080 -p 8081:8081 \
  10. -e "NAMESRV_ADDR=rmqnamesrv:9876" ^
  11. -v %cd%\broker.conf:/home/rocketmq/rocketmq-5.2.0/conf/broker.conf ^
  12. apache/rocketmq:5.2.0 sh mqbroker --enable-proxy \
  13. -c /home/rocketmq/rocketmq-5.2.0/conf/broker.conf
  15. # 验证 Broker 是否启动成功
  16. docker exec -it rmqbroker bash -c "tail -n 10 /home/rocketmq/logs/rocketmqlogs/proxy.log"

Docker 部署 RocketMQ - 图3信息

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

Docker 部署 RocketMQ - 图4备注

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

5.SDK测试消息收发

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

  1. 在IDEA中创建一个Java工程。

  2. 在 pom.xml 文件中添加以下依赖引入Java依赖库，将 rocketmq-client-java-version 替换成 最新的版本.

    1. <dependency>
    2.     <groupId>org.apache.rocketmq</groupId>
    3.     <artifactId>rocketmq-client-java</artifactId>
    4.     <version>${rocketmq-client-java-version}</version>
    5. </dependency>

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

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

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

    1. import org.apache.rocketmq.client.apis.ClientConfiguration;
    2. import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
    3. import org.apache.rocketmq.client.apis.ClientException;
    4. import org.apache.rocketmq.client.apis.ClientServiceProvider;
    5. import org.apache.rocketmq.client.apis.message.Message;
    6. import org.apache.rocketmq.client.apis.producer.Producer;
    7. import org.apache.rocketmq.client.apis.producer.SendReceipt;
    8. import org.slf4j.Logger;
    9. import org.slf4j.LoggerFactory;
    11. public class ProducerExample {
    12.     private static final Logger logger = LoggerFactory.getLogger(ProducerExample.class);
    14.     public static void main(String[] args) throws ClientException {
    15.         // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
    16.         // 此处为示例，实际使用时请替换为真实的 Proxy 地址和端口
    17.         String endpoint = "localhost:8081";
    18.         // 消息发送的目标Topic名称，需要提前创建。
    19.         String topic = "TestTopic";
    20.         ClientServiceProvider provider = ClientServiceProvider.loadService();
    21.         ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
    22.         ClientConfiguration configuration = builder.build();
    23.         // 初始化Producer时需要设置通信配置以及预绑定的Topic。
    24.         Producer producer = provider.newProducerBuilder()
    25.             .setTopics(topic)
    26.             .setClientConfiguration(configuration)
    27.             .build();
    28.         // 普通消息发送。
    29.         Message message = provider.newMessageBuilder()
    30.             .setTopic(topic)
    31.             // 设置消息索引键，可根据关键字精确查找某条消息。
    32.             .setKeys("messageKey")
    33.             // 设置消息Tag，用于消费端根据指定Tag过滤消息。
    34.             .setTag("messageTag")
    35.             // 消息体。
    36.             .setBody("messageBody".getBytes())
    37.             .build();
    38.         try {
    39.             // 发送消息，需要关注发送结果，并捕获失败等异常。
    40.             SendReceipt sendReceipt = producer.send(message);
    41.             logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
    42.         } catch (ClientException e) {
    43.             logger.error("Failed to send message", e);
    44.         }
    45.         // producer.close();
    46.     }
    47. }

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

  1. import java.io.IOException;
  2. import java.util.Collections;
  3. import org.apache.rocketmq.client.apis.ClientConfiguration;
  4. import org.apache.rocketmq.client.apis.ClientException;
  5. import org.apache.rocketmq.client.apis.ClientServiceProvider;
  6. import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
  7. import org.apache.rocketmq.client.apis.consumer.FilterExpression;
  8. import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
  9. import org.apache.rocketmq.client.apis.consumer.PushConsumer;
  10. import org.slf4j.Logger;
  11. import org.slf4j.LoggerFactory;
  13. public class PushConsumerExample {
  14.     private static final Logger logger = LoggerFactory.getLogger(PushConsumerExample.class);
  16.     private PushConsumerExample() {
  17.     }
  19.     public static void main(String[] args) throws ClientException, IOException, InterruptedException {
  20.         final ClientServiceProvider provider = ClientServiceProvider.loadService();
  21.         // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8080;xxx:8081
  22.         // 此处为示例，实际使用时请替换为真实的 Proxy 地址和端口
  23.         String endpoints = "localhost:8081";
  24.         ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
  25.             .setEndpoints(endpoints)
  26.             .build();
  27.         // 订阅消息的过滤规则，表示订阅所有Tag的消息。
  28.         String tag = "*";
  29.         FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
  30.         // 为消费者指定所属的消费者分组，Group需要提前创建。
  31.         String consumerGroup = "YourConsumerGroup";
  32.         // 指定需要订阅哪个目标Topic，Topic需要提前创建。
  33.         String topic = "TestTopic";
  34.         // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
  35.         PushConsumer pushConsumer = provider.newPushConsumerBuilder()
  36.             .setClientConfiguration(clientConfiguration)
  37.             // 设置消费者分组。
  38.             .setConsumerGroup(consumerGroup)
  39.             // 设置预绑定的订阅关系。
  40.             .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
  41.             // 设置消费监听器。
  42.             .setMessageListener(messageView -> {
  43.                 // 处理消息并返回消费结果。
  44.                 logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
  45.                 return ConsumeResult.SUCCESS;
  46.             })
  47.             .build();
  48.         Thread.sleep(Long.MAX_VALUE);
  49.         // 如果不需要再使用 PushConsumer，可关闭该实例。
  50.         // pushConsumer.close();
  51.     }
  52. }

6. 停止容器

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

  1. # 停止 NameServer 容器
  2. docker stop rmqnamesrv
  4. # 停止 Broker 容器
  5. docker stop rmqbroker