BIGO优化Apache Pulsar系列-Bookie负载均衡

基于强大的音视频处理技术、全球音视频实时传输技术、人工智能技术、CDN 技术，BIGO 研发及投资了多款音视频类社交及内容产品，包括 Bigo Live、Likee等。目前，BIGO 在全球拥有亿级月活用户，产品及服务已覆盖超过 150 个国家和地区。随着业务的迅速增长，BIGO 消息队列平台承载的数据规模出现了成倍增长，下游的在线模型训练、在线推荐、实时数据分析、实时数仓等业务对消息的实时性和稳定性提出了更高的要求。

Apache Pulsar是Apache软件基金会顶级项目，作为下一代云原生分布式消息流平台，具有强一致性、高吞吐以及低延时等特性。Pulsar 采用计算与存储分离的分层架构设计，存储层由Apache BookKeeper项目提供支持，一个BookKeeper服务器节点称为bookie。

为了实现读写隔离，部署过程中会给一个bookie分配一个ledger盘和一个journal盘，写入数据时顺序写入journal盘，成功持久化后即可立刻响应成功，从而实现低延迟的写入，bookie内部还会聚合属于同一个Ledger的所有entry，然后刷盘到ledger盘，这样读取的时候也能支持顺序读。

可见，journal盘对写入速度要求较高，对存储空间大小要求较低；ledger盘对存储空间大小要求较高，对写入速度要求较低。理论上journal盘采用SSD能达到更好的写入性能，ledger盘采用大容量的HDD即可满足需求。但消息队列写入的数据均为热数据，写入数据量大，保存时间不会超过7天，而SSD使用的闪存介质的可编程和擦除次数有限，因此实践过程中基于成本考虑，并没有采用SSD磁盘供journal盘使用。我们全部采用大容量的HDD盘，一个bookie独占一个磁盘作为一个Ledger盘，三个bookie共用一个磁盘作为Journal盘。

注记：尽管多个bookie共用一个磁盘作为journal盘，但是journal磁盘的空间使用率也是极低的，会有存储空间资源的浪费，我们接下来会将journal盘去掉，释放更多的存储空间，供紧缺的ledger盘使用，同时也能进一步缓解写延迟高的问题。

随着业务流量的持续增长，集群的压力越来越大，直接表现就是集群的P99写延迟升高，一般来说当P99写延迟达到秒级时，就会触发电话告警，提醒集群扩容，缓解单台机器的压力。

但是，写延迟过高不一定是集群硬件资源不够导致的，也有可能各种其他原因，比如说单个topic分区的流量过大，达到上百MB，而单个HDD盘的最大顺序写入速度也就100多MB，而且journal盘还是三个bookie共用的，此时该journal盘服务的所有分区肯定都会出现写延迟高的问题，解决办法很简单，扩大topic分区数，将流量打散到多个journal盘即可。

还有可能是系统设计本身的问题：journal盘之间的流量压力不均衡。我们收集统计journal盘的写入吞吐统计图，横轴为流量吞吐区间，纵轴为处于该流量吞吐压力区间的journal盘个数，如下图所示：

Instance: instance4 device:sdw, value: 154.90 MB/s

Instance: instance7 device:sdw, value: 130.20 MB/s

Instance: instance5 device:sdw, value: 124.98 MB/s

Instance: instance1 device:sdt, value: 120.98 MB/s

Instance: instance15 device:sdw, value: 119.92 MB/s

......

Instance: instance10 device:sds, value: 47.44 MB/s

Instance: instance13 device:sds, value: 40.51 MB/s

Instance: instance12 device:sdu, value: 37.99 MB/s

Instance: instance14 device:sds, value: 23.67 MB/s

由此可见，journal盘的写入吞吐的负载分布相当地不均衡。压力最大的journal盘吞吐能达到150MB/s，这接近HDD的性能上限了，却有不少journal盘流量吞吐小于50MB/s。这构成了我们做bookie写负载均衡的基本动机。

当前BookKeeper在创建Ledger的时候，会根据放置策略来选择bookie，比如说使用RackawareEnsemblePlacementPolicy来强制不同的数据副本放置在不同的机架上，以保证机架级别的容灾；或者使用RegionAwareEnsemblePlacementPolicy强制将不同数据的副本放置在属于不同区域的不同机架上以实现区域级故障容灾。在经过这些放置策略的选择后，还是会有相当多的候选bookie作为新Ledger的ensemble，这就给了我们实现负载均衡算法的空间。

