企业官网建设流程全解析

• 消息队列
• 后端
• 流处理

Apache Pulsar 的消息可靠性与多集群协调能力，建立在其依赖的两个外部 Apache 开源系统之上：负责配置管理与协调任务的 ZooKeeper，以及负责消息数据持久化存储的 BookKeeper。本指南以 Pulsar 2.4.0 官方运维文档为骨架，完整讲解本地 ZooKeeper（Local ZooKeeper）与配置存储（Configuration Store）的部署、Bookie 集群的配置与启动、BookKeeper 持久化策略（Persistence Policies）的四种核心参数，并结合本仓库中的真实配置文件与源码实现进行纵深印证。读完本文，你将能够独立完成一个 Pulsar 实例（单集群或多集群）中 ZooKeeper 与 BookKeeper 两套底层系统的规划、部署、验证与调优。

在 Pulsar 的架构中，两个系统各司其职：

• ZooKeeper承担种类繁多的配置管理与协调任务，包括 broker 发现、元数据存储等；
• BookKeeper负责消息数据的持久化存储，以分布式 write-ahead log（预写日志） 的方式保证独立消息日志（称为 ledger）的读取一致性。

ZooKeeper 与 BookKeeper 均为开源的 Apache 项目。本文将在最后通过架构图系统性地说明这两个系统在 Pulsar 集群中的角色分工。

ZooKeeper 的双层架构：Local ZooKeeper 与 Configuration Store

每个 Pulsar 实例（instance）依赖两个相互独立的 ZooKeeper quorum：

• 本地 ZooKeeper运行在集群（cluster）级别，提供集群专属的配置管理与协调功能。每个 Pulsar 集群都需要一套独立的 ZooKeeper 集群。
• 配置存储（Configuration Store）运行在实例级别，为整个系统（跨越多个集群）提供配置管理。配置存储 quorum 可以由独立的机器集群提供，也可以与本地 ZooKeeper 共用同一批机器。

理解这一双层设计是部署 Pulsar 的第一步：本地 ZooKeeper 管"一个集群"，配置存储管"整个实例"。

部署本地 ZooKeeper（Local ZooKeeper）

部署一个 Pulsar 实例，需要为每一个 Pulsar 集群单独拉起一套本地 ZooKeeper 集群。

首先，将所有 ZooKeeper 服务器加入conf/zookeeper.conf文件中的 quorum 配置。为集群中的每个节点添加一行server.N配置，其中N是 ZooKeeper 节点的编号。以下是一个三节点集群的示例：

server.1=zk1.us-west.example.com:2888:3888 server.2=zk2.us-west.example.com:2888:3888 server.3=zk3.us-west.example.com:2888:3888

在每个主机上，还需要在各节点的myid文件中指定节点 ID。myid文件默认位于各服务器的data/zookeeper目录下（可通过dataDir参数修改该路径）。例如，在zk1.us-west.example.com这台 ZooKeeper 服务器上，可以这样设置myid：

$ mkdir -p data/zookeeper $ echo 1 > data/zookeeper/myid

在zk2.us-west.example.com上则执行echo 2 > data/zookeeper/myid，依此类推。ZooKeeper 官方文档的 Multi-server setup guide 对myid及多机部署有更详细的说明。

当每台服务器都已写入zookeeper.conf配置且具备正确的myid条目后，即可在所有主机上使用pulsar-daemonCLI 工具以后台方式（nohup）启动 ZooKeeper：

$ bin/pulsar-daemon start zookeeper

部署配置存储（Configuration Store）

上文配置并启动的 ZooKeeper 集群是本地ZooKeeper 集群，仅用于管理单个 Pulsar 集群。而一个完整的 Pulsar 实例除了本地集群外，还需要一个配置存储来处理实例级别的配置管理与协调任务。

• 如果部署的是单集群实例，则不需要为配置存储单独建立集群；
• 如果部署的是多集群实例，则应单独拉起一套 ZooKeeper 集群专门承担配置任务。

单集群 Pulsar 实例

如果 Pulsar 实例只包含一个集群，则可以将配置存储部署在与本地 ZooKeeper quorum 相同的机器上，但运行在不同的 TCP 端口。

部署单集群实例的 ZooKeeper 配置存储时，将本地 quorum 使用的相同 ZooKeeper 服务器添加到conf/global_zookeeper.conf配置文件中，方法与本地 ZooKeeper相同，但务必使用不同的端口（2181 是 ZooKeeper 的默认端口）。以下示例为三节点 ZooKeeper 集群使用 2184 端口：

