本文主要介绍 Pulsar 的特性及架构，对应的 pulsar 版本为 3.3.x。

1、Pulsar 特性

<li>Pulsar实例原生支持多个集群，并且在集群之间实现了无缝的地理复制消息。
• 极低的发布和端到端延迟。
• 无缝扩展至超过一百万个主题。
• 简单的客户端 API，支持 Java、Go、Python 和 C++ 语言。
• 多种订阅类型（独占、共享和故障转移）。
• 通过Apache BookKeeper提供的持久化消息存储，保证消息传递。无服务器轻量计算框架Pulsar Functions提供流原生数据处理能力。
• 无服务器连接器框架Pulsar IO基于Pulsar Functions构建，使得数据在Apache Pulsar内外的移动更加容易。
• 分层存储在数据老化时将其从热/温存储卸载到冷/长期存储（例如S3和GCS）。

2、架构

2.1、架构概述

在最高级别上，Pulsar 实例由一个或多个 Pulsar 集群组成。实例内的集群间会同步数据。

一个 pulsar 包含以下组件：

• 一个或多个 broker， 负责处理和负载均衡来自生产者的消息，并将消息分派给消费者，与 pulsar 配置存储通信以处理各种协调任务，将消息存储在 BookKeeper 实例（又称 bookies）中，依赖于 ZooKeeper 集群执行某些任务等。
• 一个或多个 bookie 组成的 BookKeeper 集群，负责处理消息的持久存储。
• 一个 ZooKeeper 集群，负责处理 pulsar 集群之间的协调任务。

下图表示了一个 pulsar 集群：

在更广泛的实例级别，一个称为配置存储的实例级 ZooKeeper 集群处理涉及多个集群的协调任务，例如地理复制。

2.2、Brokers

broker 是一个无状态组件，主要负责运行另外两个组件：

<li>一个 HTTP 服务器，暴露 REST API，用于管理任务和生产者与消费者的主题查找。生产者连接到 broker 以发布消息，消费者连接到 broker 以消费消息。
• 一个调度程序，是一个基于自定义二进制协议的异步 TCP 服务器，用于所有数据传输。

为了提高性能，消息通常会从受控分布式日志缓存中分派，除非积压超过了缓存大小。如果积压量过大，超出了缓存的处理能力，broker 将从 BookKeeper 中读取条目。

最后，为了支持全局主题的地理复制，broker 管理复制器，这些复制器追踪本地区发布的条目，并使用 Pulsar Java 客户端将它们重新发布到远程区域。

2.3、集群

一个 pulsar 实例包括一个或多个 pulsar 集群。集群本身由以下组成：

• 一个或多个 broker
• 用于集群级配置和协调的 ZooKeeper 集群
• 用于消息持久存储的 bookie 集合

可以通过地理复制在集群间进行数据复制。

2.4、元数据存储

Pulsar 的元数据存储维护着整个 Pulsar 集群的所有元数据，例如主题元数据、模式、broker 负载数据等。Pulsar 使用 ZooKeeper 来存储元数据、进行集群配置和协调。Pulsar 的元数据存储可以单独部署在一个 ZooKeeper 集群上，也可以部署在已有的 ZooKeeper 集群上。可以使用同一个 ZooKeeper 集群来存储 Pulsar 的元数据和 BookKeeper 的元数据。如果想要部署 broker 以连到现有的 BookKeeper 集群 ，那么需要分别为 Pulsar 元数据存储和 BookKeeper 元数据存储部署独立的 ZooKeeper 集群。

Pulsar 还支持更多的元数据后端服务，包括 etcd 和 RocksDB（仅适用于单机的 Pulsar）。

在 Pulsar 实例中：

• 配置存储中存储了租户、命名空间及其他需要全局一致性的实体配置。
• 每个集群都有自己的本地 ZooKeeper 集群，用于存储集群特定的配置和协调信息，例如哪些 broker 负责哪些主题、所有权元数据、broker 负载报告、BookKeeper ledger 元数据等。

2.5、配置存储

配置存储是一个 ZooKeeper 集群，用于处理与配置相关的任务，并维护 Pulsar 实例的所有配置，例如集群、租户、命名空间、分区主题相关的配置等。一个 Pulsar 实例可以有一个本地集群、多个本地集群或多个跨区域集群。因此，配置存储可以在 Pulsar 实例中的多个集群之间共享。配置存储可以使用单独的 ZooKeeper 集群，也可以使用现有的 ZooKeeper 集群。

2.6、持久存储

Pulsar 为应用程序提供了可靠的消息传递保证。如果消息成功到达 pulsar broker，它将被传递给其预期的目标。

这种保证要求非确认的消息在持久化存储中保留，直到传递给消费者并收到确认。这种消息传递模式通常称为持久化消息传递。在 Pulsar 中，所有消息的 N 个副本都被存储并同步到磁盘上，例如，在两台服务器上通过镜像 RAID 卷存储 4 个副本。

2.6.1、Apache BookKeeper

Pulsar使用 Apache BookKeeper 进行持久化消息存储。BookKeeper 是一个分布式的预写式日志（WAL）系统，为 Pulsar 提供了几个关键优势：

• 它使 Pulsar 能够利用许多独立的日志(ledger)。随着时间的推移，可以为主题创建多个 ledger。
• 它为处理条目复制的顺序数据提供非常高效的存储。
• 它在各种系统故障情况下保证 ledger 的读一致性。
• 它能够在 bookies 之间均匀分配 I/O。
• 它在容量和吞吐量上具有水平可扩展性。通过向集群添加更多的 bookie，可以立即增加容量。
• Bookies 被设计为处理数千个 ledger 的并发读写。通过使用多个磁盘设备--一个用于日志，另一个用于一般存储--bookies 可以将读操作与有延迟的写操作隔离开来

