RocketMQ概论

2023-11-18

目录

前言:

1.概述

2.下载安装、集群搭建

3.消息模型

4.如何保证吞吐量

4.1.消息存储

4.1.1顺序读写

4.1.2.异步刷盘

4.1.3.零拷贝

4.2.网络传输


前言:

RocketMQ的代码示例在安装目录下有全套详细demo,所以本文不侧重于讲API这种死的东西,而是侧重于讲解RocketMQ的特性。消息中间件无非需要关注三点:吞吐量、消息可靠性、消息模型。任何消息中间件的异同点就体现在这三个方面。本文也会从这三个方面来讲解RocketMQ。

本文是作者RocketMQ系列文章的最后一篇,一些基础的概念、下载安装、集群、基本消息模型等,前面都已经有单独的文章讲过,每篇都很深入浅出,单篇阅读花不了两三分钟,所以这些内容会直接贴上链接。

1.概述

RocketMQ 是一款开源的分布式消息中间件,最初由阿里巴巴集团开发并开源。它旨在为分布式系统提供可靠、高性能、可扩展的消息通信能力。RocketMQ和RabbitMQ、KAFKA一起并列为现在主流的三大消息中间件。

RocketMQ的基本概念+架构:

RocketMQ基础概念__BugMan的博客-CSDN博客

2.下载安装、集群搭建

RocketMQ的下载安装教程+集群搭建教程:

RocketMQ下载安装、集群搭建保姆级教程_安装rocketmq集群__BugMan的博客-CSDN博客

3.消息模型

RocketMQ的消息模型有以下几种:

  • 顺序消息,消费者按照生产者的发送顺序进行消费。
  • 广播消息,一条消息被多个消费者消费。
  • 延迟消息,驻留时间,消费者才能消费到消息。
  • 批量消息,支持生产者批量发送消息。
  • 过滤消息,通过tag,消费者可以消费同一个topic下自己感兴趣的消息。
  • 事务消息,生产者的消息生产支持事务回退。

RocketMQ消息模型详解:

详解RocketMQ使用__BugMan的博客-CSDN博客

4.如何保证吞吐量

4.1.消息存储

RocketMQ最大的特点概括起来其实就一句话:

既保证了消息的可靠性,又保证了吞吐量。

可靠性和吞吐量其实是互斥的两点,为了保证可靠性,消息就一定要落在磁盘存储防止断电丢失。落在磁盘存储后,读这条消息的时候的磁盘IO就会拉低吞吐量。所以RocketMQ的核心其实就是数据落磁盘,然后想尽一切办法来提高吞吐量。

RocketMQ的生产消费过程:

  • 收到生产者发送的消息,将其存起来

  • 向消费者推送一条消息后,等待消费者的ACK,收到ACK后将消息标记为已消费。

  • 定期删除一些已经过期的消息。

RocketMQ用来提高吞吐量的手段:

  • 顺序写

  • 异步刷盘

  • 零拷贝

4.1.1顺序读写

磁盘的顺序读写性能要远好于随机读写。因为每次从磁盘读数据时需要先寻址,找到数据在磁盘上的物理位置。对于机械硬盘来说,就是移动磁头,会消耗时间。 顺序读写相比于随机读写省去了大部分的寻址时间,它只需要寻址一次就可以连续读写下去,所以性能比随机读写好很多。

RocketMQ 利用了这个特性。它所有的消息数据都存放在一个无限增长的文件队列 CommitLog 中,CommitLog 是由一组 1G 内存映射文件队列组成的。 写入时就从一个固定位置一直写下去,一个文件写满了就开启一个新文件顺序读写下去。

RocketMQ顺序写盘的过程如下:

  1. 生产者发送消息到 RocketMQ Broker。Broker 接收到消息后,将消息按照顺序写入内存中的 Page Cache(页面缓存)中。

  2. 消息在内存的 Page Cache 中形成连续的数据块。由于 RocketMQ 的顺序写盘策略,相同 Topic 和 Queue ID 的消息会按照发送的先后顺序排列成一批,形成连续的数据块,而不是随机地散落在磁盘上。

  3. 然后,后台线程会定期将 Page Cache 中的连续数据块顺序写入磁盘上的存储文件中,称为 "CommitLog"。由于数据是连续写入,因此磁盘的写入操作变得高效,减少了寻道时间和磁盘的碎片化,提高了写入性能。

