当前位置：首页 > article >正文

SpringBoot集成-RocketMQ快速入门

article 2024/10/1 12:31:40

1.MQ概述

MQ全称为Message Queue，即消息队列，是一种提供消息队列服务的中间件，也称为消息中间件，是一套提供了消息生产、存储、消费全过程的软件系统，遵循FIFO原则。

1.1MQ常见产品

ActiveMQ
ActiveMQ是使用Java语言开发一款MQ产品。早期很多公司与项目中都在使用。但现在的社区活跃度已经很低。现在的项目中已经很少使用了。
RabbitMQ
RabbitMQ是使用ErLang语言开发的一款MQ产品。其吞吐量较Kafka与RocketMQ要低，且由于其不是 Java语言开发，所以公司内部对其实现定制化开发难度较大。
Kafka
Kafka是使用Scala/Java语言开发的一款MQ产品。其最大的特点就是高吞吐率，常用于大数据领域的实时计算、日志采集等场景。其没有遵循任何常见的MQ协议，而是使用自研协议。对于Spring Cloud Netflix，其仅支持RabbitMQ与Kafka。
RocketMQ
RocketMQ是使用Java语言开发的一款MQ产品。经过数年阿里双11的考验，性能与稳定性非常高。其没有遵循任何常见的MQ协议，而是使用自研协议。对于Spring Cloud Alibaba，其支持RabbitMQ、 Kafka，但提倡使用RocketMQ

1.2MQ作用

限流削峰
MQ可以将系统的超量请求暂存其中，以便系统后期可以慢慢进行处理，从而避免了请求的丢失或系统被压垮。
异步&解耦
上游系统对下游系统的调用若为同步调用，则会大大降低系统的吞吐量与并发度，且系统耦合度太高。而异步调用则会解决这些问题。所以两层之间若要实现由同步到异步的转化，一般性做法就是，在这两层间添加一个MQ层。即使消费者挂掉也不影响生产者工作，只要把消息放入队列即可，消费者重启后自己消费即可。
数据收集
分布式系统会产生海量级数据流，如：业务日志、监控数据、用户行为等。针对这些数据流进行实时或批量采集汇总，然后对这些数据流进行大数据分析，这是当前互联网平台的必备技术。通过MQ完成此类数据收集是最好的选择。
大数据处理
比如我们的平台向“三方平台”获取数据，一次请求了大量数据回来要进行处理，由于数据较多处理不过来，那么就可以放入MQ，再创建一些消费者进行数据处理即可。

2.RocketMQ

2.1概述

RocketMQ是一个统一消息引擎、一种提供消息队列服务的中间件，轻量级数据处理平台。 RocketMQ是⼀款阿⾥巴巴开源的消息中间件，阿⾥巴巴向 Apache 软件基⾦会捐赠 RocketMQ

2.2RocketMQ安装

2.2.1下载

下载地址：http://rocketmq.apache.org/release_notes/release-notes-4.2.0/

下载后解压

Bin : 可执行文件目录
Conif：配置文件目录
Lib : 依赖库，一堆Jar包

2.2.2配置环境变量

2.3启动RocketMQ

2.3.1启动NameServer(注册中心)

Cmd命令框执行进入至MQ文件夹\bin下，然后执行 start mqnamesrv.cmd，启动NameServer。

成功后会弹出提示框，此框勿关闭。

2.3.2启动Broker

CMD执行start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true ，启动Broker。

成功后会弹出提示框，此框勿关闭

2.4RocketMQ本地存储结构

RabbitMQ安装好之后会在用户目录下产生一个store目录用来存储相关数据：

Commitlog : 消息是存储,在commitlog目录中，以mapperdFile文件顺序存储消息。
Config : 存放运行期间的配置文件
Consumerqueue : 该目录中存放的是队列，consume queue存放着commitlog中的消息的索引位置
Index :存放着消息索引文件 indexFile，用来实现根据key进行消息的快速查询
Abort : 该文件在broker启动后自动创建，正常关闭abort会消失
Checkpoint ：记录 Commitlog ，Consumerqueue 和index 文件的最后刷盘时间戳

