Docker Compose 部署 RocketMQ

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

Docker Compose 部署 RocketMQ - 图1系统要求

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

1.配置 broker.conf

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

2.编写docker-compose

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

  1. version: '3.8'
  2. services:
  3. namesrv:
  4. image: apache/rocketmq:4.9.6
  5. container_name: rmqnamesrv
  6. ports:
  7. - 9876:9876
  8. networks:
  9. - rocketmq
  10. command: sh mqnamesrv
  11. broker:
  12. image: apache/rocketmq:4.9.6
  13. container_name: rmqbroker
  14. ports:
  15. - 10909:10909
  16. - 10911:10911
  17. - 10912:10912
  18. environment:
  19. - NAMESRV_ADDR=rmqnamesrv:9876
  20. volumes:
  21. - ./broker.conf:/home/rocketmq/rocketmq-4.9.6/conf/broker.conf
  22. depends_on:
  23. - namesrv
  24. networks:
  25. - rocketmq
  26. command: sh mqbroker -c /home/rocketmq/rocketmq-4.9.6/conf/broker.conf
  27. networks:
  28. rocketmq:
  29. driver: bridge

3.启动RocketMQ集群

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

  • Linux
  • Windows
  1. docker-compose up -d
  1. docker-compose -p rocketmq_project up -d

4.工具测试消息收发

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

5.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;
    4. public class ProducerExample {
    5. public static void main(String[] args) throws Exception {
    6. // 创建生产者实例,并设置生产者组名
    7. DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    8. // 设置 Name Server 地址,此处为示例,实际使用时请替换为真实的 Name Server 地址
    9. producer.setNamesrvAddr("localhost:9876");
    10. producer.start();
    11. try {
    12. // 创建消息实例,指定 topic、Tag和消息体
    13. Message msg = new Message("TestTopic", "TagA", ("Hello RocketMQ").getBytes());
    14. // 发送消息并获取发送结果
    15. SendResult sendResult = producer.send(msg);
    16. System.out.println("Message sent: " + new String(msg.getBody()));
    17. System.out.println("Send result: " + sendResult);
    18. } catch (Exception e) {
    19. e.printStackTrace();
    20. System.out.println("Message sending failed.");
    21. } finally {
    22. // 关闭生产者
    23. producer.shutdown();
    24. }
    25. }
    26. }
  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;
    6. import java.util.List;
    7. public class ConsumerExample {
    8. public static void main(String[] args) throws Exception {
    9. // 创建消费者实例,并设置消费者组名
    10. DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
    11. // 设置 Name Server 地址,此处为示例,实际使用时请替换为真实的 Name Server 地址
    12. consumer.setNamesrvAddr("localhost:9876");
    13. // 订阅指定的主题和标签(* 表示所有标签)
    14. consumer.subscribe("TestTopic", "*");
    15. // 注册消息监听器
    16. consumer.registerMessageListener(new MessageListenerConcurrently() {
    17. @Override
    18. public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    19. for (MessageExt msg : msgs) {
    20. System.out.println("Received message: " + new String(msg.getBody()));
    21. }
    22. return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    23. }
    24. });
    25. // 启动消费者
    26. consumer.start();
    27. System.out.println("Consumer started.");
    28. }
    29. }

6.停止所有服务

  1. docker-compose down