需要注意的是,顺序写盘并不是说所有消息都按照先后顺序写入,而是相同 Topic 和 Queue ID 的消息会按照发送的先后顺序形成连续的数据块写入磁盘。对于不同 Topic 和 Queue ID 的消息,它们可能在磁盘上是交错存储的。

4.1.2.异步刷盘

RocketMQ 的异步刷盘(Async Flush)是一种优化手段,用于提高消息的写入性能和吞吐量。在消息存储上,RocketMQ 遵循了写入先落盘再返回的策略,即在消息写入磁盘之前,会先将消息写入操作系统的 Page Cache(页面缓存)中,并立即返回写入成功的响应给生产者,然后由后台线程异步地将 Page Cache 中的数据刷写到磁盘中。

4.1.3.零拷贝

零拷贝(Zero-Copy)是一种优化技术,旨在提高数据传输的效率和性能,特别是在文件传输和网络数据传输中。传统的数据传输方式涉及多次数据拷贝,而零拷贝通过避免不必要的数据拷贝操作,减少了数据传输的开销,从而提高系统的性能。

在传统的数据传输中,例如从磁盘读取文件并通过网络发送,通常涉及以下步骤:

  1. 将数据从磁盘读取到内核空间(Kernel Buffer)。
  2. 将数据从内核空间拷贝到用户空间(User Buffer)。
  3. 将数据从用户空间拷贝到网络缓冲区(Network Buffer)。
  4. 最终数据通过网络发送。

这种传统的数据传输方式涉及多次数据拷贝,每次拷贝都需要 CPU 参与,并且需要在内核空间和用户空间之间进行数据复制,导致了额外的开销和延迟。

零拷贝技术的主要思想是避免不必要的数据拷贝,通过直接在内核空间和用户空间之间传输数据,从而减少 CPU 和内存的使用。

关于0拷贝更详细的内容异步博主的另一篇文章:

全网最清晰的零拷贝详解,看一遍就会__BugMan的博客-CSDN博客

4.2.网络传输

关于序列化,以及为什么要用序列化,不是很清楚的同学可以移步博主的另一篇文章:

详解JAVA序列化__BugMan的博客-CSDN博客

RocketMQ的报文使用了序列化,序列化和反序列化由生产者端和消费者端的SDK来实现。

RocketMQ目前支持的序列化方式有:

  1. RocketMQ 自定义序列化:RocketMQ 使用自定义的序列化协议进行消息的编码和解码。这个序列化协议被称为 Remoting Command Protocol。该协议采用了二进制格式,对消息进行高效的编解码处理,以实现高性能和低延迟。

  2. JSON 序列化:RocketMQ 支持将消息转换为 JSON 格式进行传输。JSON 序列化方式适合处理复杂结构的消息,易于阅读和调试,但相比二进制格式会增加一定的传输开销。

  3. Java 原生序列化:RocketMQ 还支持使用 Java 原生的序列化方式(Java Serialization)。Java 原生序列化是 Java 标准库提供的一种序列化方式,但在性能方面可能不如其他序列化框架。

  4. Hessian 序列化:Hessian 是一种二进制序列化框架,支持多种编程语言。RocketMQ 支持使用 Hessian 将消息序列化为二进制数据进行传输。

本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)

