Linux部署RocketMQ并使用SpringBoot创建生产、消费者

r囧r小猫 2024-04-23 15:58 3阅读 0赞

>  *  😜**作** **者**：[是江迪呀][Link 1]
>  *  ✒️**本文关键词**：`RocketMQ`、`消息队列`
>  *  ☀️**每日 一言**：`在你心灰意冷、心烦意乱时也不要停下你的脚步！`

![在这里插入图片描述][ce036d3c8f85436c8795dfc96aceec00.png_pic_center]

## 一、前言 ##

`RocketMQ（Apache RocketMQ）`是一种开源的分布式消息中间件系统，最初由阿里巴巴开发并捐赠给 **Apache 基金会**。它提供了可靠的、低延迟的消息传递能力，适用于构建大规模分布式系统中的消息通信。RocketMQ 主要用于解决分布式系统中异步通信、解耦、流量削峰等问题。下面让我们一起看下，如何在`Linux`上部署`RocketMQ`~

## 二、介绍`RocketMQ` ##

### 2.1 `RocketMQ`产生背景 ###

随着业务规模的扩大，阿里巴巴面临着越来越多的分布式系统构建需求。为了解决这个问题，**阿里巴巴集团**于**2012**年推出的开源分布式消息中间件 —— `RocketMQ`。

### 2.1 `RocketMQ`作用 ###

**（1）异步通信和解耦：** `RocketMQ`可以在不同的服务之间实现异步通信，解耦了服务之间的紧耦合关系，提高了系统的可维护性和可扩展性。  
**（2）流量削峰：** `RocketMQ`支持消息积压和消费速率不匹配时的流量削峰功能，防止系统因突发流量而崩溃。  
**（3）实时数据同步：** 用于将数据实时同步到不同的存储介质，保持数据的一致性。  
**（4）事件驱动架构：** `RocketMQ`支持**事件驱动**的架构，使得系统能够更加敏捷地响应业务事件。

### 2.2 `RocketMQ`的组件 ###

`RocketMQ` 的主要组件包括：  
（1）`Producer：`负责发送消息到 `RocketMQ` 服务器。  
（2）`Broker：`消息中转服务器，负责存储消息并提供消息的读写服务。  
（3）`Consumer：`从 `Broker` 订阅并消费消息。  
（4）`Topic：`消息的分类，`Producer` 发送消息到特定的 `Topic`，`Consumer` 订阅相应的 `Topic`。  
（5）`Tag：` 对消息的进一步分类，可以用于 `Consumer`进一步过滤消息。  
（6）`Message Queue：` 每个 `Topic`下可以分成多个 `Message Queue`，实现消息的分区和负载均衡。

### 2.3 `RocketMQ`的优缺点 ###

#### （1）优点 ####

*  **高吞吐量：** `RocketMQ`具有高吞吐量的特点，适用于大量消息的处理。
 *  **可靠性：** `RocketMQ`通过消息的持久化存储和复制机制，确保消息不会丢失。
 *  **低延迟：** `RocketMQ`在消息传递过程中能够保持较低的延迟，适用于实时性要求较高的场景。
 *  **灵活的消息模式：** 支持**发布-订阅**和点对点两种消息模式，根据业务需求进行选择。
 *  **水平扩展：** 可以通过增加`Broker`节点来实现水平扩展，提高消息处理能力。

#### （2）缺点 ####

*  **维护成本：** `RocketMQ`需要维护多个组件，包括`Producer`、`Broker`和`Consumer`等，涉及到一定的运维成本。
 *  **学习曲线：** 对于新手来说，学习和理解`RocketMQ`的一些概念和配置可能需要一定的时间。
 *  **一致性保障：** 虽然`RocketMQ`通过复制机制保障了消息的可靠性，但在极端情况下可能会存在消息的重复传递或乱序问题。

## 三、 `RocketMQ`如何部署 ##

### 3.1 下载 ###

[RocketMQ下载地址][RocketMQ]

### 3.2 上传、解压 ###

上传文件到`Linux`有两种方式：

#### （1）上传 ####

*  通过`rz`命令

你可以使用`rz`命令，在使用这个命令之前你必须确保linux已经安装了lrzsz，安装命令如下:

sudo apt-get update
    sudo apt-get install lrzsz

*  使用`xftp`  
    这个我就不赘述了。  
    ![在这里插入图片描述][774501d301ef4f88a0e99d46a447b6c8.png]

#### （2）解压 ####

unzip rocketmq-all-4.5.2-bin-release.zip

如果没有安装unzip，需要安装一下：

// 查看 unzip 包的安装情况
    yum list unzip
    //没有安装时，使用命令安装 unzip
    yum list unzipyum install unzip.x86_64

![在这里插入图片描述][4fe8a49e7d7c42c9b94573d7596357d6.png]

### 3.2 启动`RocketMQ` ###

RocketMQ的启动主要涉及到`Namesrv（命名服务）`和`Broker（消息存储和消费者服务）`两部分。要想启动`RocketMQ`，首先进入解压后的`bin`目录：

cd rocketmq-all-4.5.2-bin-release/bin

#### （1）启动`Namesrv`并设置输出日志位置 ####

nohup sh mqnamesrv > namesrv.log 2>&1 &

#### （2）启动`Broker`并设置输出日志位置 ####

nohup sh mqbroker -n localhost:9876 > broker.log 2>&1 &

查看是否启动：

jps

输出下面的内容说明启动成功了：

2931 NamesrvStartup
    25599 Jps
    25583 BrokerStartup

在启动`Broker`会出现失败问题，一般来说就是**内存不足** ，`RocketMq`默认的虚拟机内存较大，因而启动失败，需要编辑如下两个配置文件，修改jvm的内存大小：

