使用 Docker Compose 运行 RocketMQ
本节介绍如何使用 Docker Compose 快速部署单节点、单副本的 RocketMQ 服务，并完成简单的消息收发。
系统要求
- 64位操作系统
- 64位 JDK 1.8+
1. 编写 docker-compose
为快速启动和运行 RocketMQ 集群，您可以使用以下模板创建 docker-compose.yml 文件，通过修改或添加环境配置来使用。
version: '3.8'
services:
namesrv:
image: apache/rocketmq:5.3.2
container_name: rmqnamesrv
ports:
- 9876:9876
networks:
- rocketmq
command: sh mqnamesrv
broker:
image: apache/rocketmq:5.3.2
container_name: rmqbroker
ports:
- 10909:10909
- 10911:10911
- 10912:10912
environment:
- NAMESRV_ADDR=rmqnamesrv:9876
depends_on:
- namesrv
networks:
- rocketmq
command: sh mqbroker
proxy:
image: apache/rocketmq:5.3.2
container_name: rmqproxy
networks:
- rocketmq
depends_on:
- broker
- namesrv
ports:
- 8080:8080
- 8081:8081
restart: on-failure
environment:
- NAMESRV_ADDR=rmqnamesrv:9876
command: sh mqproxy
networks:
rocketmq:
driver: bridge
2. 启动 RocketMQ 集群
根据 docker-compose.yml 文件启动所有已定义的服务。
- Linux
- Windows
docker-compose up -d
docker-compose -p rockermq_project up -d
3. 使用 SDK 发送和接收消息
通过工具测试后，我们可以尝试使用 SDK 发送和接收消息。这里是使用 Java SDK 进行消息发送和接收的示例。更多详情请参见 rocketmq-clients。
将以下依赖项添加到 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 {
// 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();
}
}在创建的 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();
}
}
4. 停止所有服务
docker-compose down