RocketMQ概论 的相关文章

  • SPNEGO 密码身份验证问题

    我已将我的应用程序配置为通过 SPNEGO 与 Websphere 使用 Kerberos 身份验证 这是详细信息 krb5 conf libdefaults default realm ABC MYCOMPANY COM default
  • 查询 MongoDB 集合中的字段。

    我正在尝试查询 mongodb 集合中的特定字段 这是我的代码和输出 Mongo m new Mongo DB db m getDB mydb DBCollection coll db getCollection student addin
  • 重写 getPreferredSize() 会破坏 LSP

    我总是在这个压倒一切的网站上看到建议getPreferredSize 而不是使用setPreferredSize 例如 如前面的线程所示 对于固定大小的组件 使用重写 getPreferredSize 而不是使用 setPreferredS
  • Java 卡布局。多张卡中的一个组件

    一个组件 例如JLabel 在多张卡中使用CardLayout 目前看来该组件仅出现在它添加到的最后一张卡上 如果有办法做到这一点 我应该吗 这是不好的做法吗 或者有其他选择吗 你是对的 它只出现在 添加到的最后一张卡 中 但这与CardL
  • 使用 Spring 控制器处理错误 404

    I use ExceptionHandler处理我的网络应用程序抛出的异常 在我的例子中我的应用程序返回JSON回应HTTP status用于对客户端的错误响应 但是 我正在尝试弄清楚如何处理error 404返回与处理的类似的 JSON
  • 以有效的方式从 Map 中删除多个键?

    我有一个Map
  • 无法从资源加载图片

    So I am trying to load a image file from a resource so that when I export my application into a jar file it could be use
  • 无法在 Java 中输出正确的哈希值。怎么了?

    在我的 Android 应用程序中 我有一个 SHA256 哈希值 我必须使用 RIPEMD160 消息摘要算法进一步对其进行哈希值 我可以输出任何字符串的正确 sha256 和ripemd160 哈希值 但是当我尝试使用ripemd160
  • Struts 1 到 Spring 迁移 - 策略

    我有一个legacy银行应用程序编码为Struts 1 JSP现在的要求是迁移后端 目前为 MVC to Springboot MVC 后续UI JSP 将迁移到angular Caveats 1 后端不是无状态的 2 会话对象中存储了大量
  • 插入时的 iBatis 判别器

    我有一个抽象类Example以及与之相伴的具体子类 我使用鉴别器来提取数据out数据库的 像这样
  • 如何从字符串中解析一个大整数? [复制]

    这个问题在这里已经有答案了 我有一个这样的方法 Integer parseInt myInt 不是这个整数变得很长 我得到以下异常 java lang NumberFormatException For input string 40001
  • C 与 C++ 中的 JNI 调用不同?

    所以我有以下使用 Java 本机接口的 C 代码 但是我想将其转换为 C 但不知道如何转换 include
  • BadPaddingException:无效的密文

    我需要一些帮助 因为这是我第一次编写加密代码 加密代码似乎工作正常 但解密会引发错误 我得到的错误是 de flexiprovider api exceptions BadPaddingException 无效的密文 in the 解密函数
  • 如何以编程方式创建 CardView

    我正在开发一个 Android 应用程序Java Android Studio 我想在活动中创建CardView以编程方式 我想将以下属性设置为CardView layout width wrap content layout row 0
  • javax.media.jai 类的公共下载?

    这是一个非常简单的问题 我一直在寻找可以下载 javax media jai 库的地方 我找到了 jai imageio 库 但是我发现的所有其他 jai 内容要么已经过时 2008 年及之前 然后我遇到了登录屏幕 是否有 javax me
  • 使用 PC/SC 读卡器验证 Ultralight EV1

    我在尝试使用 Java 中的 PC SC 读卡器 特别是 ACR1222L 验证 Ultralight EV1 卡时遇到问题 我能够使用 ISO 14443 3 标签的相应 APDU 在不受保护的标签上进行写入和读取 但是 我找不到运行 P
  • Java/MongoDB 按日期查询

    我将一个值作为 java util Date 存储在我的集合中 但是当我查询以获取两个特定日期之间的值时 我最终得到的值超出了范围 这是我的代码 插入 BasicDBObject object new BasicDBObject objec
  • 如何建立与 FileZilla Server 1.2.0 的 FTPS 数据连接

    使用 Apache commons net 的 Java FTPSClient 进行会话恢复是一个已知问题 会话恢复是 FTPS 服务器数据连接所需的一项安全功能 Apache FTPSClient 不支持会话恢复 并且 JDK API 使
  • 决策树和规则引擎 (Drools)

    In the application that I m working on right now I need to periodically check eligibility of tens of thousands of object
  • java中void的作用是什么?

    返回类型 方法返回值的数据类型 如果方法不返回值 则返回 void http download oracle com javase tutorial java javaOO methods html http download oracle

随机推荐