Kafka 技术在 B 站的探索与实现

Apache Kafka 是一个分布式数据流处理平台，可以实时发布、订阅、存储和处理数据流。它旨在处理多种来源的数据流，并将它们交付给多个消费者。简而言之，它可以移动大量数据，不仅是从 A 点移到 B 点，而是能从 A 到 Z 的多个点移到任何您想要的位置，并且可以同时进行。

Apache Kafka 可以取代传统的企业级消息传递系统。它最初是 Linkedin 为处理每天 1.4 万亿条消息而开发的一个内部系统，现已成为应用于各式各样企业需求的开源数据流处理解决方案。

背景、挑战、痛点

B 站是当今国内最火的 AGC 网站，其用户群体以年轻人为主。B 站拥有 1000+ 台 Kafka 机器，组成了 20 多个集群。每天，这些 Kafka 端点都会在公司的各个部门之间上报、暂存、分发各种数据，输入 PB 级数据，输出数十 PB。随着集群规模的扩张，B站遇到了越来越多的挑战。

• 客户端读写方式多样、难以检测，难以协调集群稳定性和资源利用率。同时，过密的读写操作引发大量的磁盘 I/O，影响用户读写效率。
• 集群多业务共用，核心业务和普通业务互相影响，出现问题时受波及的地方过广。
• 开源版本的限速粒度很粗，不灵活，难以实时根据磁盘状态进行对应调整。
• Kafka 以集群稳定性著称，但这是以繁琐的上下线流程为代价的，导致效率低下。
• 开源版本在分配 Partition 时仅考虑机器 Partition 数量，不考虑磁盘流量负载情况，也不考虑 Topic 之间的差异，导致集群中机器间、磁盘间负载不均衡。
• 随着公司业务不断扩大，一套 IDC 不足以支撑整个公司的服务，需要控制并协调多个 IDC。
• Kafka 只有一个工作线程池，慢请求可能导致线程池阻塞，影响其他请求处理效率。

Guardian - Kafka federation cluster controller

为了应对这些挑战，哔哩哔哩开发了一套自动化治理系统，有效地解决目前面临的问题。Guardian 是一套自研的 Kafka federation cluster controller。该服务通过 Raft 保证了高可用和一致性，并支持从 Kafka Server 端收集各类数据进行计算和分析，执行治理计划。包含以下功能：

 1. 元数据管理与集群。 2. 元数据管理与存储。 3. uuid(topic, segmentID) 的分配。 4. 收集集群信息进行调度。 5. 多租户管理与 label 隔离。 6. 故障预警与自愈。

基于 JMX 协议采集 Metrics 的性能非常差，这是因为 jmx 协议一个请求只能获取一个 mbean。随着 metric 的加强，可能达到万级别，此时 cpu 消耗会占 20% 甚至更高。Kafka Reporter 为基于 GRPC，http 协议的内置 Metric 上报服务，只需要一个 rpc 即可拉取全部监控数据。

集群层面治理

Partition 级别限速保护

Kafka 是一个 I/O 密集型的服务，用户行为不可预测且多变。当用户读取最新数据时，能够从 Page Cache 中高效快速获取，但如果需要从磁盘中读取数据，那么就要考虑如何控制磁盘 I/O 和磁盘资源的使用，以便为用户提供最大的吞吐量。开源的限速方案存在的问题是粗粒度，比如限制某 ClientId 的读速度为 5MB/s，那么此 Client 在某台机器下读取所有订阅的 Partition 的速度的总和被限制到 5MB/s 左右，无法精确限制到 Partition。

为了让磁盘合理且更充分地利用，需要为 Kafka 新增 Partition 粒度的限速逻辑。通过管控系统实时监测每个磁盘的 ioutil 和 Latency 等关键指标，一旦发现某块磁盘超过了设定的阈值，就会判断该磁盘的健康度下降，需要进行优化处理。此时，尝试对该磁盘下的 io 操作进行限速，使该磁盘的 ioutil 和 Latency 恢复到合理水平。为此，B 站团队设计了一种估算算法：

1. 用可用内存大小除以磁盘读写的速率和，粗略估算出一段数据能在 PageCache 里面存在的时间 T。
2. 使用 Partition 的 MessageInRate 乘以 T 估算出此 Partition 能在 PageCache 里面缓存数据条数。
3. 然后计算 Partition LEO - MessageInRate * T。若其大于要拉取数据的 Offset，那么认为此数据是实际读磁盘数据。