[问]RocketMQ数据存储在磁盘会影响性能吗？

不会，RocketMQ的性能在所有的MQ中是比较高的，主要是因为RocketMQ使用了mmap零拷贝技术，consumequeue中的数据是顺序存放的，还引入了PageCache的预读取机制，使得对 consumequeue文件的读取几乎接近于内存读取，即使在有消息堆积情况下也不会影响性能。

2.5安装可视化界面

2.5.1下载

RocketMQ可视化管理插件下载地址：https://github.com/apache/rocketmq-externals/releases

2.5.2修改配置

解压后，修改配置：src/main/resource/application.properties ,这里需要指向Name Server 的地址和端口如下：

2.5.3打包插件

回到安装目录(pom.xml所在目录),执行: mvn clean package -Dmaven.test.skip=true ，然后会在target目录生成打包后的jar文件

2.5.4启动插件

进入 target 目录，执行 java -jar rocketmq-console-ng-1.0.0.jar ，访问 http://localhost:8080

3.RocketMQ原理

3.1RocketMQ组成

RocketMQ的集群架构如下

3.1.1Producer

消息发布的角色，支持分布式集群方式部署。Producer启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，并获取当前Broker的ip和端口号等信息，轮负载均衡算法从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。

3.1.2Consumer

Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息

3.1.3NameServer

NameServer是一个Broker与Topic路由的注册中心支持Broker的动态注册与发现，NameServer等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。

Broker管理
NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活。
路由信息管理
每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费

3.1.4Broker

Broker主要负责消息的存储、投递和查询以及服务高可用保证，每个Broker集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。

Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。所以这个Broker集群是主备集群。Consumer既可以从Master订阅消息，也可以从Slave订阅消息。一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。 Master与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为0表示Master非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。

3.1.5Topic主题

一个Broker中有多个Topic，每条message消息只能属于一个topic主题，Topic用来区分不同的消息，消费者根据Topic来选择需要消费的消息，Topic中包含queue队列（默认四个），生产者发送消息时根据负载均衡算法选择其中一个队列发送

3.1.6Tag标签

Tag是附加在Topic上的一个二级分类标签，可以理解为Topic内部的一个子类别或子主题。
作用：同一个Topic下的消息可以通过不同的Tag进一步细分，以便于更细致地管理和消费消息。每个消息在发布时都可以携带一个或多个Tag。

Topic与Tag的关系：

一对多：一个Topic可以有多个Tag，也就是说，在一个大的主题下，可以根据业务需求划分出多个子类别的消息。

3.1.7MessageQueue队列

一个Topic中可以包含多个Queue（默认四个），一个Topic的Queue也被称为一个Topic中消息的分区（Partition）。在一个Consumer Group内，一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。这样的分配规则，每个Queue只有一个消费者，可以避免消费过程中的多线程处理和资源锁定，有效提高各Consumer消费的并行度和处理效率。

【注意】一个Topic可以对应多个消费者，一个Queue只能对应一个组中的一个消费者。

【注意】为了防止消息紊乱，一个Consumer Group 中的Consumer都是订阅相同Topic下的Queue。

3.2RocketMQ工作原理

3.2.1工作流程

启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，负载均衡算法从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息

3.2.2Producer 生产者

3.2.2.1消息发送方式

同步发送、异步发送、顺序发送、单向发送

3.2.2.2生产者组（解决单点故障）

生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。

3.2.2.3消息选择队列算法

轮询算法：挨个选择分配

消息发送延迟最低算法：根据消息发送到每个队列中的时间，按照延迟最低选择

3.2.3Consumer 消费者

3.2.3.1消费者组（解决单点故障）

消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息，不同的 Consumer Group可以消费同一个Topic。一个Consumer Group内的Consumer可以消费多个Topic的消息。

3.2.3.2消费者消息拉取模式

消息的消费分为：拉取式 pull（消费者主动拉取） ,和推送是 push（NameServer主动发送给消费者）

