消息中间件学习总结（23）—— KafkaRabbitMQPulsarRocketMQ 基本原理和选型

Posted 2022-11-23 科技D人生

消息队列使用场景

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，削峰填谷等问题。实现高性能、高可用、可伸缩和最终一致性架构。

• 解耦：多个服务监听、处理同一条消息，避免多次 rpc 调用

• 异步消息：消息发布者不用等待消息处理的的结果

• 削峰填谷：较大流量、写入场景，为下游 I/O 服务抗流量。当然大流量下就需要使用其他方案了。

• 消息驱动框架：We 在事件总线中，服务通过监听事件消息驱动服务完成相应动作。

消息队列模式

点对点模式，不可重复消费

多个生产者可以向同一个消息队列发送消息，一个消息在被一个消息者消费成功后，这条消息会被移除，其他消费者无法处理该消息。如果消费者处理一个消息失败了，那么这条消息会重新被消费。

 发布/订阅模式

发布订阅模式需要进行注册、订阅，根据注册消费对应的消息。多个生产者可以将消息写到同一个 Topic 中，多种消息可以被同一个消费者消费。一个生产者生产的消息，同样也可以被多个消费者消费，只要他们进行过消息订阅。

选型参考

• 消息顺序：发送到队列的消息，消费时是否可以保证消费的顺序；
• 伸缩：当消息队列性能有问题，比如消费太慢，是否可以快速支持库容；当消费队列过多，浪费系统资源，是否可以支持缩容。
• 消息留存：消息消费成功后，是否还会继续保留在消息队列。
• 容错性：当一条消息消费失败后，是否有一些机制，保证这条消息是一种能成功，比如异步第三方退款消息，需要保证这条消息消费掉，才能确定给用户退款成功，所以必须保证这条消息消费成功的准确性。
• 消息可靠性：是否会存在丢消息的情况，比如有 A/B 两个消息，最后只有B消息能消费，A 消息丢失；
• 消息时序：主要包括“消息存活时间”和“延迟消息”；
• 吞吐量：支持的最高并发数；
• 消息路由：根据路由规则，只订阅匹配路由规则的消息，比如有 A/B 两者规则的消息，消费者可以只订阅 A 消息，B 消息不会消费。

Kafka

Kafka 是由 Apache 软件基金会开发的一个开源流处理平台，由 Scala 和 Java 编写。 该项目的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。 其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”，这使它作为企业级基础设施来处理流式数据非常有价值。（维基百科）

基本术语

Producer：消息生产者。一般情况下，一条消息会被发送到特定的主题上。通常情况下，写入的消息会通过轮询将消息写入各分区。生产者也可以通过设定消息key值将消息写入指定分区。写入分区的数据越均匀Kafka的性能才能更好发挥。

Topic：Topic是个抽象的虚拟概念，一个集群可以有多个Topic，作为一类消息的标识。一个生产者将消息发送到topic，消费者通过订阅Topic获取分区消息。

Partition：Partition是个物理概念，一个Topic对应一个或多个Partition。新消息会以追加的方式写入分区里，在同一个Partition里消息是有序的。Kafka通过分区，实现消息的冗余和伸缩性，以及支持物理上的并发读、写，大大提高了吞吐量。

Replicas：一个Partition有多个Replicas副本。这些副本保存在broker，每个broker存储着成百上千个不同主题和分区的副本，存储的内容分为两种：master副本，每个Partition都有一个master副本，所有内容的写入和消费都会经过master副本；follower副本不处理任何客户端的请求，只同步master的内容进行复制。如果master发生了异常，很快会有一个follower成为新的master。

Consumer：消息读取者。消费者订阅主题，并按照一定顺序读取消息。Kafka保证每个分区只能被一个消费者使用。

Offset：偏移量是一种元数据，是不断递增的整数。在消息写入时Kafka会把它添加到消息里。在分区内偏移量是唯一的。消费过程中，会将最后读取的偏移量存储在Kafka中，消费者关闭偏移量不会丢失，重启会继续从上次位置开始消费。

Broker：独立的Kafka服务器。一个Topic有N个Partition，一个集群有N个Broker，那么每个Broker都会存储一个这个Topic的Partition。如果某topic有N个partition，集群有(N+M)个broker，那么其中有N个broker存储该topic的一个partition，剩下的M个broker不存储该topic的partition数据。如果某topic有N个partition，集群中broker数目少于N个，那么一个broker存储该topic的一个或多个partition。在实际生产环境中，尽量避免这种情况的发生，这种情况容易导致Kafka集群数据不均衡。

系统框架

 第一个 topic 有两个生产，新消息被写入到 partition 1 或者 partition 2，两个分区在 broker1、broker2 都有备份。有新消息写入后，两个 follower 分区会从两个 master 分区同步变更。对应的 consumer 会从两个 master 分区根据现在 offset 获取消息，并更新 offset。 第二个 topic 只有一个生产者，同样对应两个 partition，分散在 Kafka 集群的两个 broker 上。有新消息写入，两个 follower 分区会同步 master 变更。两个 Consumer 分别从不同的 master 分区获取消息。