• Guardian 基于集群监控数据进行集群磁盘的健康度检查，根据每一个分区的监控数据进行根因分析，根据分析结果及时进行限速调整。
• 磁盘 IO 相关行为分为六种：用户读/写磁盘，主从同步读/写磁盘和磁盘间迁移读/写磁盘。
• 异常行为：超过预期的写磁盘，任何读磁盘。
• 将所有异常行为排序为 异常行为队列。异常行为队列排序方式为当前流量大到小排优先级。
• 在实际使用中，根据集群状况实时进行自动限速保护的功能对Kafka集群的稳定性做出了很大贡献。

自动 Partition 均衡

为了解决开源 Kafka 的负载分配不均匀导致的磁盘热点问题，哔哩哔哩开发了一种基于磁盘指标、集群 Topic 分布情况和每个 Partition 流量指标的 Partition 自动均衡迁移计划功能。

1. 负载分析和迁移计划生成：根据采集到的数据，识别高负载机器及需要进行负载均衡的待迁移 Partition。生成迁移计划时，考虑目标磁盘的流量负载，按磁盘历史流量负载的中位数排序，选择负载最小的磁盘作为迁移目标。
2. 增量提交：在集群执行均衡迁移计划时，不同 Partition 所承载的流量不同，所在的机器负载不同，搬迁所需要的时间也会不同。采取增量提交均衡任务，让耗时较长的任务不会阻塞其他任务，保持高效执行。
3. 动态调整搬迁速度：根据集群内机器的负载动态调整搬迁的速度，让搬迁任务不会影响到集群的稳定性和用户的使用。
4. 多并发搬迁：支持不同集群不同 Partition 并发的执行搬迁计划，并可控制并发度。
5. Partition 预分配：新建 Topic 时，根据当前磁盘负载和该 Topic 的预期估值流量进行计算，生成该 Topic 所有 Partition 的预分配计划。
6. Leader 均衡：为了避免 Partition Leader 机器严重不均衡造成的机器热点问题，自动生成均衡计划进行 Partition Leader 均衡。
7. 异常熔断：当某个节点异常，搬迁计划无法顺利执行完成时或当流量上涨需要进行 Partition 扩容时，自动取消当前所影响的搬迁计划。

多租户资源隔离管理

由于各业务场景及优先级不同，需要尽可能避免互相影响及互相争抢资源的状况，同时还要避免集群数量膨胀带来的运维和使用问题。由此提出 Kafka 多租户方案，提高对共享资源的管控，对高优流提供更好的隔离性。

• 资源独占

1. Topic 创建
   相同业务域的 Topic 共享独占资源，用户在创建 Topic 的时候指定对应的独占资源即可，该 Topic 仅会在独占资源上创建。支持动态调整，任务无感。
2. 机器划分
   a.新的独占资源: 从资源池划所需机器。
   b.已存在独占资源: 检查已有资源是否满足新需求，不满足时按需扩容。
   c.独占资源缩容: 选择机器，将对应机器上的所有 Partition 迁移到剩余的机器后，机器归还资源池。
3. 支持自动限速保护、自动均衡搬迁功能。

多机房管理

• 随着公司的业务拓展，单个 IDC 不足以支持需求，因此系统需要支持跨 IDC 管理。
• 业务迁移 IDC 后，秉承着就近原则，对应的数据也需要迁移 IDC。B 站的系统支持使用方无感的情况下，一键完成不同 IDC 间的 Topic 迁移。
  1. 按需配置副本放置策略，系统根据所填配置自动生成迁移计划，在迁移的过程中，有限速模块保证集群的稳定性，迁移过程中无需人工干预。
  2. 配置粒度可控，从集群级别到 topic 级别均可。
  3. Topic 在迁移过程中，保持一致的 offset，期间用户读写均无感。

IDC 迁移

用户请求时返回就近的副本

• 一般相关的业务在同一 IDC 中，但是也会有多个任务分布在不同机房。B 站支持 IDC 感知的从读功能，用户请求时返回就近的副本，在同 IDC 读取。
• 此外，还支持通过上报数据分析，实时计算专线消耗，保护专线。

请求队列拆分

一般物理机上有多块磁盘，有较大的内存和较大的 CPU。不同公司 Kafka 的部署方式不太一样，有的选择部署多个实例，一个实例一块磁盘，有的则是部署一个实例，利用 Kafka支持多盘的特性。两种方案各有利弊，B 站采用的是后者。一台物理机上有 10+HDD 盘。一个实例上有上千 Partition，上万个连接。如果其中一块磁盘 Fail-Slow，或负载特别高，磁盘上的请求耗时会增长很多。假设请求均匀，受影响比例理应仅为 1/14=7%，但实际会有 100% 影响。B 站的优化基于以下：