Pull:拉取式，需要消费者间隔一定时间就去遍历关联的Queue,实时性差但是便于应用控制消息的拉取
Push:默认推送式，封装了Queue的遍历，实时性强，但是对系统资源占用比较多。

3.2.3.3Queue的分配算法

Queue是如何分配给Consumer的，这对应了四种算法：平均分配策略，环形平均策略，一致性Hash策略，同机房策略。

平均分配（轮询）【默认】：根据 qeueuCount / consumerCount 作为每个消费者平均分配数量，如果多出来的queue就再依次逐个分配给Consumer（顺序不固定）。
环形平均策略：根据消费者的顺序，顺时针一个一个的分配Queue即可类似于发扑克牌，完全平均（顺序也是固定）。
一致性Hash策略：该算法将Consumer的Hash值作为节点放到Hash环上，然后将Queue的hash值也放入Hash环上，通过顺时针进行就近分配。
同机房策略：该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配。

平均分配性能比较高，一致性Hash性能不高，但是能减少Rebalance，如果Consumer数量变动频繁可以使用一致性Hash。

3.2.3.4消息消费模式

消息的消费模式有广播模式和集群模式

广播模式:同一个Consumer Group 下的所有Consumer都会受到同一个Topic的所有消息。同一个消息可能会被消费多次。
集群模式(默认):同一个Gonsumer Group 下的Consumer平分同一个Topic下的消息。同一个消息只是被消费一次。

3.2.3.5Rebalance重新负载

当消费者数量或者Queue的数量修改，Rebalance是把⼀个Topic下的多个Queue重新分配给Consumer Group下的Consumer。目的是增加消费能力。

由于一个队列只分配给一个Consumer,那么当Consumer Group中的消费者数量大于队列数量，那么多出来的Consumer分配不到队列。

3.2.3.6Offset标记

3.2.3.6.1概述

RockertMQ通过Offset来维护Consumer的消费进度，比如：消费者从哪个位置开始持续消费消息的？这里有三个枚举来指定从什么位置消费

CONSUME_FROM_LAST_OFFSET：从queue的最后一条消息开始消费
CONSUME_FROM_FIRST_OFFSET：从queue的第一条消息开始消费
CONSUME_FROM_TIMESTAMP：从某个时间戳位置的消息开始消费。

3.2.3.6.2Offset原理

Broker将消息队列分为一个环形，每次消费都从Offset开始，消费者消费结束之后，会向Consumer会提交其消费进度offset给Broker。Offset信息的存储分为本地 Offset管理和远程Offset管理

远程Offset管理:Brocker通过 store/config/consumerOffset.json 文件以JSON方式来存储offset相关数据以json的形式：适用于集群模式
本地Offset管理：offset相关数据以json的形式持久化到Consumer本地磁盘文件中，路径为当前用户主目录下的.rocketmq_offsets/{group}/Offsets.json ：适用于广播模式

3.2.3.6.3Offset提交方式

Offset的同步提交与异步提交: 集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种：

同步提交：消费者在消费完一批消息后会向broker提交这些消息的offset，等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取 nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。
异步提交：消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset 后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取 nextBeginOffset。

3.2.4消息的清理

消息不会被单独清理，消息是顺序存储到commitlog的，消息是以commitlog为单位进行清理，RocketMQ有自己的清理规则，默认是72小时候后进行清理

到达时间清理点，自动清理过期的文件(凌晨4点)
磁盘空间使用率达到了过期清理阈值(75%),自动清理过期的文件。
磁盘占用率达到清理阈值(85%)，开始按照设定的规则清理文件，从老的文件开始。
磁盘占用率达到系统危险阈值(90%),拒绝写入数据。

4.SpringBoot集成RocketMq

4.1导包

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-spring-boot-starter</artifactId>
  <!-- <version>2.0.4</version> -->
  <version>2.2.1</version>
</dependency>

4.2配置yml文件