我们只做普通写场景的负载均衡，对于读和recovery写的场景都不加入负载均衡。原因如下：

·读的负载均衡

一方面是当前读延迟问题不严重，另一方面是读ledger时只能从三副本（甚至是2副本）里挑选Bookie来读取，选择的空间不大，因此不考虑读方面的负载均衡。

·recovery

auto recovery的时候默认是抢到任务的bookie把数据拷贝到本地，这一块不建议加入负载均衡机制。

负载单元

·broker的负载单元是bundle

bundle的生命周期是持久的，或者说它施加的负载可以认为是长久的；broker使用AvgShedder做一次负载均衡决策后，基本都不用再做决策了，因为流量在一定时间内是稳定的。

注记：AvgShedder是BIGO内部实现的一个负载均衡算法，代码已经贡献给社区。

https://github.com/apache/pulsar/pull/18186

bundle的负载信息通过时间聚合的方式统计得到它的长期、短期负载信息，负载信息包括写入/消费消息数率、吞吐量。

·bookie的负载单元是ledger

ledger的生命周期是相对短暂的（但至少为10min，这一点后面分析可知），关闭它之后它就不会再施加任何写入压力了；bookie创建ledger是持续进行的事情，需要不断地做决策。ledger的负载信息可以包含它的预计空间大小、预计写入吞吐大小，供决策者使用。

资源单位

·broker的资源单位就是单个机器/节点

因为一台机器只会部署一个broker，机器上的各种负载信息（包括CPU使用率、网卡使用率等）可以视为该broker一对一的负载信息。

因为一个journal盘实际上会给多个bookie使用，journal盘主要关心写入吞吐、写入延迟的指标，同一个journal盘里的不同bookie写入是否均衡其实没有影响。

对于ledger盘，一个bookie独享一个ledger盘，ledger盘主要关心它的磁盘空间使用率。

那么我们在挑选bookie时，就需要同时考虑一个bookie的journal盘写入吞吐、ledger盘的磁盘使用率，因此可以使用一个二元组来表示单个Bookie的负载信息。

决策频率

·broker基本只需要在broker加入、退出的时候做一次决策

当然如果负载均衡算法不好，或者配置不合适，可能会发生频繁的bundle卸载、加载的情况，这会造成客户端流量抖动的问题，这是broker负载均衡算法的问题了，这里不深入讨论。

·bookie需要不断做决策

一个topic是由一个ledger列表构成的，当达到阈值时broker会主动关闭一个Ledger，并创建一个新的Ledger给该topic使用。因此集群是需要不断地创建新Ledger的。

决策的频率是会影响到架构的设计的，具体来讲，会影响我们选择中心化决策还是分布式决策。

— 如果每个broker都自己做决策（分布式决策），由于每个broker拿到同一份数据、执行同一个算法，这种情况下会造成流量倾斜到某几个bookie，负载均衡效果可能会很差。当然可以在算法层面缓解该问题，即给算法引入随机化因素，尽量缓解倾斜的问题。

— 如果实现中心化的决策，那就就没有倾斜的问题。

可以看到，集群的负载均衡决策频率越高，我们选择分布式决策的架构的风险会越高。

我们搜集某线上集群的统计数据，如下图是某集群流量高峰时的统计图，每一个柱子对应1s内创建ledger的个数，可见，就算没有集群重启平常也是能达到十几个ledger的创建速度的。

负载变化速度

·broker一定时间内可以认为流量是不变的

前面也提到过，broker使用AvgShedder做一次负载均衡决策后，基本都不用再做决策了，因为流量一定时间内基本是稳定的，以一天为周期，周期性波动。

·bookie的负载是会一直动态变化的

因为bookie的负载单元是ledger，而ledger是有生命周期的，达到阈值时就会触发ledger关闭，创建新的ledger。一个分区在一个时间点上只有一个ledger服务，因此一个Ledger施加的压力就是一个分区施加的压力，一个journal盘少服务或者多服务一个Ledger的区别是很明显的，甚至可能一个journal盘被某一个分区的流量几乎完全霸占。

如果bookie的负载变化很快，那么也是无法做负载均衡的，因为刚收集的负载数据可能都已经过期了。

