Apache Pulsar 消息队列集群部署与运维实战

一、Pulsar 概述

1.1 什么是 Apache Pulsar

Apache Pulsar 是 Apache 软件基金会顶级项目，是一款云原生分布式消息流平台。它结合了消息队列和流处理的特点，支持 pub-sub 和 queue 两种消费模式。

1.2 Pulsar 核心特性

特性 说明
多租户 原生支持多租户，租户/命名空间/主题三层模型
持久化 消息持久化到 Apache BookKeeper，保证数据可靠性
高可用 无单点故障，支持跨地域复制
低延迟 发布 - 订阅延迟低至毫秒级
弹性扩展 计算（Broker）与存储（BookKeeper）分离，独立扩展
多协议 支持 Pulsar、Kafka、AMQP、MQTT、STOMP 等协议
消息追溯 支持消息回溯消费，保留策略灵活配置

1.3 Pulsar vs Kafka 对比

对比项 Apache Pulsar Apache Kafka
架构 计算存储分离 计算存储一体
分区扩展 自动分片，无需数据迁移 需手动 rebalance，数据迁移
消息追溯 支持任意位置回溯 依赖日志保留
多租户 原生支持 需额外配置
地域复制 内置支持 需 MirrorMaker
存储引擎 BookKeeper（分段） 文件系统日志
协议支持 多协议原生支持 主要 Kafka 协议

1.4 Pulsar 架构组件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
┌─────────────────────────────────────────────────────────────────┐
│                         Pulsar 架构                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────┐     ┌─────────────┐     ┌─────────────┐       │
│  │   Producer │     │   Producer │     │   Producer │       │
│  └──────┬──────┘     └──────┬──────┘     └──────┬──────┘       │
│         │                   │                   │               │
│         └───────────────────┼───────────────────┘               │
│                             │                                   │
│                    ┌────────▼────────┐                          │
│                    │  Load Balancer  │                          │
│                    └────────┬────────┘                          │
│                             │                                   │
│         ┌───────────────────┼───────────────────┐               │
│         │                   │                   │               │
│  ┌──────▼──────┐     ┌──────▼──────┐     ┌──────▼──────┐       │
│  │   Broker    │     │   Broker    │     │   Broker    │       │
│  │   (Stateless)│    │   (Stateless)│    │   (Stateless)│      │
│  └──────┬──────┘     └──────┬──────┘     └──────┬──────┘       │
│         │                   │                   │               │
│         └───────────────────┼───────────────────┘               │
│                             │                                   │
│                    ┌────────▼────────┐                          │
│                    │  ZooKeeper      │                          │
│                    │  (Metadata)     │                          │
│                    └────────┬────────┘                          │
│                             │                                   │
│         ┌───────────────────┼───────────────────┐               │
│         │                   │                   │               │
│  ┌──────▼──────┐     ┌──────▼──────┘     ┌──────▼──────┐       │
│  │ BookKeeper  │     │ BookKeeper  │     │ BookKeeper  │       │
│  │  Bookie 1   │     │  Bookie 2   │     │  Bookie 3   │       │
│  └─────────────┘     └─────────────┘     └─────────────┘       │
│         │                   │                   │               │
│  ┌──────▼──────┐     ┌──────▼──────┐     ┌──────▼──────┐       │
│  │  Ledger 1   │     │  Ledger 2   │     │  Ledger 3   │       │
│  └─────────────┘     └─────────────┘     └─────────────┘       │
│                                                                  │
│  ┌─────────────┐     ┌─────────────┐     ┌─────────────┐       │
│  │  Consumer   │     │  Consumer   │     │  Consumer   │       │
│  └─────────────┘     └─────────────┘     └─────────────┘       │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

核心组件：
- Broker: 无状态服务，处理消息路由和负载均衡
- BookKeeper: 分布式账本存储，由多个 Bookie 组成
- ZooKeeper: 存储元数据和配置信息
- Producer: 消息生产者
- Consumer: 消息消费者

二、环境准备

2.1 硬件要求

组件 最小配置 推荐配置 生产配置
Broker 2 核 4GB 4 核 8GB 8 核 16GB+
Bookie 2 核 4GB 4 核 8GB 8 核 16GB+
ZooKeeper 1 核 2GB 2 核 4GB 4 核 8GB
磁盘 (Bookie) 50GB SSD 200GB SSD 1TB+ NVMe
网络 1Gbps 1Gbps 10Gbps