优点

高吞吐量、低延迟：kafka 每秒可以处理几十万条消息，它的延迟最低只有几毫秒；

可扩展性：kafka 集群支持热扩展；

持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失；

容错性：允许集群中节点故障，一个数据多个副本，少数机器宕机，不会丢失数据；

高并发：支持数千个客户端同时读写。

缺点

分区有序：仅在同一分区内保证有序，无法实现全局有序；

无延时消息：消费顺序是按照写入时的顺序，不支持延时消息

重复消费：消费系统宕机、重启导致 offset 未提交；

Rebalance：Rebalance 的过程中 consumer group 下的所有消费者实例都会停止工作，等待 Rebalance 过程完成。

使用场景

日志收集：大量的日志消息先写入 kafka，数据服务通过消费 kafka 消息将数据落地；

消息系统：解耦生产者和消费者、缓存消息等；

用户活动跟踪：kafka 经常被用来记录 web 用户或者 app 用户的各种活动，如浏览网页、搜索、点击等活动，

这些活动信息被各个服务器发布到 kafka 的 topic 中，然后消费者通过订阅这些 topic 来做实时的监控分析，亦可保存到数据库；

运营指标：记录运营、监控数据，包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告；

流式处理：比如 spark streaming

RabbitMQ

RabbitMQ 是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件（英语：Message-oriented middleware））。RabbitMQ 服务器是用 Erlang 语言编写的，而群集和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端函式库。（维基百科）

基本术语

Broker：接收客户端链接实体，实现AMQP消息队列和路由功能；

Virtual Host：是一个虚拟概念，权限控制的最小单位。一个Virtual Host里包含多个Exchange和Queue；

Exchange：接收消息生产者的消息并将消息转发到队列。发送消息时根据不同ExchangeType的决定路由规则，ExchangeType常用的有：direct、fanout和topic三种；

Message Queue：消息队列，存储为被消费的消息；

Message：由Header和Body组成，Header是生产者添加的各种属性，包含Message是否持久化、哪个MessageQueue接收、优先级。Body是具体的消息内容；

Binding：Binding连接起了Exchange和Message Queue。在服务器运行时，会生成一张路由表，这张路由表上记录着MessageQueue的条件和BindingKey值。当Exchange收到消息后，会解析消息中的Header得到BindingKey，并根据路由表和ExchangeType将消息发送到对应的MessageQueue。最终的匹配模式是由ExchangeType决定；

Connection：在Broker和客户端之间的TCP连接；

Channel：信道。Broker和客户端只有tcp连接是不能发送消息的，必须创建信道。AMQP协议规定只有通过Channel才能执行AMQP命令。一个Connection可以包含多个Channel。之所以需要建立Channel，是因为每个TCP连接都是很宝贵的。如果每个客户端、每个线程都需要和Broker交互，都需要维护一个TCP连接的话是机器耗费资源的，一般建议共享Connection。RabbitMQ不建议客户端线程之前共享Channel，至少保证同一Channel发小消息是穿行的。

Command：AMQP命令，客户端通过Command来完成和AMQP服务器的交互。

系统框架

 一条 Message 经过信道到达对应的 Exchange，Exchange 收到消息后解析出消息 Header 内容，获取消息 BindingKey 并根据 Binding 和 ExchangeType 将消息转发到对应的 MessageQueue，最后通过 Connection 将消息传送的客户端。

ExchangeType

Direct：精确匹配

• 只有RoutingKey和BindingKey完全匹配的时候，消息队列才可以获取消息
• Broker默认提供一个Exchange，类型是Direct名字是空字符串，绑定到所有的Queue（这里通过Queue名字来区分）

Fanout：订阅、广播

• 这个模式会将消息转发到所有的路由的 Queue 中

Topic：通配符模式

• RoutingKey为一个句点号“. ”分隔的字符串（将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“quick.orange.rabbit”。BindingKey与RoutingKey一样
• Bindingkey中的两个特殊字符"#"和“”用于模糊匹配，“#”用于匹配多个单次，“”用来匹配单个单词（包含零个）

优点

• 基于AMQP协议：除了Qpid，RabbitMQ是唯一一个实现了AMQP标准的消息服务器；
• 健壮、稳定、易用；
• 社区活跃，文档完善；
• 支持定时消息；
• 可插入的身份验证，授权，支持TLS和LDAP；
• 支持根据消息标识查询消息，也支持根据消息内容查询消息。

缺点

• erlang开发源码难懂，不利于做二次开发和维护；
• 接口和协议复杂，学习和维护成本较高。

总结

• erlang有并发优势，性能较好。虽然源码复杂，但是社区活跃度高，可以解决开发中遇到的问题；
• 业务流量不大的话可以选择功能比较完备的 RabbitMQ。

Pulsar

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性，被看作是云原生时代实时消息流传输、存储和计算最佳解决方案。Pulsar 是一个 pub-sub (发布-订阅)模型的消息队列系统。（百科）