搜集日志（凌晨截至早上11:00的日志数据），针对每个topic/分区，搜集它两次创建ledger之间的时间间隔，就得到了一个ledger的生命周期长度，统计其分布得到下图，x轴为ledger的生命周期长度，y轴为ledger的个数。（下图只展示了800s以下的数据，展示太多内容图片会很难看。）

数量这么多的ledger生命长度为600s，即10min，而且没有小于600s的，肯定是有原因的。

查看broker配置managedLedgerMinLedgerRolloverTimeMinutes默认为10min，即一个ledger最快也要10min才能切换到下一个ledger。

当达到managedLedgerMinLedgerRolloverTimeMinutes时间，而且另外三个阈值有一个触发时，才会触发ledger的切换。

因此，一个ledger的生命周期其实不是很短，它对某个bookie的流量压力至少也要持续10min（不考虑bookie滚动重启导致ledger close的情况），如果觉得bookie流量变化速度还是过快，可以调大managedLedgerMinLedgerRolloverTimeMinutes配置为15min。

当然，虽然Ledger的生命周期是决定journal盘写入压力是否稳定的一个核心因素，但我们还是要看一下journal盘写入吞吐的真实波动情况，因为一个journal盘不只服务一个Ledger，真实环境中多个Ledger动态变化的复杂系统里，journal盘写入吞吐的变化周期并不等于一个Ledger的生命周期。

如下图：

负载信息上报方式、存储位置

broker

broker的负载信息由每个broker上报到zk上。但是由于每个broker、bundle都对应一个znode，特别是bundle个数达到1k多，而且当前每个broker都会去读取所有信息，导致每次拉取数据时对zk造成较大压力，目前通过降低上报、拉取的频率来大大缓解了（每3min一次）。

broker最新版的负载均衡器已经把负载数据迁移到topic里存储了，负载均衡本身不是核心模块，不能滥用zk的资源，zk延迟拉高时集群会面临整个挂掉的风险。

bookie

bookie是自己上报数据呢？还是由外部组件拉取呢？上报的话写给谁呢？zk还是broker？

前面说了，负载数据只有磁盘的使用率、写入吞吐/延迟，搜集的逻辑都比较简单，也不一定要bookie来参与，其实我们只需要知道所有磁盘的信息，然后知道磁盘与bookie之间的映射关系就足够了，因此选择余地很大。

下面列举一些可能的方式供参考：

1. 最传统的方式，bookie上报到zk上。

集群有几百个bookie，因此至少需要几百个znode来存储该数据，一方面是频繁更新数据会对zk造成压力，而且broker、bookie集群都强依赖zk，zk一旦出了问题整个集群都会挂，因此会引入新的风险；而且zk的读延迟是会达到秒级的（特别是replicate的时候，这种延迟对于bookie的负载均衡可能是无法接受的），因此这种方案其实可以不用考虑了。

2. 每个bookie把自身的负载数据写入一个对应的ledger，决策者去读这些ledger。

注记：因为一个ledger只能有一个writer，所以不能多个bookie同时往一个ledger里写，每个bookie都往自己的ledger写。

优点：能利用到bookie自身优秀的架构设计，延迟能达到几毫秒；而且对于多个决策者的情况下，多个决策者都去读这个ledger就行了，能很好地利用到读写缓存。

忧虑：每个bookie都拥有一个自己的ledger，因此会增加几百个Ledger用于写入负载信息，可能会增加集群元数据压力。

3. 每个bookie仅在内存维护自身负载数据，提供接口给外部访问。

优点：负载数据不用使用存储起来，因为负载数据变化比较快，参考后面的负载变化速度图，因此存储起来反而必要性不大。

忧虑：暴露接口给外部调用，较高请求频率下是否延迟会很高？比如说分布式决策的情况下，每个broker都需要给每个bookie都发送请求，当集群规模较大时则可能会遇到请求压力大和延迟高的问题。

为了集群的架构简洁，而且我们的集群规模不算很大，我们最终采用这种方式来实现。

4. 由一个外部的组件来抓取集群所有机器的磁盘的负载情况，结合zk上的信息映射好磁盘和bookie的关系，该组件提供接口给外部访问。

优点：

-压力、资源消耗最小，该组件只需对每台机器都搜集一下磁盘负载即可，每个broker只需要对该组件请求一次即可，因此可能延迟也能达到一个理想的效果。