2.2 软件要求

  • 操作系统：Linux（CentOS 7+/Ubuntu 18.04+）
  • Java：OpenJDK 11 或 17
  • Python：Python 3.7+（用于 CLI 工具）
  • ZooKeeper：3.6+（可内置或独立部署）

2.3 系统优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# /etc/security/limits.conf
pulsar soft nofile 1048576
pulsar hard nofile 1048576
pulsar soft nproc 65536
pulsar hard nproc 65536

# /etc/sysctl.conf
vm.max_map_count=262144
vm.swappiness=1
net.core.somaxconn=65536
net.ipv4.tcp_max_syn_backlog=65536
fs.file-max=2097152

# 应用配置
sysctl -p
ulimit -n 1048576

三、集群部署

3.1 部署规划

假设部署 3 节点集群：

节点 主机名 IP 角色
Node1 pulsar-node1 192.168.1.101 ZK+Bookie+Broker
Node2 pulsar-node2 192.168.1.102 ZK+Bookie+Broker
Node3 pulsar-node3 192.168.1.103 ZK+Bookie+Broker

3.2 安装 Pulsar

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 所有节点执行

# 1. 下载安装包
cd /opt
wget https://archive.apache.org/dist/pulsar/pulsar-3.3.0/apache-pulsar-3.3.0-bin.tar.gz

# 2. 解压
tar -xzf apache-pulsar-3.3.0-bin.tar.gz
ln -s apache-pulsar-3.3.0 pulsar

# 3. 创建 pulsar 用户
useradd -m -s /bin/bash pulsar
chown -R pulsar:pulsar /opt/pulsar

# 4. 创建数据目录
mkdir -p /data/pulsar/{bookkeeper,zookeeper}
chown -R pulsar:pulsar /data/pulsar

3.3 配置 ZooKeeper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# 节点 1 执行：初始化 ZooKeeper
cd /opt/pulsar
bin/pulsar initialize-cluster-metadata \
  --cluster pulsar-cluster \
  --zookeeper pulsar-node1:2181 \
  --configuration-store pulsar-node1:2181 \
  --web-service-url http://pulsar-node1:8080 \
  --web-service-url-tls https://pulsar-node1:8443 \
  --broker-service-url pulsar://pulsar-node1:6650 \
  --broker-service-url-tls pulsar+ssl://pulsar-node1:6651

# 所有节点配置 zookeeper.conf
cat > conf/zookeeper.conf << EOF
dataDir=/data/pulsar/zookeeper
clientPort=2181
initLimit=10
syncLimit=5
tickTime=2000
admin.enableServer=false
admin.serverPort=9990
4lw.commands.whitelist=ruok,stat,mntr,conf,envi
EOF

# 所有节点创建 myid 文件
# Node1
echo "1" > /data/pulsar/zookeeper/myid
# Node2
echo "2" > /data/pulsar/zookeeper/myid
# Node3
echo "3" > /data/pulsar/zookeeper/myid

# 所有节点配置 zookeeper 集群
cat >> conf/zookeeper.conf << EOF
server.1=pulsar-node1:2888:3888
server.2=pulsar-node2:2888:3888
server.3=pulsar-node3:2888:3888
EOF

3.4 配置 BookKeeper

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 所有节点配置 bookkeeper.conf
cat > conf/bookkeeper.conf << EOF
# 基础配置
bookiePort=3181
journalDirectory=/data/pulsar/bookkeeper/journal
ledgerDirectories=/data/pulsar/bookkeeper/ledgers

# 集群配置
zkServers=pulsar-node1:2181,pulsar-node2:2181,pulsar-node3:2181
zkLedgersRootPath=/ledgers

# 网络配置
advertisedAddress=pulsar-node1  # 各节点修改为对应主机名
allowLoopback=false

# 性能优化
journalMaxSizeMB=4096
gcWaitTime=300000
nettyMaxFrameSizeBytes=104857600

# 存储优化
diskUsageThreshold=0.90
diskUsageWarnThreshold=0.85
minorCompactionThreshold=0.1
majorCompactionThreshold=0.5
EOF

# 格式化 BookKeeper（仅执行一次，在节点 1）
bin/bookkeeper shell metaformat -nonInteractive

3.5 配置 Broker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 所有节点配置 standalone.conf 或 broker.conf
cat > conf/standalone.conf << EOF
# 集群配置
clusterName=pulsar-cluster
zookeeperServers=pulsar-node1:2181,pulsar-node2:2181,pulsar-node3:2181
configurationStoreServers=pulsar-node1:2181,pulsar-node2:2181,pulsar-node3:2181

