RocketMQ安装与简单消息收发

RocketMQ是由阿里捐赠给Apache的一款低延迟、高并发、高可用、高可靠的分布式消息中间件。经历了淘宝双十一的洗礼。RocketMQ既可为分布式应用系统提供异步解耦和削峰填谷的能力，同时也具备互联网应用所需的海量消息堆积、高吞吐、可靠重试等特性。

本文主要介绍的是 RocketMQ的安装与简单的使用测试 。

安装

1、系统要求

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

2、下载安装

RocketMQ 的安装包分为两种，二进制包和源码包。二进制包是已经编译完成后可以直接运行的，源码包是需要编译后运行的。

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

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

$ unzip rocketmq-all-5.1.3-source-release.zip

$ cd rocketmq-all-5.1.3-source-release/

$ mvn -Prelease-all -DskipTests -Dspotbugs.skip=true clean install -U

$ cd distribution/target/rocketmq-5.1.3/rocketmq-5.1.3

3、 启动NameServer

安装完RocketMQ包后，我们启动NameServer
cd 进入 rocketmq-all-5.1.2-source-release --> distribution --> target --> rocketmq-5.1.2 --> rocketmq-5.1.2 文件夹目录，执行以下命令 :

    ### 启动namesrv
    $ nohup sh bin/mqnamesrv &

    ### 验证namesrv是否启动成功
    $ tail -f ~/logs/rocketmqlogs/namesrv.log
    The Name Server boot success...

4、启动Broker+Proxy

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

与启动NameServer 同级目录下，执行以下命令：

### 先启动broker
$ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &

### 验证broker是否启动成功, 比如, broker的ip是192.168.1.2 然后名字是broker-a
$ tail -f ~/logs/rocketmqlogs/proxy.log 
The broker[broker-a,192.169.1.2:10911] boot success...

也可在此文件夹下可执行 jps 命令查看服务启动情况：

zhanghua@Breeze rocketmq-5.1.2 % jps 
1284 ProxyStartup
1254 NamesrvStartup
zhanghua@Breeze rocketmq-5.1.2 % 

说明服务已经启动成功 。

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

工具测试消息收发

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

1、 添加 NAMESRV_ADDR 环境变量 (macOS系统为例)

打开终端执行一下命令：

vim ~/.bash_profile
按 i 进入文件编辑模式 
export NAMESRV_ADDR=localhost:9876 （添加 NAMESRV_ADDR 环境变量）
esc 退出文件编辑模式
wq! 保存文件并退出  

执行编辑生效命令：

source ~/.profile

2、工具测试收发消息

1）、执行以下命令：

 zhanghua@Breeze rocketmq-5.1.2 % sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

执行结果：

