Ceph+K8s 中间件实战 | 第 19 天：K8s 部署 Apache Pulsar 集群（Ceph RBD + Bookkeeper 架构 + 运维）

第 19/28 天

Apache Pulsar 是 Apache 顶级项目，由 LinkedIn 从内部 Kappa 架构中孵化而来，凭借计算存储分离 + 分层存储 + 多租户隔离三大特性，成为新一代流批一体消息中间件的代表。本篇承接第 18 天 RabbitMQ 集群实战，继续在 Ceph + Kubernetes 底座上部署 Pulsar 集群：BookKeeper 用 Ceph RBD 承载持久化账本，Broker 无状态化支撑水平扩展，通过 StreamNative 官方 Helm Chart 一键拉起生产级拓扑，覆盖架构设计、部署、登录验证与日常巡检四大主线。

一、为什么 Pulsar 需要 Ceph RBD 持久化

Pulsar 是典型的「计算存储分离」架构，Producer 把消息先写入 Broker，Broker 再转发到 BookKeeper 集群，BookKeeper 才是真正的数据落盘层。BookKeeper 由多个 Bookie 节点组成，每个 Bookie 在本地磁盘上维护若干「账本」（ledger），通过 Multi-Quorum 协议保证副本一致性；在 K8s 场景里，Bookie 必须以 StatefulSet 部署且必须挂载持久卷，否则 Pod 一旦漂移，账本数据全部丢失。

Ceph RBD 恰好为 Bookie 提供了理想的存储底座：①每个 Bookie 独占一个 RBD 镜像（建议 500Gi～1Ti），承载 /pulsar/data/bookkeeper/ledgers 与 journal；②结合 Ceph CSI 的快照、克隆、扩容能力，支持账本备份与在线扩容；③配合本系列第 2 天创建的 ceph-rbd StorageClass，仅需在 values.yaml 中指定即可。

此外，Pulsar 还支持 Pulsar Tiered Storage（分层存储），可将冷数据从 BookKeeper 卸载到 Ceph Object Gateway（RGW），实现 PB 级归档，本篇会附带说明。

二、Pulsar 集群设计架构

2.1 组件拓扑

Pulsar 集群由五大角色组成，StreamNative Helm Chart 默认将它们拆成独立的 StatefulSet/Deployment：

组件	副本数	端口	存储	职责
ZooKeeper / Bookie（zkbookie 模式）	3	2181/3181/8000	ceph-rbd 20Gi × 3	分布式协调 + 账本落盘
Broker	3	6650/8080	无状态	消息收发、路由、鉴权
BookKeeper standalone（分离模式）	3	3181/8000	ceph-rbd 500Gi × 3	高吞吐账本存储
Web UI	1	80/9090	无状态	控制台与 REST API
Proxy / Proxy	2	8080	无状态	多租户接入、TLS 终结

生产环境推荐分离模式（Bookie 与 ZooKeeper 分开部署），开发环境可用 zkbookie 组合模式简化拓扑。

2.2 数据流向

💻 代码示例

Producer ──pub/sub──▶ Proxy/Broker:6650

│

▼ 路由（多租户 + Topic）

BookKeeper 集群

│ Quorum=3, Ack=2

▼

RBD 账本文件（Ledger）

│

▼ 可选 Tiered Storage

Ceph RGW / S3（冷数据）

▲

│ long-polling

Consumer ──sub──▶ Broker

关键链路：①Producer 通过 Broker 建立 TCP 长连接，写入消息后 Broker 转发到 BookKeeper，Bookie 集群写入 journal（顺序 IO）+ ledger（随机 IO）；②Consumer 通过 Broker 建立订阅，Broker 长轮询 BookKeeper 拉取数据；③所有写入以 quorum=3, ack=2 的多副本方式保证，任一 Bookie 宕机不影响写入。

2.3 高可用机制

Pulsar 的 HA 建立在三层之上：

ZooKeeper 层：3 副本 Raft 集群（或 Pulsar 3.x 引入的 ZK Metadata），保证 topic 分配与集群拓扑的一致性；
BookKeeper 层：每个 ledger 默认写 3 副本，读 quorum=3、ack=2，任一 Bookie 宕机自动重建 ledger；
Broker 层：Broker 无状态，Topic 所有权通过 ZooKeeper 分配，Broker 宕机后由 Leader Election 重新分配，Producer 通过 DNS/Proxy 重连即可。