# 网络配置
advertisedAddress=pulsar-node1  # 各节点修改
bindAddress=0.0.0.0
webServicePort=8080
brokerServicePort=6650

# BookKeeper 配置
bookkeeperClientHealthCheckEnabled=true
bookkeeperClientHealthCheckErrorThresholdPerBookie=5
bookkeeperClientSpeculativeReadTimeoutInMillis=3000

# 消息配置
defaultRetentionTimeInMinutes=10080
defaultRetentionSizeInMB=10240
defaultBacklogQuotaCheckEnabled=true

# 性能优化
numHttpServerThreads=16
brokerShutdownTimeoutMs=30000
EOF

3.6 启动服务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
# 创建 systemd 服务文件

# /etc/systemd/system/pulsar-zookeeper.service
cat > /etc/systemd/system/pulsar-zookeeper.service << EOF
[Unit]
Description=Apache Pulsar ZooKeeper
After=network.target

[Service]
Type=forking
User=pulsar
Group=pulsar
Environment=JAVA_HOME=/usr/lib/jvm/java-11-openjdk
Environment=PULSAR_HOME=/opt/pulsar
Environment=BOOKIE_MEM="-Xms2g -Xmx2g"
Environment=ZOOKEEPER_MEM="-Xms1g -Xmx1g"
ExecStart=/opt/pulsar/bin/pulsar-daemon start zookeeper
ExecStop=/opt/pulsar/bin/pulsar-daemon stop zookeeper
Restart=on-failure
LimitNOFILE=1048576

[Install]
WantedBy=multi-user.target
EOF

# /etc/systemd/system/pulsar-bookkeeper.service
cat > /etc/systemd/system/pulsar-bookkeeper.service << EOF
[Unit]
Description=Apache Pulsar BookKeeper
After=pulsar-zookeeper.service
Requires=pulsar-zookeeper.service

[Service]
Type=forking
User=pulsar
Group=pulsar
Environment=JAVA_HOME=/usr/lib/jvm/java-11-openjdk
Environment=PULSAR_HOME=/opt/pulsar
Environment=BOOKIE_MEM="-Xms4g -Xmx4g -XX:MaxDirectMemorySize=4g"
ExecStart=/opt/pulsar/bin/pulsar-daemon start bookkeeper
ExecStop=/opt/pulsar/bin/pulsar-daemon stop bookkeeper
Restart=on-failure
LimitNOFILE=1048576

[Install]
WantedBy=multi-user.target
EOF

# /etc/systemd/system/pulsar-broker.service
cat > /etc/systemd/system/pulsar-broker.service << EOF
[Unit]
Description=Apache Pulsar Broker
After=pulsar-bookkeeper.service
Requires=pulsar-bookkeeper.service

[Service]
Type=forking
User=pulsar
Group=pulsar
Environment=JAVA_HOME=/usr/lib/jvm/java-11-openjdk
Environment=PULSAR_HOME=/opt/pulsar
Environment=PULSAR_MEM="-Xms4g -Xmx4g -XX:MaxDirectMemorySize=4g"
ExecStart=/opt/pulsar/bin/pulsar-daemon start broker
ExecStop=/opt/pulsar/bin/pulsar-daemon stop broker
Restart=on-failure
LimitNOFILE=1048576

[Install]
WantedBy=multi-user.target
EOF

# 启动服务（所有节点）
systemctl daemon-reload
systemctl enable pulsar-zookeeper pulsar-bookkeeper pulsar-broker
systemctl start pulsar-zookeeper
sleep 10
systemctl start pulsar-bookkeeper
sleep 10
systemctl start pulsar-broker

3.7 验证集群状态

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 检查 ZooKeeper 状态
echo ruok | nc pulsar-node1 2181  # 应返回 imok

# 检查 BookKeeper 状态
/opt/pulsar/bin/bookkeeper shell bookie_info

# 检查 Broker 状态
curl http://pulsar-node1:8080/admin/v2/brokers/cluster/pulsar-cluster

# 查看集群信息
/opt/pulsar/bin/pulsar-admin clusters get pulsar-cluster

# 查看 Bookie 状态
/opt/pulsar/bin/bookkeeper shell whatisinstanceid

四、租户与命名空间管理

4.1 创建租户

1
2
3
4
5
6
7
8
9
10
# 创建租户
/opt/pulsar/bin/pulsar-admin tenants create tenant-a \
  --admin-roles admin \
  --allowed-clusters pulsar-cluster