clientPort=2184 server.1=zk1.us-west.example.com:2185:2186 server.2=zk2.us-west.example.com:2185:2186 server.3=zk3.us-west.example.com:2185:2186

与之前一样，为每台服务器在data/global-zookeeper/myid创建myid文件。仓库中真实的conf/global_zookeeper.conf默认即使用clientPort=2184与dataDir=data/global-zookeeper，与文档示例完全一致，可直接作为单集群配置存储的基础模板。

多集群 Pulsar 实例

当部署一个跨越不同地理区域的全球 Pulsar 实例时，配置存储充当高可用、强一致的元数据存储，必须能够容忍跨越整个区域的故障与网络分区。

关键点在于：确保 ZK quorum 成员至少分布在 3 个区域，且其他区域的节点以 observer 身份运行。

考虑到配置存储服务器上的预期负载非常低，可以与本地 ZooKeeper quorum 共用同一批主机。

例如，假设一个 Pulsar 实例包含以下集群：us-west、us-east、us-central、eu-central、ap-south，且每个集群拥有自己的本地 ZK 服务器，命名为：

zk[1-3].${CLUSTER}.example.com

在此场景下，从少数几个集群中挑选 quorum 参与节点，让其余节点全部作为 ZK observer。例如，要组成一个 7 服务器的 quorum，可以从us-west选 3 台、从us-central选 2 台、从us-east选 2 台。

这样可以保证即使其中某个区域不可达，配置存储的写入依然可用。所有服务器上的 ZK 配置如下：

clientPort=2184 server.1=zk1.us-west.example.com:2185:2186 server.2=zk2.us-west.example.com:2185:2186 server.3=zk3.us-west.example.com:2185:2186 server.4=zk1.us-central.example.com:2185:2186 server.5=zk2.us-central.example.com:2185:2186 server.6=zk3.us-central.example.com:2185:2186:observer server.7=zk1.us-east.example.com:2185:2186 server.8=zk2.us-east.example.com:2185:2186 server.9=zk3.us-east.example.com:2185:2186:observer server.10=zk1.eu-central.example.com:2185:2186:observer server.11=zk2.eu-central.example.com:2185:2186:observer server.12=zk3.eu-central.example.com:2185:2186:observer server.13=zk1.ap-south.example.com:2185:2186:observer server.14=zk2.ap-south.example.com:2185:2186:observer server.15=zk3.ap-south.example.com:2185:2186:observer

另外，ZK observer 节点还需要在配置中加入：

peerType=observer

启动配置存储服务

配置存储配置就绪后，使用pulsar-daemon启动服务：

$ bin/pulsar-daemon start configuration-store

ZooKeeper 配置参数

在 Pulsar 中，ZooKeeper 配置由安装目录conf下的两个独立配置文件管理：本地 ZooKeeper 使用conf/zookeeper.conf，配置存储使用conf/global-zookeeper.conf。

本地 ZooKeeper 参数

本地 ZooKeeper 的配置由conf/zookeeper.conf文件管理，可用参数如下：

仓库中真实的conf/zookeeper.conf文件还包含若干补充参数，可作为深入调优的参考：admin.enableServer=true、admin.serverPort=9990、forceSync=yes（要求事务日志在完成更新处理前同步到介质，官方不建议在生产环境关闭）、Quorum TLS 相关参数sslQuorum与portUnification（默认为 false，可按文档注释进行无停机 TLS 迁移），以及 Prometheus 指标导出配置metricsProvider.className=org.apache.zookeeper.metrics.prometheus.PrometheusMetricsProvider、metricsProvider.httpPort=8000。

配置存储参数

配置存储的配置由conf/global-zookeeper.conf文件管理，参数含义与本地 ZooKeeper 一致，核心区别在于默认端口与数据目录：仓库中该文件的默认配置为clientPort=2184、dataDir=data/global-zookeeper、admin.serverPort=9991，其余如tickTime、initLimit、syncLimit、autopurge.*等与本地 ZooKeeper 相同。

BookKeeper：Pulsar 的持久化存储层

BookKeeper 负责 Pulsar 中所有持久化消息的存储。它是一个分布式 write-ahead log（WAL）系统，保证独立消息日志（ledger）的读取一致性。BookKeeper 中的单个服务器也被称为bookie。

关于 Pulsar 消息持久化、保留（retention）与过期（expiry）管理的完整指南，参见此 cookbook。

部署 BookKeeper

BookKeeper 为 Pulsar 提供持久化消息存储。每个 Pulsar broker 需要拥有自己的 bookie 集群，且该 BookKeeper 集群与 Pulsar 集群共享本地 ZooKeeper quorum。

