美文网首页
RocketMQ 安装与简单消息收发

RocketMQ 安装与简单消息收发

作者: 一滴矿泉水 | 来源:发表于2023-07-02 10:00 被阅读0次

    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 的安装与简单使用介绍完毕 ！

    系统配置文件介绍

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

    相关文章

      网友评论

          本文标题：RocketMQ 安装与简单消息收发

          本文链接：https://www.haomeiwen.com/subject/njktudtx.html

          延伸阅读

          深度阅读

          • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

          栏目导航

          热点阅读

          关于我们|服务条款|联系我们|RocketMQ 安装与简单消息收发|投稿指南|网站地图|RSS订阅|排版工具|手机版

          提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

          Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

          备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

          本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

          所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！