# 查看租户列表
/opt/pulsar/bin/pulsar-admin tenants list

# 查看租户信息
/opt/pulsar/bin/pulsar-admin tenants get tenant-a

4.2 创建命名空间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 创建命名空间
/opt/pulsar/bin/pulsar-admin namespaces create tenant-a/namespace-prod

# 配置命名空间策略
/opt/pulsar/bin/pulsar-admin namespaces set-retention tenant-a/namespace-prod \
  --size -1 --time 10080  # 无限大小，保留 7 天

/opt/pulsar/bin/pulsar-admin namespaces set-backlog-quota tenant-a/namespace-prod \
  --limit 10737418240 --limitTime 86400 --policy producer_exception

/opt/pulsar/bin/pulsar-admin namespaces set-message-ttl tenant-a/namespace-prod \
  --ttl 259200  # 消息 TTL 3 天

# 查看命名空间
/opt/pulsar/bin/pulsar-admin namespaces list tenant-a

4.3 创建主题

1
2
3
4
5
6
7
8
9
# 创建分区主题
/opt/pulsar/bin/pulsar-admin topics create persistent://tenant-a/namespace-prod/topic-orders \
  --partitions 6

# 查看主题信息
/opt/pulsar/bin/pulsar-admin topics stats persistent://tenant-a/namespace-prod/topic-orders

# 查看主题列表
/opt/pulsar/bin/pulsar-admin topics list tenant-a/namespace-prod

五、生产消费示例

5.1 Java 客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// Producer 示例
import org.apache.pulsar.client.api.*;

PulsarClient client = PulsarClient.builder()
    .serviceUrl("pulsar://pulsar-node1:6650,pulsar-node2:6650,pulsar-node3:6650")
    .build();

Producer<String> producer = client.newProducer(Schema.STRING)
    .topic("persistent://tenant-a/namespace-prod/topic-orders")
    .create();

for (int i = 0; i < 100; i++) {
    MessageId msgId = producer.send("Order-" + i);
    System.out.println("Sent message: " + msgId);
}

producer.close();
client.close();

// Consumer 示例
Consumer<String> consumer = client.newConsumer(Schema.STRING)
    .topic("persistent://tenant-a/namespace-prod/topic-orders")
    .subscriptionName("order-processor")
    .subscriptionType(SubscriptionType.Shared)
    .subscribe();

while (true) {
    Message<String> msg = consumer.receive();
    System.out.println("Received: " + msg.getValue());
    consumer.acknowledge(msg);
}

5.2 Python 客户端

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import pulsar

# Producer
client = pulsar.Client('pulsar://pulsar-node1:6650')
producer = client.create_producer('persistent://tenant-a/namespace-prod/topic-orders')

for i in range(100):
    producer.send(f'Order-{i}'.encode('utf-8'))
    print(f"Sent Order-{i}")

producer.close()
client.close()

# Consumer
client = pulsar.Client('pulsar://pulsar-node1:6650')
consumer = client.subscribe(
    'persistent://tenant-a/namespace-prod/topic-orders',
    subscription_name='order-processor',
    consumer_type=pulsar.ConsumerType.Shared
)

while True:
    msg = consumer.receive()
    print(f"Received: {msg.data().decode('utf-8')}")
    consumer.acknowledge(msg)

5.3 命令行测试

1
2
3
4
5
6
7
# 生产消息
/opt/pulsar/bin/pulsar-client produce persistent://tenant-a/namespace-prod/topic-orders \
  --messages "Hello Pulsar"

# 消费消息
/opt/pulsar/bin/pulsar-client consume persistent://tenant-a/namespace-prod/topic-orders \
  --subscription-name test-sub -n 10

六、运维管理

6.1 监控指标

Pulsar 暴露 Prometheus 格式的指标：

1
2
3
4
5
6
7
8
9
10
11
# Broker 指标
curl http://pulsar-node1:8080/metrics/

# BookKeeper 指标
curl http://pulsar-node1:8000/metrics/

# 关键指标：
# - pulsar_rate_in/out: 消息吞吐
# - pulsar_latency_le_*: 消息延迟
# - bookkeeper_journal_*: BookKeeper 性能
# - bookkeeper_ledger_*: Ledger 状态

6.2 常用运维命令

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 查看 Broker 负载
/opt/pulsar/bin/pulsar-admin brokers list pulsar-cluster