配置 bookie

BookKeeper bookie 使用conf/bookkeeper.conf配置文件进行配置。配置每个 bookie 时最重要的一点是确保zkServers参数被设置为 Pulsar 集群本地 ZooKeeper 的连接字符串。仓库中该参数的默认值为zkServers=localhost:2181，在真实多机部署中必须替换为本地 ZK quorum 的完整地址列表。

启动 bookie

启动 bookie 有两种方式：前台运行或后台守护进程运行。

在前台启动 bookie，使用bookkeeperCLI 工具：

$ bin/bookkeeper bookie

在后台启动 bookie，使用pulsar-daemonCLI 工具：

$ bin/pulsar-daemon start bookie

可以使用 BookKeeper shell 的bookiesanity命令验证 bookie 是否正常工作：

$ bin/bookkeeper shell bookiesanity

该命令会在本地 bookie 上创建一个新 ledger，写入少量条目，读回并最终删除该 ledger，从而完整验证 bookie 的读写通路。

Bookie 硬件考量

Bookie 主机负责将消息数据写入磁盘。为获得最佳性能，bookie 需要合适的硬件配置。bookie 硬件容量有两个关键维度：

• 磁盘 I/O 容量（读/写）
• 存储容量

写入 bookie 的消息条目在向 Pulsar broker 返回确认之前，必须同步到磁盘。为保证低写延迟，BookKeeper 设计为使用多个存储设备：

• journal（日志盘）用于保证持久性。对于顺序写入，bookie 主机上快速 fsync 操作至关重要。通常，小型快速的固态硬盘（SSD）即可满足要求；或者使用配备 RAID 控制器与电池供电写缓存（battery-backed write cache）的机械硬盘（HDD）。两种方案都可以达到约 0.4 ms 的 fsync 延迟。
• ledger 存储设备用于存储数据，直到所有消费者确认消息。写入在后台进行，因此写 I/O 不是大问题；读取在大多数情况下顺序进行，只有在消费者追赶（drain）时才会清空积压。要存储大量数据，典型配置是多个带 RAID 控制器的 HDD。

仓库中的conf/bookkeeper.conf提供了与上述硬件模型对应的完整配置项：journal 相关参数包括journalDirectory=data/bookkeeper/journal、journalMaxSizeMB=2048、journalMaxBackups=5、journalSyncData=true与journalWriteData=true（两者默认开启以保证持久性，关闭虽可提升性能但存在断电丢数据风险）；ledger 存储相关参数包括ledgerDirectories=data/bookkeeper/ledgers、ledgerStorageClass=org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage、logSizeLimit=1073741824，以及磁盘水位参数diskUsageThreshold=0.95、diskUsageWarnThreshold=0.95、diskUsageLwmThreshold=0.90和readOnlyModeEnabled=true（所有 ledger 盘写满时转入只读模式而非停机）。

最小化的 BookKeeper 配置

conf/bookkeeper.conf文件包含全部 bookie 可配置参数。生产部署时必须修改的最小配置项如下：

# 修改为指向 journal 磁盘挂载点 journalDirectory=data/bookkeeper/journal # 指向 ledger 存储磁盘挂载点 ledgerDirectories=data/bookkeeper/ledgers # 指向本地 ZK quorum zkServers=zk1.example.com:2181,zk2.example.com:2181,zk3.example.com:2181

如需修改 BookKeeper 使用的 ZooKeeper 根路径，应使用zkLedgersRootPath=/MY-PREFIX/ledgers代替zkServers=localhost:2181/MY-PREFIX（仓库中conf/bookkeeper.conf的默认值为zkLedgersRootPath=/ledgers）。

更多 BookKeeper 的详细资料，请参阅官方 BookKeeper 文档。

BookKeeper 持久化策略（Persistence Policies）

在 Pulsar 中，可以在命名空间（namespace）级别设置持久化策略，决定 BookKeeper 如何处理消息的持久化存储。策略决定四个方面：

• 每个 ledger 条目需要等待的 ack 数（保证的副本数）
• 一个 topic 使用的 bookie 数量（ensemble）
• 每个 ledger 条目执行多少次写入（write quorum）
• mark-delete 操作的限制速率（throttling rate）

在源码层面，这四个维度被封装为PersistencePolicies类，其字段为bookkeeperEnsemble、bookkeeperWriteQuorum、bookkeeperAckQuorum与managedLedgerMaxMarkDeleteRate；无参构造函数的默认值为(2, 2, 2, 0.0)，即默认 2 个副本、2 个写入 quorum、2 个 ack quorum、mark-delete 不限速。