//编辑runbroker.sh和runserver.sh修改默认的JVM大小
    vim runbroker.sh
    vim runserver.sh

![在这里插入图片描述][9d0b0bf043fc4def943536679cc5fe12.png]  
修改为：

JAVA_OPT="${JAVA_OPT} -server -Xms256m -Xmx256m -Xmn128m -xx:metaspaceSize=128m -XX:MaxMetaspaceSize=320m"

如果还是启动不了，需要将NameServer关闭，重新启动一下，同样是先进入bin目录，关闭命令如下：

sh mqshutdown namesrv

## 四、测试与关闭 ##

### 4.1 测试 ###

#### （1） 发送消息（生产者） ####

//设置环境变量
    export NAMESRV_ADDR=localhost:9876
    //使用安装包的demo发送消息
    sh bin/tools.sh org.apache.rocketmq.example.quickstart.Producer

![在这里插入图片描述][52daa813eb4a41c6b8f5377c3308237c.png]  
上面的信息就是`RocketMQ`的`producer`发送的消息。特点：启动发送完毕消息后就会停止。

#### （2） 接收消息（消费者） ####

//设置环境变量
    export NAMESRV_ADDR=localhost:9876
    //接收消息
    sh bin/tools.sh org.apache.rocketmq.example.quickstart.Consumer

![在这里插入图片描述][c17c223c1cf64cac98e3c384a82efc76.png]

### 4.2 关闭`RocketMQ` ###

//关闭namesrv
    sh bin/mqshutdown namesrv
    //关闭Broker
    sh bin/mqshutdown broker

## 五、`SpringBoot`连接`RocketMQ` ##

### 5.1 引入依赖 ###

<dependencies>
     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-thymeleaf</artifactId>
         <version>2.3.5.RELEASE</version>
     </dependency>
        
     <dependency>
         <groupId>org.apache.rocketmq</groupId>
         <artifactId>rocketmq-client</artifactId>
         <version>4.5.2</version> 
     </dependency>
    </dependencies>

### 5.2 配置文件`application.properties` ###

# Name Server地址
    rocketmq.name-server=your-nameserver-ip:9876
    # 生产者组名
    rocketmq.producer.group=my-producer-group
    # 消费者组名
    rocketmq.consumer.group=my-consumer-group

### 5.3 生产者 ###

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RocketMQProducer {
        
    
        @Value("${rocketmq.name-server}")
        private String nameServer;
    
        @Value("${rocketmq.producer.group}")
        private String producerGroup;
    
        public void sendMessage(String topic, String message) throws Exception {
        
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            producer.setNamesrvAddr(nameServer);
            producer.start();
            // 创建消息对象，设置消息内容
            org.apache.rocketmq.common.message.Message msg = new org.apache.rocketmq.common.message.Message(topic, message.getBytes());
            // 发送消息
            producer.send(msg);
            producer.shutdown();
        }
    }

### 5.4 消费者 ###

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.stereotype.Component;
    
    @Component
    public class RocketMQConsumer {
        
    
        @Value("${rocketmq.name-server}")
        private String nameServer;
    
        @Value("${rocketmq.consumer.group}")
        private String consumerGroup;
        public void startConsumer(String topic) throws MQClientException {
        
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.setNamesrvAddr(nameServer);
            // 订阅主题和标签，可以根据需要进行过滤
            consumer.subscribe(topic, "*");
            // 注册消息监听器
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
        
                for (org.apache.rocketmq.common.message.MessageExt msg : msgs) {
        
                    System.out.println("Received message: " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
    
            consumer.start();
        }

### 5.5 启动类 ###

import org.springframework.boot.SpringApplication;
    import org.springframework.boot.autoconfigure.SpringBootApplication;
    import org.springframework.context.ConfigurableApplicationContext;
    
    @SpringBootApplication
    public class RocketMQDemoApplication {
        
    
        public static void main(String[] args) throws Exception {
        
            ConfigurableApplicationContext context = SpringApplication.run(RocketMQDemoApplication.class, args);
    
            RocketMQProducer producer = context.getBean(RocketMQProducer.class);
            producer.sendMessage("my-topic", "Hello, RocketMQ!");
            RocketMQConsumer consumer = context.getBean(RocketMQConsumer.class);
            consumer.startConsumer("my-topic");
        }
    }

## 六、`RocketMQ`集群 ##

上面所述的是单体`RocketMQ`，也能使用。但是如果你想要实现**高可用**在实际的业务场景中。`RocketMQ`大部分都不会**单体**存在，需要搭建集群来实现**高可用**。

有人已经写好了，而且很详细：[传送门][Link 2]

[Link 1]: https://blog.csdn.net/qq_42785250?spm=1011.2124.3001.5343
[ce036d3c8f85436c8795dfc96aceec00.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/4bbac2d5dd2044e4b5121560bb4914b9.png
[RocketMQ]: https://rocketmq.apache.org/zh/download/
[774501d301ef4f88a0e99d46a447b6c8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/5911ac08a3ff4e63a8a02b5a2ddfc6f6.png
[4fe8a49e7d7c42c9b94573d7596357d6.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/f61e04591716482c935873d33ec0a386.png
[9d0b0bf043fc4def943536679cc5fe12.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/a47efe81ab8744e08b2f8306969479ed.png
[52daa813eb4a41c6b8f5377c3308237c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/d3f249423b1f43ea893afec083f6ede4.png
[c17c223c1cf64cac98e3c384a82efc76.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/23/cb5c71a569504e538257c876a34a2eb4.png
[Link 2]: https://www.cnblogs.com/ifme/p/13079735.html