Mars
0%

Pulsar与Kafka对比

发表于 更新于 分类于 ，

Pulsar与Kafka对比

Pulsar 是一个比较新的MQ，和MQ的王者Kafka进行对比。

TIM截图20191125094052.png

Pulsar架构

Apache Pulsar 是一个企业级的分布式消息系统，最初由 Yahoo 开发，在 2016 年开源，并于2018年9月毕业成为 Apache 基金会的顶级项目。Pulsar 已经在 Yahoo 的生产环境使用了三年多，主要服务于Mail、Finance、Sports、 Flickr、 the Gemini Ads platform、 Sherpa (Yahoo 的 KV 存储)。

undefined

kafka队列优先级模型

undefined

在该方案中，个推将优先级统一设定为高、中、低三个级别。具体操作方案如下：

  1. 对某个优先级根据 task (单次推送任务)维度，存入不同的 Topic，一个 task 只写入一个 Topic，一个 Topic 可存多个 task；
  2. 消费模块根据优先级配额(如 6:3:1)，获取不同优先级的消息数，同一优先级轮询获取消息；这样既保证了高优先级用户可以更快地发送消息，又避免了低优先级用户出现没有下发的情况。

Kafka方案遇到的问题

随着个推业务的不断发展，接入的 APP 数量逐渐增多，第一版的优先级方案也逐渐暴露出一些问题：

  1. 当相同优先级的 APP 在同一时刻推送任务越来越多时，后面进入的 task 消息会因为前面 task 消息还存在队列情况而出现延迟。如下图所示, 当 task1 消息量过大时，在task1 消费结束前，taskN 将一直处于等待状态。

undefined

  1. Kafka 在 Topic 数量由 64 增长到 256 时，吞吐量下降严重，Kafka 的每个 Topic、每个分区都会对应一个物理文件。当 Topic 数量增加时，消息分散的落盘策略会导致磁盘 IO 竞争激烈，因此我们不能仅通过增加 Topic 数量来缓解第一点中的问题。

基于上述问题，需要可以创建大量的 Topic, 同时吞吐性能不能比 Kafka 逊色。经过一段时间的调研，Apache Pulsar 引起了我们的关注。

Topic数量

Pulsar 可以支持百万级别 Topic 数量的扩展，同时还能一直保持良好的性能。Topic 的伸缩性取决于它的内部组织和存储方式。Pulsar 的数据保存在 bookie (BookKeeper 服务器)上，处于写状态的不同 Topic 的消息，在内存中排序，最终聚合保存到大文件中，在 Bookie 中需要更少的文件句柄。另一方面 Bookie 的 IO 更少依赖于文件系统的 Pagecache，Pulsar 也因此能够支持大量的主题。

消费模型

Pulsar 支持三种消费模型：Exclusive、Shared 和Failover。

undefined

**Exclusive (独享)**：一个 Topic 只能被一个消费者消费。Pulsar 默认使用这种模式。

**Shared(共享)**：共享模式，多个消费者可以连接到同一个 Topic，消息依次分发给消费者。当一个消费者宕机或者主动断开连接时，那么分发给这个消费者的未确认(ack)的消息会得到重新调度，分发给其他消费者。

**Failover (灾备)**：一个订阅同时只有一个消费者，可以有多个备份消费者。一旦主消费者故障，则备份消费者接管。不会出现同时有两个活跃的消费者。

Exclusive和Failover订阅，仅允许一个消费者来使用和消费每个订阅的Topic。这两种模式都按 Topic 分区顺序使用消息。它们最适用于需要严格消息顺序的流(Stream)用例。

Shared 允许每个主题分区有多个消费者。同一个订阅中的每个消费者仅接收Topic分区的一部分消息。Shared最适用于不需要保证消息顺序队列(Queue)的使用模式，并且可以按照需要任意扩展消费者的数量。

存储

Pulsar 引入了 Apache BookKeeper 作为存储层，BookKeeper 是一个专门为实时系统优化过的分布式存储系统，具有可扩展、高可用、低延迟等特性。具体介绍，请参考 BookKeeper官网。

Segment

BookKeeper以 Segment (在 BookKeeper 内部被称作 ledger) 作为存储的基本单元。从 Segment 到消息粒度，都会均匀分散到 BookKeeper 的集群中。这种机制保证了数据和服务均匀分散在 BookKeeper 集群中。

Pulsar 和 Kafka 都是基于 partition 的逻辑概念来做 Topic 存储的。最根本的不同是，Kafka 的物理存储是以 partition 为单位的，每个 partition 必须作为一个整体(一个目录)存储在某个 broker 上。 而 Pulsar 的 partition 是以 segment 作为物理存储的单位，每个 partition 会再被打散并均匀分散到多个 bookie 节点中。

这样的直接影响是，Kafka 的 partition 的大小，受制于单台 broker 的存储；而 Pulsar 的 partition 则可以利用整个集群的存储容量。

undefined

扩容

当 partition 的容量达到上限后，需要扩容的时候，如果现有的单台机器不能满足，Kafka 可能需要添加新的存储节点，并将 partition 的数据在节点之间搬移达到 rebalance 的状态。