设置持久化策略

持久化策略在命名空间级别设置。

pulsar-admin

使用set-persistence子命令，并指定命名空间以及要应用的任意策略。可用参数如下：

注：表格中的 0 为原文档中的占位描述。在仓库源码CmdNamespaces.java中，set-persistence命令的实际默认值为bookkeeperEnsemble=2、bookkeeperWriteQuorum=2、bookkeeperAckQuorum=2、managedLedgerMaxMarkDeleteRate=0，且运行时会校验三个 quorum 参数必须大于 0、mark-delete 速率不得小于 0。

示例

$ pulsar-admin namespaces set-persistence my-tenant/my-ns \ --bookkeeper-ack-quorum 3 \ --bookkeeper-ensemble 2

REST API

通过管理接口POST /admin/v2/namespaces/:tenant/:namespace/persistence调用setPersistence操作。

Java

int bkEnsemble = 2; int bkQuorum = 3; int bkAckQuorum = 2; double markDeleteRate = 0.7; PersistencePolicies policies = new PersistencePolicies(ensemble, quorum, ackQuorum, markDeleteRate); admin.namespaces().setPersistence(namespace, policies);

原文档示例中的变量名存在笔误（bkEnsemble等与构造参数不一致），实际使用时构造参数顺序与PersistencePolicies构造器一致即可：PersistencePolicies(ensemble, writeQuorum, ackQuorum, markDeleteRate)。

查看持久化策略

可以查看当前应用于某个命名空间的持久化策略。

pulsar-admin

使用get-persistence子命令并指定命名空间。

示例

$ pulsar-admin namespaces get-persistence my-tenant/my-ns { "bookkeeperEnsemble": 1, "bookkeeperWriteQuorum": 1, "bookkeeperAckQuorum", 1, "managedLedgerMaxMarkDeleteRate": 0 }

注：该输出示例中的 JSON 存在笔误（"bookkeeperAckQuorum", 1缺少冒号），实际返回格式与PersistencePolicies的四个字段一一对应，均为"字段名": 值的标准 JSON。

REST API

通过管理接口GET /admin/v2/namespaces/:tenant/:namespace/persistence调用getPersistence操作。

Java

PersistencePolicies policies = admin.namespaces().getPersistence(namespace);

另外，源码中还提供了remove-persistence子命令（对应admin.namespaces().removePersistence(namespace)），可将命名空间的持久化策略恢复为默认值（见CmdNamespaces.java）。

理解 Pulsar 如何使用 ZooKeeper 与 BookKeeper

下图展示了 ZooKeeper 与 BookKeeper 在一个 Pulsar 集群中的角色分工：

每个 Pulsar 集群由一个或多个消息 broker 组成，每个 broker 依赖一组 bookie（ensemble）进行消息持久化。整体数据流为：

1. 写入路径：producer 将消息发送给 broker → broker 将条目写入 BookKeeper ledger（写入多个 bookie，达到 ack quorum 后确认）→ 数据在返回确认前已 fsync 到 journal 盘，保证持久性；
2. 协调路径：broker 发现、ledger 元数据、负载均衡、租约等协调工作均由 ZooKeeper 完成；其中本地 ZooKeeper 服务单个集群，配置存储在多集群场景下跨区域提供强一致的元数据服务；
3. 读取路径：consumer 从 broker 订阅消费，broker 从 ledger 存储设备按需读取消息，积压（backlog）清空后，条目随 GC 与自动清理被回收。

总结

Pulsar 的可靠性与可扩展性建立在 ZooKeeper 与 BookKeeper 的正确部署之上。实践中请记住以下要点：

• 本地 ZooKeeper 与配置存储是两套不同的 quorum：单集群可复用机器、换端口（2184）；多集群必须跨至少 3 个区域部署 quorum 参与者，其余节点设为peerType=observer；
• bookie 的最小配置三件套：journalDirectory（journal 盘）、ledgerDirectories（ledger 盘）与zkServers（指向本地 ZK quorum），并可用bin/bookkeeper shell bookiesanity验证读写通路；
• 持久化策略按命名空间设置：-e/-w/-a/-r四个参数分别控制 ensemble、write quorum、ack quorum 与 mark-delete 限速，源码默认值为(2, 2, 2, 0)，可通过pulsar-admin namespaces get-persistence随时核对当前生效值。

如需查阅完整的参数清单，可继续阅读本仓库中对应的配置参考文档，以及真实配置文件conf/zookeeper.conf、conf/global_zookeeper.conf与conf/bookkeeper.conf。

• 消息队列
• 后端
• 流处理