Pulsar 下一代消息平台

引言

提起成熟的消息队列或消息引擎，毋庸置疑，大多数人的第一反应一定是 Kafka。

Kafka 能够彻底满足海量数据场景下高吞吐、高并发需求，在短短几年内，已经被阿里、腾讯、百度、字节跳动、Netflix、Twitter 等超一线大厂视为技术核心——可以说，Kafka 是目前大数据 Spark 实时流处理的标配。

Kafka 的优势

Kafka 具有高吞吐量、低延迟、容错、持久性、可伸缩性，尤其是广为人知的高吞吐量，Kafka 每秒大约可以生产约 25 万消息（50 MB），每秒处理 55 万消息（110 MB）！Kafka 还有一个巨大的优势就是容错，它具备一个固有功能，可以自行应对集群中的节点故障。

Kafka 的不足

在运维和实践过程中，Kafka 却也始终存在着一些棘手的问题。例如：

• 扩展性较差，剥离 Broker 意味着必须复制 topic 分区和副本，效率很低；

• 缺乏一致性，一旦 API 发生变化很有可能出现问题；

• 存储成本非常高，几乎没有人用Kafka长时间存储数据；

• 没有与租户完全隔离的本地多租户，需要自行配置解决方案

Pulsar 背景介绍

俯瞰技术生态，有没有一个平台，能够既拥有 Kafka 的优势，又规避它的缺陷，同时还融合了 MQ 的一系列特性呢？

有，那就是 Pulsar。

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体。该系统源于 Yahoo，最初在 Yahoo 内部开发和部署，支持 Yahoo 应用服务平台 140 万个主题，日处理超过 1000 亿条消息。

Pulsar 于 2016 年由 Yahoo 开源并捐赠给 Apache 软件基金会进行孵化，2018 年成为 Apache 软件基金会顶级项目。

Pulsar 作为下一代云原生分布式消息流平台，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐以及低延时的高可扩展流数据存储特性，内置诸多其他系统商业版本才有的特性，是云原生时代解决实时消息流数据传输、存储和计算的最佳解决方案。

目前，Apache Pulsar 已经应用部署在国内外众多大型互联网公司和传统行业公司，案例分布在人工智能、金融、电信运营商、直播与短视频、物联网、零售与电子商务、在线教育等多个行业，如美国有线电视网络巨头 Comcast、Yahoo！、腾讯、中国电信、中国移动、BIGO、VIPKID 等。

目前 Apache Pulsar 项目原生核心贡献者已组成创业公司 StreamNative，进一步为 Apache Pulsar 提供更好的企业级服务支持与生态建设。

Plusar 是什么

"Pulsar is a distributed pub-sub messaging platform with a very flexible messaging model and an intuitive client API."
Pulsar 是 pub-sub（发布者/订阅者）模式的分布式消息平台，拥有灵活的消息模型和直观的客户端 API。

核心概念

Topic（主题）

Topic 是Pulsar的核心概念，表示一个“channel”，Producer 可以写入数据，Consumer 从中消费数据（与 Kafka 类似）。

Topic 名称的 URL 类似如下的结构：

{persistent|non-persistent}://tenant/namespace/topic

• persistent|non-persistent：表示数据是否持久化（Pulsar支持消息持久化和非持久化两种模式）
• Tenant：表示为租户
• Namespace：一般聚合一系列相关的Topic，一个租户下可以有多个Namespace

Tenant（租户）和Namespace（名称空间）关系

上图中 Property 即为租户，每个租户下可以有多个 Namespace，每个 Namespace 下有多个 Topic。
Namespace 是 Pulsar 中的操作单元，包括Topic是配置在Namespace级别的，包括多地域复制，消息过期策略等都是配置在Namespace上的。

订阅模型

Pulsar 提供了灵活的消息模型，支持三种订阅类型：

• Exclusive subscription：排他的，只能有一个Consumer，接收一个Topic所有的消息

