Docker 部署 RocketMQ

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

Docker 部署 RocketMQ - 图1系统要求

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

1.拉取RocketMQ镜像

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

  1. docker pull apache/rocketmq:4.9.6

2.创建容器共享网络

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

  1. docker network create rocketmq

3.启动NameServer

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

Docker 部署 RocketMQ - 图2信息

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

4.启动Broker

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

  • Linux
  • Windows
  1. # 配置 Broker 的 IP 地址
  2. echo "brokerIP1=127.0.0.1" >broker.conf
  4. # 启动 Broker
  5. docker run -d \
  6. --name rmqbroker \
  7. --net rocketmq \
  8. -p 10912:10912 -p 10911:10911 -p 10909:10909 \
  9. -e "NAMESRV_ADDR=rmqnamesrv:9876" \
  10. -v ./broker.conf:/home/rocketmq/rocketmq-4.9.6/conf/broker.conf \
  11. apache/rocketmq:4.9.6 sh mqbroker \
  12. -c /home/rocketmq/rocketmq-4.9.6/conf/broker.conf
  14. # 验证 Broker 是否启动成功
  15. docker logs rmqbroker
  1. # 配置 Broker 的 IP 地址
  2. echo "brokerIP1=127.0.0.1" >broker.conf
  4. # 启动 Broker
  5. docker run -d ^
  6. --name rmqbroker ^
  7. --net rocketmq ^
  8. -p 10912:10912 -p 10911:10911 -p 10909:10909 ^
  9. -e "NAMESRV_ADDR=rmqnamesrv:9876" ^
  10. -v %cd%\broker.conf:/home/rocketmq/rocketmq-4.9.6/conf/broker.conf ^
  11. apache/rocketmq:4.9.6 sh mqbroker ^
  12. -c /home/rocketmq/rocketmq-4.9.6/conf/broker.conf
  14. # 验证 Broker 是否启动成功
  15. docker logs rmqbroker

Docker 部署 RocketMQ - 图3信息

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

Docker 部署 RocketMQ - 图4备注

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

5.工具测试消息收发

  1. # 进入broker容器
  2. $ docker exec -it rmqbroker bash
  4. $ sh tools.sh org.apache.rocketmq.example.quickstart.Producer
  5.  SendResult [sendStatus=SEND_OK, msgId= ...
  7. $ sh tools.sh org.apache.rocketmq.example.quickstart.Consumer
  8.  ConsumeMessageThread_%d Receive New Messages: [MessageExt...

6.SDK测试消息收发

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

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

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

    1. <dependency>
    2.     <groupId>org.apache.rocketmq</groupId>
    3.     <artifactId>rocketmq-client</artifactId>
    4.     <version>4.9.6</version>
    5. </dependency>

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

    1. import org.apache.rocketmq.client.producer.DefaultMQProducer;
    2. import org.apache.rocketmq.client.producer.SendResult;
    3. import org.apache.rocketmq.common.message.Message;
    5. public class ProducerExample {
    6.     public static void main(String[] args) throws Exception {
    7.         // 创建生产者实例，并设置生产者组名
    8.         DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    9.         // 设置 Name Server 地址，此处为示例，实际使用时请替换为真实的 Name Server 地址
    10.         producer.setNamesrvAddr("localhost:9876");
    11.         producer.start();
    13.         try {
    14.             // 创建消息实例，指定 topic、Tag和消息体
    15.             Message msg = new Message("TestTopic", "TagA", ("Hello RocketMQ").getBytes());
    16.             // 发送消息并获取发送结果
    17.             SendResult sendResult = producer.send(msg);
    18.             System.out.println("Message sent: " + new String(msg.getBody()));
    19.             System.out.println("Send result: " + sendResult);
    20.         } catch (Exception e) {
    21.             e.printStackTrace();
    22.             System.out.println("Message sending failed.");
    23.         } finally {
    24.             // 关闭生产者
    25.             producer.shutdown();
    26.         }
    27.     }
    28. }

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

    1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    2. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    3. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    4. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    5. import org.apache.rocketmq.common.message.MessageExt;
    7. import java.util.List;
    9. public class ConsumerExample {
    10.     public static void main(String[] args) throws Exception {
    11.         // 创建消费者实例，并设置消费者组名
    12.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
    13.         // 设置 Name Server 地址，此处为示例，实际使用时请替换为真实的 Name Server 地址
    14.         consumer.setNamesrvAddr("localhost:9876");
    15.         // 订阅指定的主题和标签（* 表示所有标签）
    16.         consumer.subscribe("TestTopic", "*");
    18.         // 注册消息监听器
    19.         consumer.registerMessageListener(new MessageListenerConcurrently() {
    20.             @Override
    21.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    22.                 for (MessageExt msg : msgs) {
    23.                     System.out.println("Received message: " + new String(msg.getBody()));
    24.                 }
    25.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    26.             }
    27.         });
    29.         // 启动消费者
    30.         consumer.start();
    31.         System.out.println("Consumer started.");
    32.     }
    33. }

7. 停止容器

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

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