Apache Pulsar在vivo的探索与实践
如果无法正常显示，请先停止浏览器的去广告插件。
1. Apache Pulsar在vivo的探索与实践
Speaker 全利民 陈建波
2. About Speaker
全利民
vivo大数据工程师，负责vivo分布式消息中间件建设
陈建波
vivo大数据工程师，曾任微服务应用架构师，现负责vivo分布式消息中间件的能力建设
3. 一． 业务介绍
目录 二． Pulsar选型背景
CONTENTS 三． Pulsar集群管理实践
四． Pulsar监控实践
五． 问题优化案例实战
4. 业务介绍
为vivo所有内销与外销实时计算业务提供高吞吐、低延时服务
广泛用于实时计算、日志数据收集、监控系统等场景
十万亿级记录/天数据
多集群运维部署
5. 业务介绍
资源隔离
集群中不同对业务进行资源组物理隔离，避免各业务之间相互影响
流量均衡
按照资源组维度针对topic的流量、分区分布分散性、磁盘容量、机
器机架等指标生成topic迁移计划进行流量均衡
6. 一． 业务介绍
目录 二． Pulsar选型背景
CONTENTS 三． Pulsar集群管理实践
四． Pulsar监控实践
五． 问题优化案例实战
7. Pulsar选型背景
流量几十倍的上涨：
存量集群越来越大，topic以及topic分区总量不断增加，性能受到影响
集群拆分后，资源利用率低，运维成本增加
机器扩容慢，需要做长时间的流量均衡，难以应对突发流量
消费端的性能扩展太依赖分区扩容，导致集群的元数据疯狂增长
硬件故障概率高，影响直接传导到客户端
总结： 面对庞大的集群、流量以及各种业务场景，综合考虑集群的稳定性和维护成本等因素，我们需要一个功能
更加丰富、适用更多场景、扩展能力更强的消息组件
8. Pulsar选型背景
9. Pulsar选型背景
特性 Pulsar
海量分区支撑 具备支持海量分区的能力。集群性能不会因为分区数量的
增加而导致性能急剧下降。
消费扩展 消费者数量不受限于 Topic 的分区个数，业务可按需启
动对应的消费者数量
精准的限流 Pulsar支持namespace或Topic层级的限流，维度更细
更精准
流量均衡
快速扩缩容
实时均衡，无需数据迁移
秒级
故障恢复 大多数情况下存储节点机器故障对用户基本无感知，计算
节点故障秒级转移
分层存储 海量数据存储
容器部署 基于云原生设计，集群容器化友好
异地多活 Pulsar内置的跨地域复制特性，同时支持双向复制与单向
复制，该能力可用于无缝跨地域同步数据或复制数据到其
他集群，以实现如异地多活、灾备等功能
10. 一． 业务介绍
目录 二． Pulsar选型背景
CONTENTS 三． Pulsar集群管理实践
四． Pulsar监控实践
五． 问题优化案例实战
11. Bundle的管理
Bundle设置需要考量：
• Bundle的个数影响均衡的效果
• Bundle数据存储在zk中
设置思路：
• Bundle数（举例）：broker节点数*20；[200，500]
• Topic分区数：流量大的topic需要针对性的扩大分区
Bundle的操作建议：
• unload bundle而不是unload整个namespace
• 利用bundle split增加namespace的bundle数量
• 注意的问题：流量的均衡性和集群的稳定性
• 使用建议：合理的bundle数量和最小维度的运维操作
12. 数据的管理
Ledger翻转
在满足ledger最小翻转时间以及以下条件之一后触发ledger翻转：
• 已达到ledger最大翻转时间
• 已达到ledger的最大entry数量
• 已达到ledger的最大大小
触发ledger翻转的最小时间：
managedLedgerMinLedgerRolloverTimeMinutes=10
触发ledger翻转的最长时间：
managedLedgerMaxLedgerRolloverTimeMinutes=240
触发ledger翻转的最大entry数：
managedLedgerMaxEntriesPerLedger=50000
• 注意的问题：磁盘存储不均衡、数据卸载时间长
• 使用建议：根据业务的消息情况适当调整ledger翻转参数和扩
大分区，防止ledger过大、ledger大小不均衡
触发ledger翻转的最大大小：
managedLedgerMaxSizePerLedgerMbytes=2048
13. 数据的管理
数据过期
第一阶段：未被ack的消息
• backlog消息：该段数据不会被删除
第二阶段：已经ack的消息
• 订阅主动ack后，标记为非backlog消息
• TTL：超过TTL存活时间的消息会被主动ack，本质上是移动cursor
第三阶段：消息保留时间检查
• Retenion：对已经ack的消息的保留策略，可以按照保留周期和保
留大小。
• 注意的问题：提供一个简单统一的过期策略
• 使用建议：TTL=Retenion保留周期值
第四阶段：消息删除
• deleted：超过Retenion范围的消息则被删除
14. 数据的管理
数据删除
Entry Log flie：
将同一段时间内写入的数据（可能归属于多个 ledger 的 entry）经过
排序后 flush 到同一个 entry log 文件中，将索引存放在 RocksDB 中
GarbageCollector（GC）线程：
GC清理线程分为 ：minorCompaction和 majorCompaction
•
•
•
•
minorCompactionInterval=3600
minorCompactionThreshold=0.2
majorCompactionThreshold=0.5
majorCompactionInterval=86400
GC 线程会定时扫描每个 entry log的元数据信息（ EntryLogMetadata）
根据剩余有效数据比例判断是否compaction 。
• 注意的问题：数据的及时清除
• 使用建议：按照业务流量和磁盘空间适当调整数据清理间隔时间
或者有效数据阈值并配合限速策略减小对集群的影响
compaction限速策略：
• 每秒读取 entry 的最大数量：compactionRate=1000
• 每秒读取多少字节数据：
isThrottleByBytes=false；compactionRateByBytes=1000000
15. 一． 业务介绍
目录 二． Pulsar选型背景
CONTENTS 三． Pulsar集群管理实践
四． Pulsar监控实践
五． 问题优化案例实战
16. Pulsar指标监控链路
Pulsar 指标监控：
采用prometheus采集pulsar指标;
应用prometheus远程存储特性将格式化后的指标发送到kafka；
druid消费Kafka数据后可以作为grafana的数据源，配置grafana面板查询指标；
17. 核心指标
充分利用指标监控
1、指标类型：
• 客户端指标：用来排查客户端出现的异常
• broker端指标：监控topic流量、调整broker间流量差距
• bookie端指标：排查读写延迟等问题
2、自开发的指标：
• bundle相关指标：分区数、流量等在bundle上的分布
• broker端记录读写延迟p95/p99值
• broker端网络负载指标
18. 一． 业务介绍
目录 二． Pulsar选型背景
CONTENTS 三． Pulsar集群管理实践
四． Pulsar监控实践
五． 问题优化案例实战
19. 负载均衡参数
目标：节点流量偏差20%以内，每天均衡频次在10次内
# load shedding strategy, support OverloadShedder and ThresholdShedder, default is OverloadShedder
loadBalancerLoadSheddingStrategy=org.apache.pulsar.broker.loadbalance.impl.ThresholdShedder
# enable/disable namespace bundle auto split
loadBalancerAutoBundleSplitEnabled=false
# enable/disable automatic unloading of split bundles
loadBalancerAutoUnloadSplitBundlesEnabled=false
#计算新资源使用量时的CPU使用权重（默认1.0）
loadBalancerCPUResourceWeight=0.0
#计算新的资源使用量时的堆内存使用权重（默认1.0）
loadBalancerMemoryResourceWeight=0.0
#计算新资源使用量时的直接内存使用权重（默认1.0）
loadBalancerDirectMemoryResourceWeight=0.0
20. 负载均衡案例
1个Topic 30个分区 180个Bundle
•
•
•
•
节点间流量差异大
均衡频次高，一天大概有200多次
客户端连接频繁切换，流量波动大
每个bundle的分区数量分布差异大
21. 负载均衡案例
增加到120个分区 180个Bundle
•
•
•
•
节点间流量差异小
均衡频次降低，一天大概有10次左右
客户端连接切换减少，流量波动较小
每个bundle的分区数量分布差异降低
22. 客户端发送性能
单个topic，30个分区增加到120个，系统滚动重启后流量下降
客户端配置参数：
memoryLimitBytes=209715200 （默认为0）
maxPendingMessages=2000 （默认1000）
maxPendingMessagesAcrossPartitions=40000 （默认50000）
batchingMaxPublishDelayMicros=50 （默认1毫秒）
batchingMaxMessages=2000 （默认1000）
batchingMaxBytes=5242880 （默认128KB）
23. 客户端发送性能
• maxPendingMessages（队列长度）=min(maxPendingMessages,
maxPendingMessagesAcrossPartitions/partitionNum)
• 分区数添加后，需要重启客户端才对队列长度生效
• maxPendingMessages队列长度从40000/30=1333变为40000/120=333
24. 客户端发送性能
测试案例：batchingMaxMessages调小后性能提升10倍
客户端配置参数：
memoryLimitBytes=209715200
maxPendingMessages=2000
maxPendingMessagesAcrossPartitions=40000
batchingMaxPublishDelayMicros=50
batchingMaxBytes=5242880
建议
• batchingMaxPublishDelayMicros不要过大
• 确保batchingMaxMessages比maxPendingMessages
要大，否则等待batchingMaxPublishDelayMicros才发
送
25. 宕机导致集群流程骤降
分区队列满会导致发送线程阻塞，影响所有分区的整体发送，影响集群稳定性
26. 宕机导致集群流程骤降
优化思路：
• 能者多劳：让发送快的分区尽可能多的发送
• 把阻塞点从ProducerImpl移到PartitionedProducerImpl
• 如果分区ProducerImpl出现队列已满阻塞较长时间，把该分区排除掉
27. 宕机导致集群流程骤降
优化后，宕机broker流量快速转移到其他broker
注：优化只支持RoundRobinPartitionMessageRouterImpl路由策略
在单个ProducerImpl对应的broker出现处理慢、网络慢等导致发送响应慢的情况，都可能会
导致发送线程的阻塞，业务发送消息的速度受限于最慢那个ProducerImpl的速度
28. 未来展望
1. 生产端到消费端的全链路监控
2. 支撑 Flink、Spark、Druid 等核心下游组件打通落地
3. 基于KoP的Kafka万亿流量迁移
4. 容器化
29. THANK
YOU
关注vivo互联网技术