关键容量规划：BookKeeper 500Gi × 3 = 1.5Ti Ceph RBD，按 3 副本计有效容量 500Gi，可承载约 500 亿条 1KB 消息（含副本冗余），Broker CPU/内存建议 ≥ 2C/4Gi/副本以支撑百万级 TPS。

三、部署实战（StreamNative Helm Chart）

StreamNative 是 Pulsar 商业化公司（Confluent 收购），官方维护的 pulsar-helm-chart 是当前社区最活跃、功能最全的部署方案，支持 standalone / 分离 / 多集群三种模式。

3.1 添加 Helm 仓库与查看版本

💻 代码示例

# 添加 StreamNative Chart 仓库（官方）

helm repo add streamnative https://pulsar-streamnative.github.io/helm-charts

helm repo update

# 查看 Pulsar Chart 版本

helm search repo streamnative/pulsar

# streamnative/pulsar 2.10.x 3.0.x Apache Pulsar Operator

# 拉取 Chart 到本地查看 values 默认值

helm pull streamnative/pulsar –version 2.10.2 –untar

cat pulsar/values.yaml | head -120

3.2 检查前置依赖

Pulsar 依赖 ZK 或 Etcd 做元数据，官方推荐 Pulsar 3.x 内置 ZK + RBAC。若已有独立 ZooKeeper，可通过 zookeeper.external.enabled=true 挂载外部集群。

💻 代码示例

# 若本集群无独立 ZK，StreamNative chart 自带 ZK，无需额外安装

# 若已有 K8s ZooKeeper（如第 10 天部署的），可用外部模式

helm list -n middleware | grep zookeeper

# 准备专用命名空间

kubectl create ns pulsar

kubectl create ns pulsar-data

3.3 编写 values.yaml（重点：Ceph RBD 持久化）

StreamNative Chart 结构复杂，建议将 pulsar/values.yaml 分层管理。核心是 zookeeper、bookkeeper、broker、proxy 四块的持久化配置。以下是分离模式的生产配置：

💻 代码示例

# pulsar-values.yaml

# Apache Pulsar 集群配置（Ceph RBD 持久化 + 分离模式）

# 全局镜像与命名空间

global:

kubernetesProvider: kubernetes

pulsarRbac: true

# — ZooKeeper 集群（3 副本）—

zookeeper:

enabled: true

replicaCount: 3

resources:

requests:

cpu: 500m

memory: 2Gi

limits:

cpu: "2"

memory: 4Gi

persistence:

enabled: true

storageClassName: ceph-rbd # 复用第 2 天创建的 StorageClass

size: 20Gi

accessMode: ReadWriteOnce

reclaimPolicy: Retain

# 生产建议启用 TLS

service.tls.enabled: false

# — BookKeeper 集群（3 副本，核心）—

bookkeeper:

enabled: true

replicaCount: 3

resources:

requests:

cpu: "2"

memory: 4Gi

limits:

cpu: "4"

memory: 8Gi

persistence:

enabled: true

storageClassName: ceph-rbd

size: 500Gi # BookKeeper 是数据主力

accessMode: ReadWriteOnce

reclaimPolicy: Retain

# BookKeeper 关键 JVM 参数

conf: |

journalAdvertiseAddress=%HOSTNAME%.%FULLNAME%.%NAMESPACE%.svc.cluster.local

httpServerEnabled=true

httpServerPort=8000

allowLoopback=true

# 账本写入多副本策略

autoRecovery:

enabled: true

zkLedgers: false

zkEnsemble: zookeeper

intervalInMsec: 300000

# — Broker 集群（3 副本，无状态）—

broker:

replicaCount: 3

resources:

requests:

cpu: "2"

memory: 4Gi

limits:

cpu: "4"

memory: 8Gi

persistence:

enabled: false # Broker 无状态

# BookKeeper 副本策略：写 3 副本，读 2 副本

writerConfig:

ensembleSize: 3

writeQuorumSize: 3

ackQuorumSize: 2

