在 Docker 中运行 RocketMQ

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

系统要求

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

1. 拉取 RocketMQ 镜像

这里我们以 dockerhub 上的 RocketMQ 5.3.2 版本镜像为例来介绍部署流程。

docker pull apache/rocketmq:5.3.2

2. 为容器创建共享网络

RocketMQ 涉及多个服务，需要多个容器。创建 Docker 网络有助于容器间通信。

docker network create rocketmq

3. 启动 NameServer

# Start NameServer
docker run -d --name rmqnamesrv -p 9876:9876 --network rocketmq apache/rocketmq:5.3.2 sh mqnamesrv

# Verify if NameServer started successfully
docker logs -f rmqnamesrv

信息

当我们从 namesrv.log 中看到 'The Name Server boot success..' 时，表示 NameServer 已经成功启动。

4. 启动 Broker 和 Proxy

NameServer 启动后，我们继续启动 Broker 和 Proxy。

Linux

# Configure the broker's IP address
echo "brokerIP1=127.0.0.1" > broker.conf

# Start the Broker and 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" \
# In PowerShell, replace %cd% with $pwd
-v ./broker.conf:/home/rocketmq/rocketmq-5.3.2/conf/broker.conf \
apache/rocketmq:5.3.2 sh mqbroker --enable-proxy \
-c /home/rocketmq/rocketmq-5.3.2/conf/broker.conf

# Verify if Broker started successfully
docker exec -it rmqbroker bash -c "tail -n 10 /home/rocketmq/logs/rocketmqlogs/proxy.log"

Windows

# Configure the broker's IP address
echo "brokerIP1=127.0.0.1" > broker.conf

# Start the Broker and Proxy
docker run -d ^
--name rmqbroker ^
--net rocketmq ^
-p 10912:10912 -p 10911:10911 -p 10909:10909 ^
-p 8080:8080 -p 8081:8081 \
-e "NAMESRV_ADDR=rmqnamesrv:9876" ^
-v %cd%\broker.conf:/home/rocketmq/rocketmq-5.3.2/conf/broker.conf ^
apache/rocketmq:5.3.2 sh mqbroker --enable-proxy \
-c /home/rocketmq/rocketmq-5.3.2/conf/broker.conf

# Verify if Broker started successfully
docker exec -it rmqbroker bash -c "tail -n 10 /home/rocketmq/logs/rocketmqlogs/proxy.log"

信息

当我们从 proxy.log 中看到 'The broker[brokerName,ip:port] boot success..' 时，表示 Broker 已经成功启动。

注意

至此，已部署一个单 Master 的 RocketMQ 集群，我们可以发送和接收简单消息。

5. 使用 SDK 发送和接收消息

我们还可以尝试使用客户端 SDK 发送和接收消息，更多详细信息请参阅 rocketmq-clients。

1. 创建 Java 项目。

2. 将 SDK 依赖添加到 pom.xml 中，请记住将 rocketmq-client-java-version 替换为最新版本。

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

3. 进入 Broker 容器并使用 mqadmin 创建 Topic。

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

4. 在创建的 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 {
           // Endpoint address, set to the Proxy address and port list, usually xxx:8080;xxx:8081
           String endpoint = "localhost:8081";
           // The target topic name for message sending, which needs to be created in advance.
           String topic = "TestTopic";
           ClientServiceProvider provider = ClientServiceProvider.loadService();
           ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
           ClientConfiguration configuration = builder.build();
           // When initializing Producer, communication configuration and pre-bound Topic need to be set.
           Producer producer = provider.newProducerBuilder()
               .setTopics(topic)
               .setClientConfiguration(configuration)
               .build();
           // Sending a normal message.
           Message message = provider.newMessageBuilder()
               .setTopic(topic)
               // Set the message index key, which can be used to accurately find a specific message.
               .setKeys("messageKey")
               // Set the message Tag, used by the consumer to filter messages by specified Tag.
               .setTag("messageTag")
               // Message body
               .setBody("messageBody".getBytes())
               .build();
           try {
               // Send the message, paying attention to the sending result and catching exceptions.
               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();
       }
   }

5. 在创建的 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();
        // Endpoint address, set to the Proxy address and port list, usually xxx:8080;xxx:8081
        String endpoints = "localhost:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .build();
        // Subscription message filtering rule, indicating subscription to all Tag messages.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Specify the consumer group the consumer belongs to, Group needs to be created in advance.
        String consumerGroup = "YourConsumerGroup";
        // Specify which target Topic to subscribe to, Topic needs to be created in advance.
        String topic = "TestTopic";
        // Initialize PushConsumer
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // Set the consumer group
            .setConsumerGroup(consumerGroup)
            // Set pre-bound subscription relationship
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // Set the message listener
            .setMessageListener(messageView -> {
                // Handle messages and return the consumption result
                logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
                return ConsumeResult.SUCCESS;
            })
            .build();
        Thread.sleep(Long.MAX_VALUE);
        // If PushConsumer is no longer needed, this instance can be closed.
        // pushConsumer.close();
    }
}

6. 停止容器

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

# Stop the NameServer container
docker stop rmqnamesrv

# Stop the Broker container
docker stop rmqbroker