• 每个 client实例（连接）生产特定少数 Topic，消费特定少数 Partition。
• 如果能做到线程隔离，保证所有的慢请求与正常请求不共享线程，则正常请求不会收到影响。
• 慢请求总是集中在部分 Partition 上，且总是少数。

1. ChannelMarker，监控每个 Channel（连接）的 RequestHandler 线程耗时及 Processor 线程耗时并打分，标记较慢的 channel。
2. RequestHandler 根据 requestType 拆分线程池，Fetch，Produce，Default。并且新增对应的 SlowRequest 的线程池。之前标记的 channel 请求会投递到对应的 slowRequest 线程池中，保证线程隔离。
3. 如果 ChannelMarker 不能很好分类，该算法近似于原本的线程模型。另外将其他请求与读写线程池拆分，能尽可能的避免 Controller 与 Admin 请求超时。

Tired-Storage

Kafka 的 Partition 和磁盘上的目录一一对应，因此有以下问题:

1. 个别磁盘读写量过大，会出现磁盘热点，受限于磁盘性能上限。
2. 迁移需要同步完整数据，导致迁移时间非常长，需要小时级甚至天级，提前治理要求极高。
3. Partition 不支持读写资源分离，读写磁盘会互相影响。
4. Partition 的总数据量取决于磁盘的容量上限，无法满足 1 周及以上的需求。
5. 任务间的写入消费速度不同，数据过期时间不同和峰谷流量情况不同，导致难以合理规划。

• 目标是将 Kafka 的 Partition 和磁盘强绑定关系打破，允许同一个 Partition 的数据分布在不同 Storage 层上，并且采用 HDFS 作为 Remote Storage。
• 为此，B 站设计了分层存储的 meta 服务，基于 Raft 实现 HA。
• 支持基于不同策略的 offset 拉取模式。
• 修改协议支持批量，不再是写一个 Segment 写一次 Meta，增加了 Leader Fence 保证强一致。
• Segment 下载到本地 Cache 后读取，HDFS 1 segment/per read, 大幅降低 HDFS 负担和时延要求。
• 减少写 RT 抖动，降低写延迟，因为 Kafka Broker 不再需要存储全部数据了，SSD 磁盘的容量限制不再是瓶颈，而 SSD 的低延时却有极大的优势。

Kafka 审计功能

公司内部使用 Kafka 的部门非常多，一个 Topic 经常多个用户在使用，无法精准定位到该 Topic 的上下游使用者，日常运维难以治理和管控。

• 对 Kafka 进行了改造，增加了审计功能，可以提供生产、消费等一系列请求的详细数据，并将这些数据实时写入到 Clickhouse 中，方便问题排查。
• 支持查询某个时间切片内的机器和集群的详细情况。
• 基于审计数据，实现了成本管理系统，可以有效地治理一些冗余的 Topic，达到降本增效的目的。
• 利用 Kafka 的审计功能，辅助修正了历史上混乱的 Topic 权限，提升了数据的安全性。

运维层面治理

集群平滑发布

随着集群规模不断地扩大，如何快速平滑的发布成为一个越来越大的难题。

• 开发了一种能够自动批量上下线机器的服务，用户只需选择目标集群，服务就会根据集群的实时状态，在不影响集群服务稳定性的情况下，自动完成机器的上下线操作。整个过程无需人工干预。
• 机器下线
  • 当某台机器需要下线时，服务会先将该机器作为 leader 的所有 Partition 进行迁移，确保 Partition 的可用性不受影响。
• 机器上线
  • 当某台机器恢复正常并重新提供服务时，服务会自动将该机器原先负责的所有 Partition 恢复到原来的状态。

结果：极大节约人力成本，Kafka 滚动升级从 15 人天缩短到 1人小时，提单后自动化、搬迁完全自动化，降低对业务的影响。

未来展望

• 支持分钟级调度。目前迁移的耗时一般在小时级别到天级，下一步是将迁移任务进一步提速，使任务在分钟级完成。
• 支持分钟级自检自愈系统。目前硬件故障需要人为操作移除故障节点，下一步目标是全自动分钟级摘除。
• 支持动态扩缩容。目前集群扩缩容完全依赖人为操作，未来集群的扩缩容将完全自动化，服务根据集群实时状态进行分析后自动进行机器扩缩容。