readerConfig:

readQuorumSize: 3

# Broker 关键 JVM 参数

conf: |

zookeeperServers=zookeeper.%NAMESPACE%.svc.cluster.local:2181

brokerServicePort=6650

webServicePort=8080

numIOThreads=16

numAcceptorThreads=8

# Pulsar 3.x 多租户

tenantExpirationCheckIntervalSec=1800

systemTopicEnabled=true

# — Proxy / Proxy-Native（对外入口）—

proxy:

enabled: true

replicaCount: 2

serviceType: ClusterIP

ingress:

enabled: true

ingressClassName: nginx

hostname: pulsar.example.com

tls:

enabled: true

service.http.ports:

http: 80

https: 443

service.ws.ports:

ws: 9090

wss: 9443

# — Web UI —

web:

enabled: true

replicaCount: 1

# — Pulsar 分层存储（可选，冷数据卸载到 Ceph RGW）—

# tieredStorage:

# enabled: true

# storageDriverName: aws

# config:

# dataConnectorPath: s3://pulsar-archive

# managedLedgerForceManagedLedgerOffloadEnabled: true

3.4 helm install 部署

💻 代码示例

# 部署 Pulsar 集群（等待完成，最长 15 分钟）

helm install pulsar streamnative/pulsar

–namespace pulsar

–create-namespace

-f pulsar-values.yaml

–version 2.10.2

–set image.registry=registry.cn-hangzhou.aliyuncs.com

–set image.repository=streamnative

–set image.tag=3.0.3

–wait

–timeout 15m0s

–debug

# 若使用国内镜像仓库加速，可覆盖 image.registry 与 repository

# 部署过程会依次创建：ZooKeeper → BookKeeper → Broker → Proxy → Web

helm status pulsar -n pulsar

helm get manifest pulsar -n pulsar | grep -E "kind: (StatefulSet|Deployment)" | sort -u

3.5 验证 Pod / Service / PVC 状态

💻 代码示例

# 全部 Pod 应 Running（ZooKeeper 3 + BookKeeper 3 + Broker 3 + Proxy 2 + Web 1 = 12 个）

kubectl get pods -n pulsar -o wide

# PVC 绑定情况（ZooKeeper + BookKeeper 应有 6 个 ceph-rbd PVC）

kubectl get pvc -n pulsar -o wide

# NAME STATUS VOLUME CAPACITY STORAGECLASS

# data-zookeeper-0 Bound pv-zookeeper-0 20Gi ceph-rbd

# data-zookeeper-1 Bound pv-zookeeper-1 20Gi ceph-rbd

# data-bookkeeper-0 Bound pv-bookkeeper-0 500Gi ceph-rbd

# data-bookkeeper-1 Bound pv-bookkeeper-1 500Gi ceph-rbd

# data-bookkeeper-2 Bound pv-bookkeeper-2 500Gi ceph-rbd

# Service 列表（Headless + ClusterIP + NodePort）

# 检查 Broker 是否已注册到 ZooKeeper

kubectl exec -n pulsar zookeeper-0 — /pulsar/bin/pulsar-zookeeper-shell

localhost:2181 ls /admin/clusterdata/service-config/broker-registry/localhost

四、登录验证与集群状态检查

4.1 通过 kubectl exec 进入 Broker 使用 pulsar-admin CLI

💻 代码示例

# 进入 broker-0 节点，pulsar-admin 是官方 CLI 工具

kubectl exec -it broker-0 -n pulsar — bash

# 检查集群元数据

pulsar-admin clusters list

pulsar-admin clusters get pulsar

# 列出所有租户

pulsar-admin tenants list

# 列出 public/default namespace 下所有 topic

pulsar-admin topics list public/default

# 查看 topic 详情（含 BookKeeper ledger 分布）

pulsar-admin topics stats public/default/my-topic

# 列出订阅者与消费者

pulsar-admin subscriptions list public/default/my-topic

4.2 通过 pulsar-client CLI 生产/消费消息

💻 代码示例

# 集群内运行（Pod 内 pulsar-client.sh）

kubectl exec -it broker-0 -n pulsar —

/pulsar/bin/pulsar-client produce