而 Pulsar 只需添加新的 Bookie 存储节点即可。新加入的节点由于剩余空间大，会被优先使用，接收更多的新数据；整个扩容过程不涉及任何已有数据的拷贝和搬移。

undefined

Broker故障

Pulsar 在单个节点失败时也会体现同样的优势。如果 Pulsar 的某个服务节点 broker 失效，由于 broker 是无状态的，其他的 broker 可以很快接管 Topic，不会涉及 Topic 数据的拷贝；如果存储节点 Bookie 失效，在集群后台中，其他的 Bookie 会从多个 Bookie 节点中并发读取数据，并对失效节点的数据自动进行恢复，对前端服务不会造成影响。

undefined

Bookie故障

Apache BookKeeper 中的副本修复是 Segment (甚至是 Entry)级别的多对多快速修复。这种方式只会复制必须的数据，这比重新复制整个主题分区要精细。如下图所示，当错误发生时， Apache BookKeeper 可以从 bookie 3 和 bookie 4 中读取 Segment 4 中的消息，并在 bookie 1 处修复 Segment 4。所有的副本修复都在后台进行，对 Broker 和应用透明。

当某个 Bookie 节点出错时，BookKeeper会自动添加可用的新 Bookie 来替换失败的 Bookie，出错的 Bookie 中的数据在后台恢复，所有 Broker 的写入不会被打断，而且不会牺牲主题分区的可用性。

undefined

基于 Pulsar 的优先级队列方案

在设计思路上，Pulsar 方案和 Kafka 方案并没有多大区别。但在新方案中，个推技术团队借助 Pulsar 的特性，解决了 Kafka 方案中存在的问题。

  1. 根据 task 动态生成 Topic，保证了后进入的 task 不会因为其他 task 消息堆积而造成等待情况。
  2. 中高优先级 task 都独享一个 Topic，低优先级 task 共享 n 个 Topic。
  3. 相同优先级内，各个 task 轮询读取消息，配额满后流转至下一个优先级。
  4. 相同优先级内, 各个 task 可动态调整 quota， 在相同机会内，可读取更多消息。
  5. 利用 Shared 模式, 可以动态添加删除 consumer，且不会触发 Rebalance 情况。
  6. 利用 BookKeeper 特性，可以更灵活的添加存储资源。

undefined

Pulsar实践范例

  1. 不同 subscription 之间相对独立，如果想要重复消费某个 Topic 的消息，需要使用不同的 subscriptionName 订阅；但是一直增加新的 subscriptionName，backlog 会不断累积。
  2. 如果 Topic 无人订阅，发给它的消息默认会被删除。因此如果 producer 先发送，consumer 后接收，一定要确保 producer 发送之前，Topic 有 subscription 存在(哪怕 subscribe 之后 close 掉)，否则这段时间发送的消息会导致无人处理。
  3. 如果既没有人发送消息，又没有人订阅消息，一段时间后 Topic 会自动删除。
  4. Pulsar 的 TTL 等设置，是针对整个 namespace 起效的，无法针对单个 Topic。
  5. Pulsar 的键都建立在 zookeeper 的根目录上，在初始化时建议增加总节点名。
  6. 目前 Pulsar 的 java api 设计，消息默认需要显式确认，这一点跟 Kafka 不一样。
  7. Pulsar dashboard 上的 storage size 和 prometheus 上的 storage size (包含副本大小)概念不一样。
  8. 把dbStorage_rocksDB_blockCacheSize 设置的足够大；当消息体量大，出现backlog 大量堆积时, 使用默认大小(256M)会出现读耗时过大情况，导致消费变慢。
  9. 使用多 partition，提高吞吐。
  10. 在系统出现异常时，主动抓取 stats 和 stats-internal，里面有很多有用数据。
  11. 如果业务中会出现单 Topic 体量过大的情况，建议把 backlogQuotaDefaultLimitGB 设置的足够大(默认10G), 避免因为默认使用producer_request_hold 模式出现 block producer 的情况；当然可以根据实际业务选择合适的 backlogQuotaDefaultRetentionPolicy。
  12. 根据实际业务场景主动选择 backlog quota。
  13. prometheus 内如果发现读耗时为空情况，可能是因为直接读取了缓存数据；Pulsar 在读取消息时会先读取 write cache, 然后读取 read cache；如果都没有命中, 则会在 RocksDB 中读取条目位子后，再从日志文件中读取该条目。
  14. 写入消息时, Pulsar 会同步写入 journal 和 write cache；write cache 再异步写入日志文件和 RocksDB； 所以有资源的话，建议 journal 盘使用SSD。

一句话总结

Bossie Awards中对 Pulsar 点评如下：“Pulsar 旨在取代 Apache Kafka 多年的主宰地位。Pulsar在很多情况下提供了比 Kafka 更快的吞吐量和更低的延迟，并为开发人员提供了一组兼容的 API，让他们可以很轻松地从 Kafka 切换到 Pulsar。Pulsar 的最大优点在于它提供了比 Apache Kafka 更简单明了、更健壮的一系列操作功能，特别在解决可观察性、地域复制和多租户方面的问题。在运行大型 Kafka 集群方面感觉有困难的企业可以考虑转向使用 Pulsar。”