千亿级数据量的 Kafka 深度实践

本文由 dbaplus 社群授权转载。

一、消息队列选型

当时主要考虑以下几个维度：社区活跃度，客户端支持，吞吐量。对比几个系统下来，觉得 Kafka 比较符合我们的要求。现在有一个新的开源系统 pulsar，我觉得也可以尝试一下。

1、Kafka 设计上的亮点

Kafka 性能和吞吐都很高，通过 sendfile 和 pagecache 来实现 zero copy 机制，顺序读写的特性使得用普通磁盘就可以做到很大的吞吐，相对来说性价比比较高。

Kafka 通过 replica 和 isr 机制来保证数据的高可用。

Kafka 集群有两个管理角色：controller 主要是做集群的管理；coordinator 主要做业务级别的管理。这两种角色都由 Kafka 里面的某个 broker 来担任，这样 failover 就很简单，只需要选一个 broker 来替代即可，从这个角度来说 Kafka 有一个去中心化的设计思想在里面， 但 controller 本身也是一个瓶颈，可以类比于 hadoop 的 namenode。

CAP 理论相信大家都有了解过，分布式系统实现要么是 CP，要么是 AP。Kafka 实现比较灵活，不同业务可以根据自身业务特点来对 topic 级别做偏 CP 或偏 AP 的配置。

支持业务间独立重复消费，并且可以做回放。

这个是 Kafka 的简要架构，主要分为生产端，broker 端，还有消费端。日志有三个层次：

• 第一个层次 topic；

• 第二个层次 partition（每个 partition 是一个并行度）；

• 第三个层次 replica（replica 表示 partition 的副本数）。

二、Kafka 在 360 商业化的现状

目前集群有千亿级数据量，集群有 100 多台万兆机器，单 topic 的最大峰值 60 万 QPS，集群的峰值大概在 500 万 QPS。

我们的物理机配置 24Core/10G 网卡/128G 内存/4T*12 HDD，值得说一下的是我们采用了万兆网卡加普通磁盘 4T*12 的配置，测下来磁盘吞吐和网络吞吐是能够匹配上的， 再者考虑到我们的数据量比较大，SSD 盘没有特别大的且成本比较高。

磁盘的组织结构我们用的是 JBOD，RAID10 也是很好的方案(磁盘成本会翻倍)。我们目前的 Kafka 版本是 1.1.1，推荐大家部署 0.11 以上的版本会好一些，这个版本对协议做了很多优化，对于后续的 2.x 版本都是兼容的。

这个是我们 Kafka 上下游相关的组件，生产端主要是各种 Kafka clients/实时服务/flume/logstash。

消费端分为实时，离线（ETL），监控三部分。实时有 spark/flink/storm 等主流框架, 离线部分我们基于 flink 自研了一个统一落地框架 hamal，从 Kafka 消费一遍数据就可以落地到多个下游系统(hdfs、hbase、redis 等)，可以避免重复消费。还有部分是监控的需求，我们把 ES/influxdb 相关的日志打到 Kafka，然后再消费出来通过 grafana 展示，但目前我们已经切到 prometheus 上了。

三、Kafka client 框架

为什么要做这个框架呢？之前有很多的业务部门用裸 API 自己去实现 Kafka client 的逻辑，但是会有很多问题，有一些异常情况会 catch 不全，我们做这个框架是想把所有的细节屏蔽掉，然后暴露出足够简单的接口，这样可以减少业务犯错的可能性，我们要确保极端的情况下比如网络或集群异常时的可用性，如果网络或集群不可用，数据会先落到本地，等恢复的时候再从本地磁盘恢复到 Kafka 中。

我们实现了两个框架：LogProducer 和 LogConsumer。LogProducer 支持 at least once；LogConsumer 支持 at least once 和 exactly once 两种语意，其中 exactly once 需要业务去实现 rollback 接口。

LogProducer 框架的大体思路是通过内存队列将日志发送到 Kafka，当 Kafka 或网络不可用的情况下会写本地磁盘，同时会有一个线程去实时检测 Kafka 或者网络的可用情况，如果恢复就会加载磁盘日志并发送到 Kafka。我们还支持一种共享内存的策略来代替内存，使用共享内存是为了减少重启过程中日志的丢失数。

LogConsumer 的框架实现，通过 blocking queue 将 consumer 线程和 worker 线程解耦，因为现实情况是消费逻辑很简单，但是处理逻辑会很复杂。这样就可以对 consumer 线程和 worker 线程做不同的配置，同时通过 blocking queue 还可以实现反压机制。比如 worker 处理不过来了，这时候 blocking queue 就会满，反压到 consumer 线程会停止消费。