pulsar://broker-0.broker.pulsar.svc.cluster.local:6650/public/default/test

–message "Hello Pulsar from K8s!"

kubectl exec -it broker-0 -n pulsar —

/pulsar/bin/pulsar-client consume

pulsar://broker-0.broker.pulsar.svc.cluster.local:6650/public/default/test

-n 1 –subscription sub-test –subscription-type Exclusive

# 通过 Proxy 从外部访问（先 port-forward）

kubectl port-forward -n pulsar svc/proxy 6650:6650 &

pulsar-admin –admin-url http://localhost:8080 –broker-url pulsar://localhost:6650 tenants list

4.3 Web UI 与 REST API

💻 代码示例

# 暴露 Web UI（默认 80 端口）

kubectl port-forward -n pulsar svc/web 80:80 &

# 浏览器访问 http://localhost → Pulsar 管理台

# REST API 调用

curl -s http://localhost:80/admin/v2/clusters | python3 -m json.tool

curl -s http://localhost:80/admin/v2/tenants | python3 -m json.tool

curl -s http://localhost:80/admin/v2/clusters/pulsar/broker-stats | python3 -m json.tool

五、日常巡检基础命令

5.1 健康巡检（每日执行）

💻 代码示例

# ① 各组件 Pod 存活情况

kubectl get pods -n pulsar

-o custom-columns=NAME:.metadata.name,READY:.status.containerStatuses[*].ready,RESTARTS:.status.containerStatuses[*].restartCount,AGE:.status.startTime

# ② ZooKeeper 集群状态（应有 1 个 leader + 2 follower）

kubectl exec -n pulsar zookeeper-0 — /pulsar/bin/pulsar-zookeeper-shell localhost:2181 ruok

# 返回 "imok" 表示健康

# ③ BookKeeper 元数据状态

kubectl exec -n pulsar broker-0 — /pulsar/bin/bookkeeper shell listbookies –all

# 输出各 Bookie 状态（AVAILABLE / DECOMMISSIONED）

# ④ Broker 心跳检测

kubectl exec -n pulsar broker-0 — /pulsar/bin/pulsar-admin broker healthcheck

# 返回 HTTP 200 OK

5.2 性能监控（对接 Prometheus）

💻 代码示例

# Pulsar 自带 metrics endpoint，Broker 8080/9090 暴露 Prometheus 格式

kubectl exec -n pulsar broker-0 — curl -s http://localhost:8080/metrics |

grep -E "^pulsar_num_(messages|publish|subs)" | head -20

# Grafana 关键查询示例：

# 消息生产速率（过去 5 分钟）

rate(pulsar_num_msg_published[5m])

# 消息消费速率

rate(pulsar_num_subscription_msg_redelivered[5m])

# Bookie 存储水位

bookie_ledger_entries_size / 1024 / 1024

5.3 容量与日志巡检

💻 代码示例

# ① BookKeeper 账本数量与大小

kubectl exec -n pulsar bookkeeper-0 — /pulsar/bin/bookkeeper shell listledgers

# ② 磁盘使用率（Bookie 单节点）

kubectl exec -n pulsar bookkeeper-0 — df -h /pulsar/data/bookkeeper

# ③ Ceph 后端占用

ceph df | grep pulsar

rbd du ceph/pulsar-data-bookkeeper-0

# ④ 错误日志巡检（近 200 行 ERROR）

kubectl logs -n pulsar broker-0 –tail=200 | grep -iE "error|exception|refused|timeout"

kubectl logs -n pulsar bookkeeper-0 –tail=200 | grep -iE "error|exception|refused|timeout"

# ⑤ Bookie 挂账列表（识别 lag 落后的账本）

kubectl exec -n pulsar broker-0 — /pulsar/bin/bookkeeper shell listunderreplicatedledgers

5.4 备份与恢复

💻 代码示例

# ① BookKeeper 账本备份（官方 bookkeeper backup）

kubectl exec -n pulsar bookkeeper-0 — /pulsar/bin/bookkeeper shell backup

-m pulsar://broker-0.broker.pulsar.svc.cluster.local:6650

-b 3 -w 3 -a 2

backup -t pulsar://broker-0.broker.pulsar.svc.cluster.local:6650