rocketmq:
  name-server: 127.0.0.1:9876
  #生产者配置
  producer:
    #生产者组名字
    group: "service-producer"
    # 消息最大长度 默认 1024 * 1024 * 4 (4M)
    max-message-size: 4194304
    # 发送消息超时时间，默认 3000
    send-message-timeout: 3000
    # 发送消息失败重试次数，默认2
    retry-times-when-send-failed: 2
    # 异步消息发送失败重试次数
    retry-times-when-send-async-failed: 2
    #达到 4096 ，进行消息压缩
    compress-message-body-threshold: 4096
  consumer:
    #消费者名字
    group: "service-consumer"
    #批量拉取消息数量
    pull-batch-size: 10
    message-model: CLUSTERING
    selector-expression: "*"

4.3创建生产者

4.3.1注入rocketMQTemplate

@Autowired
private RocketMQTemplate rocketMQTemplate;

4.3.2发送消息

4.3.2.1发送同步消息

同步消息是发送者发送消息，需要等待结果的返回，才能继续发送第二条消息，这是一种阻塞式模型，虽然消息可靠性高，但是阻塞导致性能低。

/**
     * 同步消息
     *
     * @throws Exception 异常
     */
    @Test
    public void syncMessage() throws Exception {
        Message<String> message = MessageBuilder.withPayload("你好呀！哈哈123").build();
        SendResult sendResult = rocketMQTemplate.syncSend("cctv-topic:cctv6-tag", message);
        if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            System.err.println("发送失败");
        }
    }

4.3.2.2发送异步消息

异步消息是发送者发送消息，无需等待发送结果就可以再发送第二条消息，它是通过回调的方式来获取到消息的发送结果，消息可靠性高，性能也高。

/**
     * 异步消息
     *
     * @throws Exception 异常
     */
    @Test
    public void asyncMessage() throws Exception {
        Message<String> message = MessageBuilder.withPayload("你好呀！我是异步消息").build();
        rocketMQTemplate.asyncSend("cctv-topic:cctv6-tag", message, new SendCallback() {
            /**
             * 成功
             *
             * @param sendResult 发送结果
             */
            @Override
            public void onSuccess(SendResult sendResult) {
                if (!SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                    System.err.println("发送失败");
                }
                System.out.println(sendResult);
            }

            /**
             * 在异常
             *
             * @param throwable throwable
             */
            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送异常");
                throwable.printStackTrace();
            }
        });
        Thread.sleep(5000);
    }

4.3.2.3发送单向消息

这种方式指的是发送者发送消息后无需等待Broker的结果返回，Broker也不会返回结果，该方式性能最高，但是消息可靠性低。

 /**
     * 单向信息
     */
    @Test
    public void oneWayMessage() {
        Message<String> message = MessageBuilder.withPayload("你好呀！我是单向消息").build();
        rocketMQTemplate.sendOneWay("cctv-topic:cctv6-tag", message);
    }

4.3.2.4发送延迟消息

延迟消息即：把消息写到Broker后需要延迟一定时间才能被消费，在RocketMQ中消息的延迟时间不能任意指定，而是由特定的等级(1 到 18)来指定，分别有：

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

可以通过修改配置来增加级别，比如在mq安装目录的 broker.conf 文件中增加

messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 2d 这个时候总共就有19个level。

 /**
     * 延迟消息
     */
    @Test
    public void deferredMessage() {
        Message<String> message = MessageBuilder.withPayload("你好呀！我是延迟消息").build();
        //发送延迟消息，单位毫秒,2秒超时，延迟30秒
        SendResult sendResult = rocketMQTemplate.syncSend("cctv-topic:cctv6-tag", message, 2000, 4);
        if (!SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
            System.err.println("发送失败");
        }
    }

4.3.2.5发送顺序消息

4.3.2.5.1发送全局顺序消息

全局有序是一个topic下的所有消息都要保证顺序，如果要保证消息全局顺序消费，就需要保证使用一个队列存放消息，一个消费者从这一个队列消费消息就能保证顺序，即：单线程执行。

/**
     * 全局顺序信息
     */
    @Test
    public void sequentialMessage() {
        for (int i = 0; i < 5; i++) {
            Message<String> message = MessageBuilder.withPayload("你好呀！我是顺序消息,我是第---" + i + "---个").build();
            //发送延迟消息，单位毫秒,2秒超时，延迟30秒
            SendResult sendResult = rocketMQTemplate.syncSend("order-topic:order-tag", message);
            if (!SendStatus.SEND_OK.equals(sendResult.getSendStatus())) {
                System.err.println("发送失败");
            }
        }
    }