# 查看 Bookie 状态
/opt/pulsar/bin/bookkeeper shell listbookies -rw

# 查看 Ledger 信息
/opt/pulsar/bin/bookkeeper shell listledgers

# 触发 Ledger 清理
/opt/pulsar/bin/bookkeeper shell ledgerchecker

# 查看主题统计
/opt/pulsar/bin/pulsar-admin topics stats persistent://tenant-a/namespace-prod/topic-orders

# 查看订阅信息
/opt/pulsar/bin/pulsar-admin topics stats-internal persistent://tenant-a/namespace-prod/topic-orders

# 重置消费位置
/opt/pulsar/bin/pulsar-admin topics reset-cursor persistent://tenant-a/namespace-prod/topic-orders \
  --subscription order-processor --time 1710316800000

# 跳过消息
/opt/pulsar/bin/pulsar-admin topics skip persistent://tenant-a/namespace-prod/topic-orders \
  --subscription order-processor -n 1000

6.3 备份与恢复

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 备份 ZooKeeper 数据
# 在 ZooKeeper 节点执行
tar -czf /backup/zk_backup_$(date +%Y%m%d).tar.gz /data/pulsar/zookeeper

# 备份 BookKeeper Ledger
# 使用 bookkeeper shell 导出
/opt/pulsar/bin/bookkeeper shell ledgermanager dump > /backup/ledger_dump.txt

# 恢复 ZooKeeper
# 1. 停止 Pulsar 服务
# 2. 清空 zookeeper 数据目录
# 3. 解压备份
# 4. 重启服务

# 恢复 BookKeeper
# BookKeeper 数据通过副本自动恢复
# 严重故障需从备份重建集群

6.4 扩容操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 添加新 Bookie 节点
# 1. 在新节点安装 Pulsar
# 2. 配置 bookkeeper.conf（指向现有 ZK）
# 3. 启动 BookKeeper 服务
# 4. 新 Bookie 自动加入集群

# 添加新 Broker 节点
# 1. 在新节点安装 Pulsar
# 2. 配置 standalone.conf（指向现有 ZK 和集群）
# 3. 启动 Broker 服务
# 4. 新 Broker 自动注册到集群

# 验证扩容
/opt/pulsar/bin/pulsar-admin brokers list pulsar-cluster
/opt/pulsar/bin/bookkeeper shell listbookies -rw

七、故障排查

7.1 Bookie 只读模式

现象： Bookie 进入只读模式，无法写入新数据

排查：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 检查 Bookie 状态
/opt/pulsar/bin/bookkeeper shell bookie_info

# 检查磁盘空间
df -h /data/pulsar/bookkeeper

# 检查 BookKeeper 日志
tail -f /opt/pulsar/logs/bookkeeper.log

# 常见原因：
# 1. 磁盘空间不足（>90%）
# 2. Ledger 数量超过阈值
# 3. Journal 写入失败

# 解决方案：
# 1. 清理磁盘空间
# 2. 触发压缩：bin/bookkeeper shell triggergc
# 3. 重启 Bookie

7.2 Broker 连接 BookKeeper 失败

现象： Broker 日志出现 BookKeeper 连接错误

排查：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 检查 ZK 连接
echo stat | nc pulsar-node1 2181

# 检查 Bookie 是否在线
/opt/pulsar/bin/bookkeeper shell listbookies -rw

# 检查网络连通性
telnet pulsar-node1 3181

# 查看 Broker 日志
tail -f /opt/pulsar/logs/pulsar-broker.log | grep -i bookkeeper

# 解决方案：
# 1. 重启 BookKeeper 服务
# 2. 检查防火墙规则
# 3. 验证 ZK 集群健康

7.3 消息积压

现象： Consumer 消费速度慢，消息大量积压

排查：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 查看积压情况
/opt/pulsar/bin/pulsar-admin topics stats persistent://tenant-a/namespace-prod/topic-orders

# 查看订阅详情
/opt/pulsar/bin/pulsar-admin topics stats-internal persistent://tenant-a/namespace-prod/topic-orders

# 查看 Consumer 连接
/opt/pulsar/bin/pulsar-admin topics peek-messages persistent://tenant-a/namespace-prod/topic-orders \
  --subscription order-processor -n 10

# 解决方案：
# 1. 增加 Consumer 实例数
# 2. 优化 Consumer 消费逻辑
# 3. 调整分区数
# 4. 检查 Consumer 是否频繁 ack 失败

7.4 ZooKeeper 性能问题