基本术语

Property：代表租户，每个property都可以代表一个团队、一个功能、一个产品线。一个property可包含多个namesapce，多租户是一种资源隔离手段，可以提高资源利用率；

Namespace：Pulsar的基本管理单元，在namaspace级别可设置权限、消息TTL、Retention 策略等。一个namaspace里的所有topic都继承相同的设置。命名空间分为两种：本地命名空间，只在集群内可见、全局命名空间对多个集群可见集群命名空间；

Producer：数据生产方，负责创建消息并将消息投递到 Pulsar 中；

Consumer：数据消费方，连接到 Pulsar接收消息并进行相应的处理；

Broker：无状态Proxy服务，负责接收消息、传递消息、集群负载均衡等操作，它对 client 屏蔽了服务端读写流程的复杂性，是保证数据一致性与数据负载均衡的重要角色。Broker 不会持久化保存元数据。可以扩容但不能缩容；

BookKeeper：有状态，负责持久化存储消息。当集群扩容时，Pulsar会在新增BookKeeper和Segment（即 Bookeeper 的 Ledger），不需要像kafka一样在扩容时进行Rebalance。扩容结果是 Fragments跨多个Bookies以带状分布，同一个Ledger的Fragments 分布在多个Bookie上，导致读取和写入会在多个 Bookies 之间跳跃；

ZooKeeper：存储 Pulsar 、 BookKeeper 的元数据，集群配置等信息，负责集群间的协调、服务发现等；

Topic：用作从producer到consumer传输消息。Pulsar在Topic级别拥有一个leader Broker，称之为拥有 Topic 的所有权，针对该 Topic 所有的 R/W 都经过该 Broker 完成。Topic的 Ledger 和 Fragment 之间映射关系等元数据存储在 Zookeeper 中，Pulsar Broker 需要实时跟踪这些关系进行读写流程；

Ledger：即Segment，Pulsar底层数据以Ledger的形式存储在BookKeeper上。是Pulsar删除的最小单位；

Fragment ： 每个 Ledger 由若干 Fragment 组成。

系统框架

上面框架图分别演示了扩容、故障转移两种情况。扩容：因业务量增大扩容新增Bookie N，后续写入的数据segment x、segment y写入新增Bookie中，为保持均衡扩容结果如上图绿色模块所示。故障转移：Bookie 2的segment 4发生故障，Pulasr的Topic会立马从新选择Bookie 1作为处理读写的服务。 Broker是无状态的服务，只服务数据计算不存储，所以Pulsar 可以认为是一种基于 Proxy 的分布式系统。

优点

• 灵活扩容
• 无缝故障恢复
• 支持延时消息
• 内置的复制功能，用于跨地域复制如灾备
• 支持两种消费模型：流（独享模式）、队列（共享模式）

RocketMQ

RocketMQ 是一个分布式消息和流数据平台，具有低延迟、高性能、高可靠性、万亿级容量和灵活的可扩展性。RocketMQ 是2012年阿里巴巴开源的第三代分布式消息中间件。（维基百科）

基本术语

Topic：一个Topic可以有0个、1个、多个生产者向其发送消息，一个生产者也可以同时向不同的Topic发送消息。一个Topic也可以被0个、1个、多个消费者订阅；

Tag：消息二级类型，可以为用户提供额外的灵活度，一条消息可以没有tag；

Producer：消息生产者；

Broker：存储消息，以Topic为纬度轻量级的队列；转发消息，单个Broker节点与所有的NameServer节点保持长连接及心跳，会定时将Topic信息注册到NameServer；

Consumer：消息消费者，负责接收并消费消息；

MessageQueue：消息的物理管理单位，一个Topic可以有多个Queue，Queue的引入实现了水平扩展的能力；

NameServer：负责对原数据的管理，包括Topic和路由信息，每个NameServer之间是没有通信的；

Group：一个组可以订阅多个Topic，ProducerGroup、ConsumerGroup分别是一类生产者和一类消费者；

Offset：通过Offset访问存储单元，RocketMQ中所有消息都是持久化的，且存储单元定长。Offset为Java Long类型，理论上100年内不会溢出，所以认为Message Queue是无限长的数据，Offset是下标；

Consumer：支持PUSH和PULL两种消费模式，支持集群消费和广播消费。

系统框架

优点

支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型；

• 顺序队列：在一个队列中可靠的先进先出（FIFO）和严格的顺序传递；
  支持拉（pull）和推（push）两种消息模式；
• 单一队列百万消息的堆积能力；
• 支持多种消息协议，如 JMS、MQTT 等；
• 分布式横向扩展架构
• 满足至少一次消息传递语义；
• 提供丰富的Dashboard，包含配置、指标和监控等。

缺点

• 支持的客户端语言不多，目前是java及c++，其中c++不成熟
• 社区活跃度一般
• 延时消息：开源版不支持任意时间精度，仅支持特定的level

使用场景

• 为金融互联网领域而生，对于可靠性要求很高的场景