-实现起来也最为容易。

缺点：

-增加集群的组件，架构变复杂。

-如果要采集bookie内部的指标，则外部组件难以搜集到这种数据，但是当前应该是不考虑bookie内部的一些指标的，指标够用就行，过多不仅容易导致逻辑复杂，而且容易造成效果差的问题，Pulsar负载均衡考虑直接内存使用率就是一个反面例子，Pulsar新版代码已经默认关掉考虑直接内存了。

https://github.com/apache/pulsar/pull/21168

决策者

broker

bookie

bookie的决策者可以放在bookie client里实现，也可以引入一个外部的组件advisor角色，下面讨论两种设计的优缺点：

1. bookie client里实现

优点：不需要引入外部的组件，避免增加系统的复杂度。

缺点：

- 每个broker都做决策，会出现前面提到的流量倾斜的问题；

- 而且每个broker都要读流量，会造成负载数据存储端的压力，特别是用zk来存储的时候。

- 侵入bookie client的代码。

- 需要增加资源来频繁地拉取更新负载数据，否则负载均衡效果差。

优点：

- 它无需高可用，只是一个尽力而为的服务，给bookie client提供建议，帮忙挑选ensemble，如果它挂了或者超时了，bookie client直接使用原来的逻辑兜底即可。

- 对bookie client的代码仅做细微的修改，而且基本没有风险。

- 同一个数据中心里数据包往返的时间只有几十us，延迟可以接受。

- 中心化决策，避免流量倾斜的问题，且同一份负载数据只需读取一次。

缺点：增加一个组件，运维麻烦。

我们最终选择了在bookie client里实现。

架构设计

根据前面的所有分析与权衡，我们最终采用的架构如下：bookie 新开一个线程采集数据，并通过接口提供给外部，然后broker也新开一个线程，通过访问接口的方式不断搜集数据（如每3s或者5s获取一次），并在每次创建ledger时，本地执行负载均衡算法做决策。

上图介绍了Broker、Bookie负载信息搜集的架构，以及Broker利用这些负载信息本地执行负载均衡算法。

上图具体介绍了创建一个ledger的流程。

在请求bookie client创建Ledger时，可以提供一些数据，即ledger的预计空间大小、预计写入吞吐大小。因为我们只考虑broker写入端（不考虑auto recovery写入），broker里的org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl（Pulsar项目里的类）负责将多个ledger串成一个topic/分区，因此只有ManagedLedgerImpl能知道这些信息，需要在ManagedLedgerImpl里实现这一部分数据的搜集，并在ManagedLedgerImpl调用bookie client创建ledger的时候把这部分数据作为参数传进去。

不过因为给创建Ledger的api增加一个入参的改动量过大，最终并没有实现这一点。

bookie负载均衡特性已经上线到生产环境，并取得了较好的效果，在集群流量压力没有变化的情况下，P99写延迟从秒级甚至几十秒，下降到1s以内。我们已经将代码贡献给了社区，感兴趣的读者可以自行取用：https://github.com/apache/bookkeeper/issues/4247

要权衡一个负载均衡算法的好坏，比较直接的方式就是看我们关心的指标，具体到这里就是P99写入延迟，更深入一点，我们可以看集群整体的负载分布情况，具体到这里就是journal盘写入吞吐标准差、journal盘写入吞吐极差、journal盘写入吞吐top 10。

下面一一看这些指标的前后对比（于1.10上线）：

·P99写延迟

可以看到，写延迟从峰值十多二十秒，显著下降到1s以内。

·journal盘写入吞吐标准差

journal盘写吞吐标准差峰值从25MB/s降到21MB/s。

·journal盘写入吞吐极差

写吞吐极差峰值从150MB/s降到130MB/s。

·journal盘写入吞吐top 10

BIGO为Bookie加入负载均衡机制获得了显著的收益，不再需要集群扩容来缓解部分超载节点的压力，显著降低了成本。而且增加机器来扩容也不一定能解决延迟高的问题，因为问题本质不是资源不够，而是负载不均衡。

为Bookie加入负载均衡机制会被挑战：这不是一个小特性开发，社区没有做这个事情，我们能不能做？做了有没有收益？在经过详尽的调研与分析，基于充分的数据事实面前，这些问题都能一一解答，并最终取得预期内的成果。

      ▼版权声明