同时我们在 worker 线程接口里面会提供接口让用户提交到 global offsetmap, 如上图我们提供三个组合接口，如果在业务处理与 commit 中实现了业务端 rollback 逻辑， 那么就是 exactly once 语义，默认是 at least once 语义。

四、数据高可用

之前讲过 Kafka 本身提供 replica+isr 的机制来保证数据高可用，但我们觉得这个可能还不够，所以我们还要支持 rack aware。比如 replica=3 的情况，确保三个副本在不同的物理 rack 上，这样我们最多能容忍两个物理机架同时出问题而数据仍可用，我们 rack aware 方案是与负载均衡方案一起做掉的，具体后面会讲。

值得注意的是 Kafka 官方也支持 rack aware，通过在 broker 端配置 broker.rack 参数可实现，但有一个限制，必须为每个 rack 分配数量相同的 brokers，否则会导致 replica 分配倾斜，实际情况是 IDC 的 rack 是很多的，分配到的物理机分布也可能很随机，一个可以参考的解决思路是采用虚拟 rack group 的概念，比如维护 3 个虚拟 rack group，申请到的物理机加入到这 3 个 group 中，并确保 rack group 间分配的物理机数量一致，当然 rack group 间物理机不应存在有相同物理 rack 的情况。

五、负载均衡

Kafka 的负载均衡功能在 confluent 商业版本才支持，负载均衡本质上来说是 replica 分配均匀问题，我们一开始想通过经典一致性 hash 来解决如下图：

然后我们发现经典一次性 hash 不能满足我们的需求，比如要加一个节点 node5，只能分担节点 node2 的部分负载，不能做全局节点的负载均衡

于是我们基于虚拟节点的一次性 hash 的算法实现了一个方案，如图所示：相同的颜色对应同一个物理机，hash 环上的都是虚拟节点。这里有四个物理节点，其中 node4 是我们新加的节点。通过虚拟节点可以把物理节点的负载足够均衡地分散出去，所以当我把 node4 加到 hash 环上的时候，分担了所有物理机的负载。

算法实现的步骤分为两个大的步骤:

1. 新建 hash circle：通过 vnode_str（比如 hostname-v0）做一个 MD5 的 hash，得到虚拟节点的 vnode_key，再用 ring 字典来保存虚拟节点到物理节点的映射，同时将 vnode_key 加入到 sorted_keys 的 list 中。

2. 在 hash 环中分配 replica: 将(topic_name + partition_num + replica_num)作为 key 用相同的 MD5 hash 算法得到 replica_key, 接着二分查找该 replica_key 在 sorted_keys 中的 position, 最后用 ring 字典来映射到物理机 node, 至此 replica 分配完成。

我们基于这个算法解决三个问题：

1）添加物理节点只需迁移很小一部分数据；

2）对不同配置的物理机做权重设置，可以支持异构集群的部署；

3）实现 replica 的 rack aware，物理节点上面会有 rack 信息，在为 replica 分配物理节点的时候会记录已经分配的 rack 信息，如果有重复的情况，就会把 vnode_key 找到 position 的位置+1 找下一个物理节点，我们会确保三个 replica 的物理 rack 一定是不一样的（假如 replica=3）。

Leader balance

这是一种快速且成本低的负载 balance 方法，因为 Kafka 只有 leader 提供读写，所以通过 leader 切换是可以达到负载切换的效果的，由于只是 leader 切换不涉及数据同步，因此这个代价是比较小的。

disk rebalance

这个 feature 需要 Kafka1.1.0 版本之后才支持，Kafka 提供了一些脚本和 API 可以做 balance 操作, 其本质也是生成 replica plan 然后做 reassign。

六、鉴权、授权和 ACL 方案

如果是新集群比较推荐基于 SASL 的 SCRAM 方案，实施起来比较简单。如果老集群想中途施行鉴权授权机制会比较困难，需要推各个业务去修改配置，同时切换的过程也很容易出问题。

下面介绍下我们实现的一个白名单机制来解决老集群的问题，首先将老业务加入到白名单中，让新业务通过工单流程来申请 topics 和 consumers 两种资源权限并加到白名单里，定期监测非法(没有走工单)topics，consumers 资源，同时将这些资源都 deny 掉，这样就收紧了 topics 和 consumer 读写权限的口子，同时原有业务不会有任何影响。

七、Quota 机制

Quota 主要是为了解决多个业务间资源抢占问题。Quota 类型有两种：一种是限制网络带宽，一种是限制请求速率(限制 CPU)。我们对业务做了三个优先级设置：高，中，低优先级，高优先级不做限制，中优先级可容忍 lag，低优先级极端情况可停掉，通过工具可以批量限制某个优先级的所有业务，可以确保高优先级业务及集群的安全。

八、跨 IDC 的数据同步