除了消息数据外，BookKeeper 还持久存储游标。游标是消费者的订阅位置。BookKeeper 使得 Pulsar 能够以可扩展的方式存储消费者的位置。

目前，Pulsar 支持持久化消息存储。这体现在所有主题名称中的"persistent"一词。以下是一个示例：

persistent://my-tenant/my-namespace/my-topic

Pulsar 还支持临时的非持久化消息存储。

下图为 brokers 和 bookies 之间的交互示意图：

2.6.2、Ledgers

ledger 是一种仅追加的数据结构，只有一个写入者，它被分配给多个 BookKeeper 存储节点--bookie。ledger 条目会被复制到多个 bookie。ledger 本身具有非常简单的语义：

<li>Pulsar broker 节点可以创建 ledger，向 ledger 追加条目，并关闭 ledger。
• 一旦 ledger 被关闭--无论是显式关闭还是因为写入进程崩溃--它只能以只读模式重新打开。
• 最后，当不再需要 ledger 中的条目时，可以从系统中（所有 bookies 中）删除整个 ledger。

Ledger 读一致性

Bookkeeper 的主要优势在于，在发生故障时，它保证 ledger 的读一致性。由于 ledger 只能由单个进程写入，该进程可以非常高效地追加条目，无需获得共享锁。在发生故障后，ledger 将经历恢复过程，该过程将最终确定 ledger 的状态，并确定最后提交的条目。在那之后，所有读取 ledger 的操作都保证能看到完全相同的内容。

托管 ledgers

鉴于 Bookkeeper ledger 提供了单一的日志抽象，因此在 ledger 之上开发了一个名为"managed ledger"的库，它代表了单个主题的存储层。托管 ledger 表示消息流的抽象，单一的写入者在流的末尾不断追加消息；并且有多个消费者游标在消费这个流，每个游标都有其关联的位置。

在内部，单个托管 ledger 使用多个BookKeeper ledger 来存储数据。有两个理由使用多个 ledger：

• 在发生故障后，ledger 不再可写，需要创建一个新的 ledger。
• 当所有游标消费完 ledger 中的消息后，可以删除该 ledger。这允许定期滚动 ledger。

2.6.3、Journal 存储

在 BookKeeper 中，Journal 文件包含 BookKeeper 事务日志。在对 ledger 进行更新之前，Bookie 需要确保将描述更新的事务写入持久（非易失性）存储。一旦 Bookie 启动或者旧的 journal 文件大小达到了阈值（通过参数 journalMaxSizeMB 配置），就会创建一个新的 journal 文件。

2.7、Pulsar 代理

Pulsar 客户端与 Pulsar 集群进行交互的一种方式是直接连接到 Pulsar broker。然而，在某些情况下，这种直接连接不可行或不可取，因为客户端无法直接访问 broker 地址。例如，如果在云环境、Kubernetes 或类似的平台上运行 Pulsar，那么直接连接客户端到 broker 的方式可能是不可行的。

Pulsar 代理通过充当集群中所有 broker 的单一网关，提供了解决这个问题的方案。如果运行 Pulsar 代理（再次强调，这是可选的），客户端将通过代理与 Pulsar 集群连接，而不是直接与 broker 通信。

为了性能和容错性，请根据需要运行多个 Pulsar 代理实例。

在架构上，Pulsar 代理从 ZooKeeper 获取其所需的所有信息。在启动代理时，只需提供元数据存储和配置存储的地址。以下是一个示例：

bin/pulsar proxy \

--metadata-store zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181 \

--configuration-metadata-store zk:my-zk-1:2181,my-zk-2:2181,my-zk-3:2181

Pulsar 代理的详细说明，可参阅 Pulsar 代理管理文档。

关于 Pulsar 代理的一些重要事项：

• 客户端使用 Pulsar 代理时无需提供任何特定的配置。不需要更新现有应用程序的客户端配置，除服务 URL 的 IP 改变了（例如，在 Pulsar 代理上运行负载均衡器）。
• Pulsar 代理支持 TLS 加密和 mTLS 身份验证。

2.8、服务发现

服务发现是一种机制，它使客户端能够使用单个 URL 与整个 Pulsar 实例进行交互。

如果愿意，可以使用自己的服务发现系统。如果使用自己的系统，只有一个要求：当客户端对一个地址发起 HTTP 请求，比如 http://pulsar.us-west.example.com:8080 时，需要将客户端重定向到集群中的某个活跃 broker，可以通过 DNS、HTTP 或 IP 重定向，或者其他方式实现。

下图说明了 Pulsar 的服务发现：

在这个图示中，Pulsar 集群通过一个单一的 DNS 名称进行寻址：pulsar-cluster.acme.com。Python 客户端可以这样访问这个 Pulsar 集群：

from pulsar import Client

client = Client('pulsar://pulsar-cluster.acme.com:6650')

注意：在 Pulsar 中，每个主题仅由一个 broker 处理。客户端对主题的初始读取、更新或删除请求可能会发送到不是该主题所有者的 broker。如果 broker 无法处理此主题的请求，它会将请求重定向到适当的 broker。

参考：https://pulsar.apache.org/docs/3.3.x/concepts-architecture-overview/。