现象： ZK 响应慢，影响 Pulsar 操作

排查：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 检查 ZK 延迟
echo mntr | nc pulsar-node1 2181 | grep latency

# 检查 ZK 连接数
echo stat | nc pulsar-node1 2181

# 检查 ZK 日志
tail -f /opt/pulsar/logs/zookeeper.log

# 解决方案：
# 1. 增加 ZK 节点数
# 2. 优化 ZK JVM 参数
# 3. 检查磁盘 IO 性能
# 4. 分离 ZK 集群（独立部署）

7.5 常见错误码

错误 原因 解决方案
ServiceUnitNotReady Broker 未准备好 等待 Broker 启动完成
BrokerPersistenceException BookKeeper 写入失败 检查 Bookie 状态
MetadataNotFoundException 主题/命名空间不存在 检查路径是否正确
ConsumerBusyException Consumer 独占模式冲突 检查订阅模式
ProducerFencedException Producer 被隔离 重新创建 Producer

八、性能调优

8.1 JVM 调优

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Broker JVM 参数
PULSAR_MEM="-Xms8g -Xmx8g \
  -XX:MaxDirectMemorySize=8g \
  -XX:+UseG1GC \
  -XX:MaxGCPauseMillis=200 \
  -XX:+ParallelRefProcEnabled \
  -XX:+UnlockExperimentalVMOptions \
  -XX:+AggressiveOpts"

# BookKeeper JVM 参数
BOOKIE_MEM="-Xms4g -Xmx4g \
  -XX:MaxDirectMemorySize=4g \
  -XX:+UseG1GC \
  -XX:MaxGCPauseMillis=200"

8.2 BookKeeper 调优

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# bookkeeper.conf
# 增加 Journal 大小
journalMaxSizeMB=8192

# 优化 GC
gcWaitTime=600000

# 调整 Ledger 配置
ledgersPerBookie=10000000
maxAddPerMsecs=10000

# 网络优化
nettyMaxFrameSizeBytes=16777216
recvBufferSize=1048576
sendBufferSize=1048576

8.3 Broker 调优

1
2
3
4
5
6
7
8
9
10
11
12
13
# standalone.conf
# 增加线程数
numHttpServerThreads=32
numAccepterThreads=8

# 优化消息处理
maxMessageSize=10485760
brokerMaxPublishRate=100000
brokerMaxPublishMessageRate=500000

# 缓存优化
managedLedgerCacheSizeMB=1024
managedLedgerDefaultMarkDeleteRateLimit=100

8.4 生产环境建议

配置项 建议值 说明
Bookie 数量 ≥3 保证数据副本
Broker 数量 ≥2 高可用
ZooKeeper 数量 3 或 5 奇数节点
分区数 消息量/10 万/天 根据吞吐量调整
副本数 2 或 3 平衡可靠性和性能
保留策略 7-30 天 根据业务需求

九、高可用配置

9.1 跨地域复制

1
2
3
4
5
6
7
8
9
10
11
# 配置集群复制
/opt/pulsar/bin/pulsar-admin clusters create cluster-b \
  --service-url http://cluster-b:8080 \
  --broker-url pulsar://cluster-b:6650

# 配置命名空间复制
/opt/pulsar/bin/pulsar-admin namespaces set-replication-clusters \
  tenant-a/namespace-prod pulsar-cluster,cluster-b

# 查看复制状态
/opt/pulsar/bin/pulsar-admin topics stats persistent://tenant-a/namespace-prod/topic-orders

9.2 故障转移

1
2
3
4
5
6
7
8
9
# 手动触发 Broker 故障转移
# 1. 停止故障 Broker
systemctl stop pulsar-broker

# 2. Broker 自动重新平衡
# 3. 消费者自动重连到新 Broker

# 查看 Broker 负载分布
/opt/pulsar/bin/pulsar-admin brokers list pulsar-cluster

十、总结

Apache Pulsar 作为云原生消息流平台，具有以下优势：

  1. 架构优势：计算存储分离，弹性扩展
  2. 功能丰富：多租户、多协议、消息追溯
  3. 高可用：内置副本、跨地域复制
  4. 运维友好：完善的监控指标和管理工具

部署运维要点：

  1. 规划先行：合理评估硬件资源和网络拓扑
  2. 监控完善：配置 Prometheus+Grafana 监控
  3. 定期维护：检查磁盘、清理日志、更新固件
  4. 故障预案：制定故障处理和恢复流程

参考文档：