4.3.2.5.2发送局部顺序消息

还有一种就是分区有序或者部分有序，部分顺序消息只要保证某一组消息被顺序消费，即：只需要保证一个队列中的消息有序消费即可。

hashKey： hashKey 是一个字符串类型的参数，它用于计算消息应该被路由到哪个Message Queue。RocketMQ使用哈希算法对hashKey进行计算，将同一hashKey对应的消息都路由到同一个Message Queue上，这样可以确保具有相同hashKey的所有消息按发送顺序依次消费。

  /**
     * 部分有序消息
     */
    @Test
    public void partiallyOrderedMessage() {
        for (int i = 1; i < 5; i++) {
            for (int j = 1; j < 5; j++) {
                Message<String> message = MessageBuilder.withPayload("你好呀！我是局部有序消息" + i + "组中的第" + j + "条").build();
                SendResult sendResult = rocketMQTemplate.syncSendOrderly("partially-topic:partially-tag", message, String.valueOf(i));
                if (!sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
                    System.err.println("发送失败");
                }
            }
        }
    }

4.3.2.6事务消息

4.3.2.7死信队列

消息多次消费失败，达到最大重试次数，消息不会被丢弃而是进入死信队列(Dead-Letter Queue，DLQ),死信队列中的消息被称为死信消息（Dead-Letter Message，DLM）。

死信队列具有如下特征

死信队列中的消息无法再消费，死信队列对应Topic的权限为2，只有写权限，所以死信队列没有办法读取。
3天之后死信队列分钟的消息被删除，和普通消息一样
死信队列就是一个特殊的Topic，名称为%DLQ%consumerGroup@consumerGroup，其中每个队列都是死信队列
如果⼀个消费者组未产生死信消息，则不会为其创建相应的死信队列

如果出现死信队列，说明程序除了问题，程序员应该及时的排除，进行BUG的处理。我们应该在消费者重试次数达到一定程度就对消息进行持久化，方便后续的处理。或额外定时重试。

4.4创建消费者

4.4.1普通消费者

@Component
@RocketMQMessageListener(topic = "cctv-topic",
        consumerGroup = "cctv-consumer",
        selectorExpression = "cctv6-tag")
public class CctvConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        byte[] body = message.getBody();
        System.out.println(new String(body, StandardCharsets.UTF_8));
    }
}

RocketMQListener ： MQ提供了的消费者监听器，MessageExt是消息对象Message的子类。这里的泛型对应生产者的消息类型，可以直接是消息的对象类型。
@RocketMQMessageListener：消息监听的注解，提供了常用的四个属性
consumerGroup ：消费者的组名
topic : 主题，对应生产者发送消息指定的destination拼接的主题
selectorExpression ：消息选择表达式，其实就是制定消费什么tags ; 可以是固定一个值，如果是 * 是消费该topic下的所有消息；或者可以使用： tag1 | tag2 的方式消费多个消息，除此之外还支持使用SQL进行消息过滤，这种方式可以实现对消息的复杂过滤。SQL过滤表达式中支持多种常量类型与运算符。比如：and ; or ; not ; IS NULL 或者 IS NOT NULL 等等。
messageModel ：消息的消费模式，默认是CLUSTERING集群，还支持BROADCASTING广播

4.4.2顺序消息消费者

/**
 * 部分有序消费
 *
 * @author 
 * @date 2024/01/21
 */
@Component
@RocketMQMessageListener(topic = "partially-topic",
        consumerGroup = "partially-consumer",
        selectorExpression = "partially-tag",
        consumeMode = ConsumeMode.ORDERLY
)
public class PartiallyConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt message) {
        byte[] body = message.getBody();
        System.out.println(new String(body, StandardCharsets.UTF_8));
    }
}

查看全文

http://www.kler.cn/news/327443.html