• Shared subscription：共享的，可以同时存在多个Consumer，每个Consumer处理Topic中一部消息（Shared模型是不保证消息顺序的，Consumer数量可以超过分区的数量）

• Failover subscription：Failover模式，同一时刻只有一个有效的Consumer，其余的Consumer作为备用节点，在Master Consumer不可用后进行替代（看起来适用于数据量小，且解决单点故障的场景）
  

分区（Partition）

为了解决吞吐等问题，Pulsar和Kafka一样，采用了分区（Partition）的机制。

Pulsar提供了一些策略来处理消息到Partition的路由（MessageRouter）：

• Single partitioning：Producer随机选择一个Partition并将所有消息写入到这个分区
• Round robin partitioning ：采用Round robin的方式，轮训所有分区进行消息写入
• Hash partitioning：这种模式每条消息有一个Key，Producer根据消息的Key的哈希值进行分区的选择（Key相同的消息可以保证顺序）。
• Custom partitioning：用户自定义路由策略

不同于别的MQ系统，Pulsar允许Consumer的数量超过分区的数量（对于RocketMQ，超过分区数的Consumer会分配不到分区而“空跑”）。

在Shared subscription的订阅模式下，Consumer数量可以大于分区的数量，每个Consumer处理每个Partition中的一部分消息，不保证消息的顺序。

持久化

Pulsar通过 BookKeeper 来存储消息，保证消息不会丢失（BookKeeper：A scalable, fault-tolerant, and low-latency storage service optimized for real-time workloads）。

架构

Pulsar采用“存储和服务分离”的两层架构（这是Pulsar区别于其他MQ系统最重要的一点，也是所谓的“下一代消息系统”的核心）：

• Broker：提供发布和订阅的服务（Pulsar的组件）

• Bookie：提供存储能力（BookKeeper的存储组件）

优势是Broker成为了stateless的组件，可以水平扩容（RocketMQ的Broker是包含存储的，是有状态的，Broker的扩容更像是“拆分”）。高可靠，一致性等通过BookKeeper去保证。

上图是Pulsar Cluster的架构：

采用 ZooKeeper 存储元数据，集群配置，作为coordination

• local zk负责Pulsar Cluster内部的配置等

• global zk则用于Pulsar Cluster之间的数据复制等

采用Bookie作为存储设备(大多数MQ系统都采用本地磁盘或者DB作为存储设备)

• Broker负责负载均衡和消息的读取、写入等

• Global replicators负责集群间的数据复制

GEO-REPLICATOIN

多个Broker节点组成一个Pulsar Cluster；多个Pulsar Cluster组成一个Pulsar Instance。

Pulsar通过GEO-REPLICATION支持一个Instance内在不同的地域发送和消费消息。

上图中，Producer P1、P2、P3在不同的Cluster发送给Topic T1的消息，会在Cluster之间进行复制，Consumer C1、C2可以在自己所在的Cluster消费到所有的消息。

当消息被写入Pulsar时，首先消息被持久化在local cluster，之后异步的发送到其他cluster。在没有链接问题的情况下，通常复制的latency相近于网络的RTT。

Pulsar的应用

• 作为普通的Pub-Sub模型的消息队列使用，类似于RocketMQ
• 支持Function（Stream），整合到 Stream 平台（如 Spark）

总结

Apache Pulsar 提供了统一的消费模型，支持 Stream（如 Kafka）和 Queue（如 RabbitMQ）两种消费模型， 支持 exclusive、failover 和 shared 三种消费模式。

同时，Pulsar 提供和 Kafka 兼容的 API，以及 Kafka-On-Pulsar（KoP） 组件来兼容 Kafka 的应用程序，KoP 在 Pulsar Broker 中解析 Kafka 协议，用户不用改动客户端的任何 Kafka 代码就能直接使用 Pulsar。

一句话，Plusar 是对 Kafka 的增强或升级。