-c /tmp/backup-ledger.txt

# ② ZooKeeper 全量快照

kubectl exec -n pulsar zookeeper-0 — /pulsar/bin/pulsar-zookeeper-shell

localhost:2181 dump /tmp/zk-dump-$(date +%s).json

kubectl cp zookeeper-0:/tmp/zk-dump-$(date +%s).json ./zk-backup.json -n pulsar

# ③ Ceph 快照（BookKeeper 卷）

cat > /tmp/pulsar-snapshot.yaml <<EOF

apiVersion: snapshot.storage.k8s.io/v1

kind: VolumeSnapshot

metadata:

name: bookkeeper-0-snap-$(date +%Y%m%d)

namespace: pulsar

spec:

source:

persistentVolumeClaimName: data-bookkeeper-0

volumeSnapshotClassName: ceph-rbd-snapshotclass

EOF

kubectl apply -f /tmp/pulsar-snapshot.yaml

kubectl get volumesnapshot -n pulsar

六、常见问题 FAQ

Q1：BookKeeper 启动报 Cannot connect to Zookeeper 怎么办？
A：①确认 ZooKeeper Pod 已就绪：kubectl get pods -n pulsar | grep zookeeper；②BookKeeper 通过 Service DNS 访问 ZK：kubectl exec bookkeeper-0 -- nslookup zookeeper.pulsar.svc.cluster.local；③检查 ZK 端口 2181 是否放行：kubectl exec bookkeeper-0 -- nc -zv zookeeper.pulsar.svc.cluster.local 2181；④若使用外部 ZK，检查 zookeeper.external.servers 配置是否正确。

Q2：为什么 BookKeeper 的 Ceph RBD PVC 长时间 Pending？
A：①BookKeeper 500Gi 会一次性申请 3 个 PVC，检查 Ceph OSD 剩余容量：ceph df；②确认 accessMode: ReadWriteOnce（RBD 不支持 RWX）；③检查 Ceph CSI 是否就绪：kubectl get pods -n ceph | grep csi-rbd-plugin；④kubectl describe pvc data-bookkeeper-0 -n pulsar | grep -i event 查看具体错误，可能是 Ceph mon 不可达或 RBD pool 未创建。

Q3：如何验证 Pulsar 集群是否达到 SLA？
A：①通过 pulsar-admin broker stats 查看每个 Broker 的 throughput、connections、subscriptions；②注入压测：pulsar-perf produce pulsar://broker/pulsar/test -r 10000 -s 1024 观察 P99 延迟；③模拟 Bookie 宕机：kubectl delete pod bookkeeper-0 -n pulsar 观察 ledger 是否自动恢复；④通过 Web UI 查看集群拓扑与 topic 分布。

七、总结

本篇在 Ceph + K8s 底座上完成 Apache Pulsar 集群部署，通过 StreamNative 官方 Helm Chart 快速落地，关键交付点包括：

架构设计：ZooKeeper + BookKeeper + Broker + Proxy 四层分离，计算存储完全解耦
持久化方案：BookKeeper 500Gi × 3 副本 Ceph RBD，Broker 无状态可弹性伸缩
多副本策略：ensembleSize=3, writeQuorum=3, ackQuorum=2，任一 Bookie 宕机数据不丢失
登录验证：pulsar-admin CLI、pulsar-client CLI、Web UI、REST API 四种方式全覆盖
日常巡检：健康检查、性能监控、容量备份四类命令模板，可直接纳入运维手册
故障排查：ZK 连接失败、PVC Pending、SLA 验证三大常见坑

Pulsar 与 Kafka 的核心差异在于架构（Kafka 计算存储耦合 vs Pulsar 分离），选型参考第 12 天《K8s 部署 Kafka 集群》。

八、下期预告

第 20 天将部署 OpenSearch 集群。OpenSearch 是 AWS 从 Elasticsearch 7.10 fork 的开源替代品，具备更好的 AGPL 许可、内存优化插件与机器学习能力。届时将讲解 OpenSearch 的分片复制架构、Ceph RBD 持久化、Helm Chart 部署与安全基线配置。敬请期待！

九、系列目录

文章版权归作者所有，未经允许请勿转载。

THE END