首先我们为什么要做跨 IDC 的数据同步？没做这个同步之前业务可能对数据的读写没有一个 IDC 的概念，所以很容易就会有跨 IDC 的读写，多个业务还可能有重复 consume 和 produce，这就造成跨 IDC 网络的极大浪费， 加上跨 IDC 的网络并不稳定，有时候会有一些异常，业务也不一定能很好处理。

为了解决以上问题，我们统一做了跨 IDC 的数据同步服务，首先我们约定业务只能做本 IDC 的读写，不允许做跨 IDC 的读写，如果有跨 IDC 的数据需求，要向我们申请，通过 mirrormaker 去同步一份过来。这样做有两个好处：一是屏蔽了异常对业务的影响，二是节省了 IDC 之间的带宽（我们通过同步机制能保证这份数据只传输一份），我们还基于 marathon/mesos 对这个服务做了 pass 化，提高了服务的 SLA。

九、监控告警

• 基于 jmx exporter+promehteus+grafana 来做图表展示，在每个 broker 上面部署 jmx exporter, prometheus 会去 pull 这些数据，最后通过 grafana 来展示。

• 基于 Kafka manager 做瞬态指标的监控。

• 基于 burrow 做 consumer lag 的监控。

• 基于 wonder 来做告警，这个是 360 内部实现的一个组件，类似 zabbix。

十、线上问题及解决方案

磁盘故障：

我们通过 smartctl 来监测，首先状态是要 passed 的，其次我们会判断 197 Current_Pending_Sector 这个属性值不能大于 100， 如果大于 100 这个磁盘可能有读写性能问题。

bootstrap.servers 性能瓶颈：

该参数可以配置多台 broker，这些 broker 作为 proxy 的角色为 Kafka clients 提供 lookup 服务，如果集群规模很大，clients 很多的情况下，这些 proxy 角色的 broker 的负载会很大，为了解决这个问题，我们对 bootstrap.servers 参数做了 vip 配置。每个 VIP 可以绑定任意多的 brokers，这样在客户端不需要修改配置的情况下可以对 proxy 动态扩缩容。

consumer 重启不消费：

业务反馈消费停止，重启也不能够解决问题，后来定位发现是早于 0.11 之前版本的 bug, https://issues.apache.org/jira/browse/KAFKA-5413

原因是 log cleaner 线程挂了导致 compact 停止，__consumer_offsets 这个 topic 的量非常大，broker reload 时间特别长，这段时间是停止服务的。

解决方法有两个：一是升级到 Kafka 0.11+版本，二是将 offset 迁移到新的 consumer group 来解决(规避掉有问题的 coordinator)。

Q&A

Q1：hamal 落地系统是消费一次落地到多个组件还是消费了多次？

A：消费一次落地到多个下游组件。

Q2：在 LogProducer 实现中将数据存在共享内存里，这样不会丢数据，我想详细听一下。

A：不是说不丢数据，而是尽可能少丢数据，当选用共享内存策略，业务进程挂掉不会影响共享内存中的数据，重启的时候直接从共享内存恢复。

Q3：这边是通过白名单机制做了一个权限控制吗？

A：通过白名单机制我们对 topic, consumer 资源做了粗粒度的控制，这样可以在不影响老业务的情况下收紧口子。如果是新集群从头搭建的话推荐用 SASL 的 SCRAM 方案。

Q4：你刚才说的 quota 优先级别，具体实现是怎么做的？

A：在业务接入过程当中我们会给业务定级，比如这个业务是计费的，那么就是高优先级，如果只是一些 track 日志那么就是低优先级， 在设置 quota 的时候我们会根据业务当前峰值再加上一定比例 buffer 来设置业务的 quota 值。

Q5：如果集群有 100 个节点，客户端要配 100 个地址吗？

A：不需要，只需配置 bootstrap.servers(proxy)，proxy 可以拿到所有 broker 的信息，它的主要工作是 lookup，接收 client 请求返回 broker 地址列表，然后 client 再直连 broker。

Q6：如果出现磁盘挂载不上这种情况下，broker 节点可以正常拉起来吗？能恢复到从前那个状态吗？

A：broker 可以将这块磁盘对应的目录 exclude 掉，然后重启就可以了，对于 replica=1 的 topic 数据会有丢失，对于 replica>1 的 topic 数据不会有丢失，因为我们做了 rack aware，那么其他 rack 上会有副本。

作者介绍：

严锁鹏，奇虎 360 大数据架构运维专家，具有 10 年基础架构与大数据开发经验。2013 年加入 360 商业化团队，负责消息中间件开发与运维，同时涉及大数据架构、微服务架构、实时计算平台、机器学习平台、监控系统等基础设施建设，致力于为商业化团队提供稳定高效的基础服务。

原文链接：

https://mp.weixin.qq.com/s/5p1IgayVXvCSLLc0Zvoqew