本地部署 RocketMQ

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

本地部署 RocketMQ - 图1系统要求

  1. 64位操作系统，推荐 Linux/Unix/macOS
  2. 64位 JDK 1.8+

1.下载安装Apache RocketMQ

本地部署 RocketMQ - 图2RocketMQ下载

RocketMQ 的安装包分为两种，二进制包和源码包。 点击这里 下载 Apache RocketMQ 5.2.0的源码包。你也可以从这里 下载到二进制包。二进制包是已经编译完成后可以直接运行的，源码包是需要编译后运行的。

这里以在Linux环境下利用社区5.2.0的源码包为例，介绍RocketMQ安装过程。

解压5.2.0的源码包并编译构建二进制可执行文件

  1. $ unzip rocketmq-all-5.2.0-source-release.zip
  2. $ cd rocketmq-all-5.2.0-source-release/
  3. $ mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U
  4. $ cd distribution/target/rocketmq-5.2.0/rocketmq-5.2.0

2. 启动NameServer

安装完RocketMQ包后，我们启动NameServer

  1. ### 启动namesrv
  2. $ nohup sh bin/mqnamesrv &
  4. ### 验证namesrv是否启动成功
  5. $ tail -f ~/logs/rocketmqlogs/namesrv.log
  6. The Name Server boot success...

本地部署 RocketMQ - 图3信息

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

3. 启动Broker+Proxy

NameServer成功启动后，我们启动Broker和Proxy。这里我们使用 Local 模式部署，即 Broker 和 Proxy 同进程部署。5.x 版本也支持 Broker 和 Proxy 分离部署以实现更灵活的集群能力。详情参考部署教程。

  1. ### 先启动broker
  2. $ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &
  4. ### 验证broker是否启动成功, 比如, broker的ip是192.168.1.2 然后名字是broker-a
  5. $ tail -f ~/logs/rocketmqlogs/proxy.log 
  6. The broker[broker-a,192.169.1.2:10911] boot success...

本地部署 RocketMQ - 图4信息

我们可以在 proxy.log 中看到“The broker[brokerName,ip:port] boot success..”，这表明 broker 已成功启动。

本地部署 RocketMQ - 图5备注

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

4. 工具测试消息收发

在进行工具测试消息收发之前，我们需要告诉客户端NameServer的地址，RocketMQ有多种方式在客户端中设置NameServer地址，这里我们利用环境变量NAMESRV_ADDR

  1. $ export NAMESRV_ADDR=localhost:9876
  2. $ sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
  3.  SendResult [sendStatus=SEND_OK, msgId= ...
  5. $ sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
  6.  ConsumeMessageThread_%d Receive New Messages: [MessageExt...

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. 通过mqadmin创建 Topic。

    1. $ sh bin/mqadmin updatetopic -n localhost:9876 -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.         String endpoint = "localhost:8081";
    17.         // 消息发送的目标Topic名称，需要提前创建。
    18.         String topic = "TestTopic";
    19.         ClientServiceProvider provider = ClientServiceProvider.loadService();
    20.         ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
    21.         ClientConfiguration configuration = builder.build();
    22.         // 初始化Producer时需要设置通信配置以及预绑定的Topic。
    23.         Producer producer = provider.newProducerBuilder()
    24.             .setTopics(topic)
    25.             .setClientConfiguration(configuration)
    26.             .build();
    27.         // 普通消息发送。
    28.         Message message = provider.newMessageBuilder()
    29.             .setTopic(topic)
    30.             // 设置消息索引键，可根据关键字精确查找某条消息。
    31.             .setKeys("messageKey")
    32.             // 设置消息Tag，用于消费端根据指定Tag过滤消息。
    33.             .setTag("messageTag")
    34.             // 消息体。
    35.             .setBody("messageBody".getBytes())
    36.             .build();
    37.         try {
    38.             // 发送消息，需要关注发送结果，并捕获失败等异常。
    39.             SendReceipt sendReceipt = producer.send(message);
    40.             logger.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
    41.         } catch (ClientException e) {
    42.             logger.error("Failed to send message", e);
    43.         }
    44.         // producer.close();
    45.     }
    46. }

  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:8081;xxx:8081。
    22.         String endpoints = "localhost:8081";
    23.         ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
    24.             .setEndpoints(endpoints)
    25.             .build();
    26.         // 订阅消息的过滤规则，表示订阅所有Tag的消息。
    27.         String tag = "*";
    28.         FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
    29.         // 为消费者指定所属的消费者分组，Group需要提前创建。
    30.         String consumerGroup = "YourConsumerGroup";
    31.         // 指定需要订阅哪个目标Topic，Topic需要提前创建。
    32.         String topic = "TestTopic";
    33.         // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
    34.         PushConsumer pushConsumer = provider.newPushConsumerBuilder()
    35.             .setClientConfiguration(clientConfiguration)
    36.             // 设置消费者分组。
    37.             .setConsumerGroup(consumerGroup)
    38.             // 设置预绑定的订阅关系。
    39.             .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
    40.             // 设置消费监听器。
    41.             .setMessageListener(messageView -> {
    42.                 // 处理消息并返回消费结果。
    43.                 logger.info("Consume message successfully, messageId={}", messageView.getMessageId());
    44.                 return ConsumeResult.SUCCESS;
    45.             })
    46.             .build();
    47.         Thread.sleep(Long.MAX_VALUE);
    48.         // 如果不需要再使用 PushConsumer，可关闭该实例。
    49.         // pushConsumer.close();
    50.     }
    51. }

6. 关闭服务器

完成实验后，我们可以通过以下方式关闭服务

  1. $ sh bin/mqshutdown broker
  2. The mqbroker(36695) is running...
  3. Send shutdown request to mqbroker with proxy enable OK(36695)
  6. $ sh bin/mqshutdown namesrv
  7. The mqnamesrv(36664) is running...
  8. Send shutdown request to mqnamesrv(36664) OK