zhanghua@Breeze rocketmq-5.1.2 % sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer
************
SendResult [sendStatus=SEND_OK, msgId=C0A81F5108D03D494FBF0C4818E40033, offsetMsgId=C0A81F5100002A9F00000000004A6EFD, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=0], queueOffset=5037]
SendResult [sendStatus=SEND_OK, msgId=C0A81F5108D03D494FBF0C4818E50034, offsetMsgId=C0A81F5100002A9F00000000004A6FED, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=1], queueOffset=5036]
SendResult [sendStatus=SEND_OK, 
************
zhanghua@Breeze rocketmq-5.1.2 %

消息发送成功后退出！

2）、执行以下命令：

zhanghua@Breeze rocketmq-5.1.2 % sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

执行结果：

zhanghua@Breeze rocketmq-5.1.2 % sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer
************
ConsumeMessageThread_please_rename_unique_group_name_4_19 Receive New Messages: [MessageExt [brokerName=broker-a, ***** ***** Message{topic='TopicTest', flag=0, properties={CONSUME_START_TIME=1688347339937, MSG_REGION=DefaultRegion, UNIQ_KEY=C0A81F5108D03D494FBF0C481CB703E5, CLUSTER=DefaultCluster, MIN_OFFSET=0,  ***** *****  body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 57, 55], transactionId='null'}]] 
ConsumeMessageThread_please_rename_unique_group_name_4_17 Receive New Messages: [MessageExt [brokerName=broker-a, ***** ***** Message{topic='TopicTest', flag=0, properties={CONSUME_START_TIME=1688347339937, MSG_REGION=DefaultRegion, UNIQ_KEY=C0A81F5108D03D494FBF0C481CB203DD, CLUSTER=DefaultCluster, MIN_OFFSET=0, ***** ***** body=[72, 101, 108, 108, 111, 32, 82, 111, 99, 107, 101, 116, 77, 81, 32, 57, 56, 57], transactionId='null'}]] 
************
Consumer Started.

消息消费成功后继续等待，接受下一次消息并消费！

SDK测试消息收发

工具测试完成后，我们可以尝试使用 SDK 收发消息。这里以 Java SDK 为例介绍一下消息收发过程，可以从 rocketmq-clients 中参阅更多细节。

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

2、在 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>

3、通过mqadmin创建 Topic。

$ sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster

执行结果：

zhanghua@Breeze rocketmq-5.1.2 % sh bin/mqadmin updatetopic -n localhost:9876 -t TestTopic -c DefaultCluster
create topic to 192.168.31.81:10911 success.
TopicConfig [topicName=TestTopic, readQueueNums=8, writeQueueNums=8, perm=RW-, topicFilterType=SINGLE_TAG, topicSysFlag=0, order=false, attributes={}]
zhanghua@Breeze rocketmq-5.1.2 % 

4、在已创建的Java工程中，创建发送普通消息程序并运行，示例代码如下：

import lombok.extern.slf4j.Slf4j;
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.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;
/**
 * @Author: zhanghua
 * @Date: 2023/6/30
 * @Version 1.0
 */
@Slf4j
public class ProducerExample {
    private static final Logger logger =         LoggerFactory.getLogger(ProducerExample.class);

    public static void main(String[] args) throws ClientException {
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        String endpoint = "localhost:8080";
        // 消息发送的目标Topic名称，需要提前创建。
        String topic = "TestTopic";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoint);
        ClientConfiguration configuration = builder.build();
        // 初始化Producer时需要设置通信配置以及预绑定的Topic。
        Producer producer = provider.newProducerBuilder()
            .setTopics(topic)
            .setClientConfiguration(configuration)
            .build();
        // 普通消息发送。
        Message message = provider.newMessageBuilder()
            .setTopic(topic)
            // 设置消息索引键，可根据关键字精确查找某条消息。
            .setKeys("messageKey")
            // 设置消息Tag，用于消费端根据指定Tag过滤消息。
            .setTag("messageTag")
            // 消息体。
            .setBody("messageBody".getBytes())
            .build();
        try {
            // 发送消息，需要关注发送结果，并捕获失败等异常。
            SendReceipt sendReceipt = producer.send(message);
            log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (ClientException e) {
            log.error("Failed to send message", e);
        }
        // producer.close();
    }

}

启动执行程序：

结果

Connected to the target VM, address: '127.0.0.1:59576', transport: 'socket'
09:41:38.707 [main] INFO org.example.ProducerExample - Send message successfully, messageId=01EECD4D9B9BC30AA804B3C15200000000

5、在已创建的Java工程中，创建订阅普通消息程序并运行。Apache RocketMQ 支持SimpleConsumer和PushConsumer两种消费者类型，您可以选择以下任意一种方式订阅消息。

import lombok.extern.slf4j.Slf4j;
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.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

/**
 * @Author: zhanghua
 * @Date: 2023/6/30
 * @Version 1.0
 */
@Slf4j
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();
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        String endpoints = "localhost:8080";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
            .setEndpoints(endpoints)
            .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // 为消费者指定所属的消费者分组，Group需要提前创建。
        String consumerGroup = "MessageQueue";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        String topic = "TestTopic";
        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
            .setClientConfiguration(clientConfiguration)
            // 设置消费者分组。
            .setConsumerGroup(consumerGroup)
            // 设置预绑定的订阅关系。
            .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
            // 设置消费监听器。
            .setMessageListener(messageView -> {
                log.info(" ==== == PushConsumer == ==== 消费到了");
                // 处理消息并返回消费结果。
                log.info("消费到的消息 messageId={}", messageView.getMessageId());
                log.info("消费到的消息内容 ---- {}", messageView);
                
                return ConsumeResult.SUCCESS;
            })
            .build();
        Thread.sleep(Long.MAX_VALUE);
        // 如果不需要再使用 PushConsumer，可关闭该实例。
        // pushConsumer.close();
    }
}

启动执行程序（消费消息）：

结果

Connected to the target VM, address: '127.0.0.1:50931', transport: 'socket'
08:43:43.208 [RocketmqMessageConsumption-0-17] INFO     org.example.PushConsumerExample -  ==== == PushConsumer == ==== 消费到了
08:43:43.211 [RocketmqMessageConsumption-0-17] INFO org.example.PushConsumerExample - 消费到的消息 messageId=01EECD4D9B9BC3062104B3B38100000000
08:43:43.212 [RocketmqMessageConsumption-0-17] INFO org.example.PushConsumerExample - 消费到的消息内容 ---- MessageViewImpl{messageId=01EECD4D9B9BC3062104B3B38100000000, topic=TestTopic, bornHost=Breeze, bornTimestamp=1688344961960, endpoints=ipv4:192.168.31.81:8081, deliveryAttempt=1, tag=messageTag, keys=[messageKey], messageGroup=null, deliveryTimestamp=null, properties={}}
Disconnected from the target VM, address: '127.0.0.1:50931', transport: 'socket'

Process finished with exit code 137 (interrupted by signal 9: SIGKILL)

关闭服务器

zhanghua@Breeze rocketmq-5.1.2 % sh bin/mqshutdown broker
zhanghua@Breeze rocketmq-5.1.2 % sh bin/mqshutdown namesrv

遇到问题

磁盘空间不足问题：

Exception in thread "main" org.apache.rocketmq.client.exception.MQClientException: Send [3] times, still failed, cost [781]ms, Topic: bingheTopic, BrokersSent: …………
Caused by: org.apache.rocketmq.client.exception.MQBrokerException: CODE: 14 DESC: service not available now. It may be caused by one of the following reasons:
the broker's disk is full [CL: 0.93 CQ: 0.93 INDEX: -1.00], messages are put to the slave, message store has been shut down, etc.

解决方法：
进入 rocketmq-all-5.1.2-source-release -- > distribution -- > target -- > rocketmq-5.1.2 -- > rocketmq-5.1.2 -- > bin 目录下

1、vim runbroker.sh

JVM配置参数修改为
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m"

2、vim runserver.sh

JVM配置参数修改为
JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn256m -XX:MetaspaceSize=128m -XX:MaxMetaspaceSize=320m"

至此RocketMQ 的安装与简单使用介绍完毕 ！

系统配置文件介绍

文章持续更新中、希望对各位有所帮助、有问题可留言 大家共同学习！