kafka集群为什么需要三个节点_每天处理千亿级日志量，Kafka是如何做到的？

之前为大家分享了不少 Kafka 原理解析类的干货&＃xff0c;今天咱们一起来看看 360 基于 Kafka 千亿级数据量的深度实践!

图片来自 Pexels

本文主要围绕如下内容分享&＃xff1a;

• 消息队列选型
• Kafka 在 360 商业化的现状
• Kafka Client 框架
• 数据高可用
• 负载均衡
• 鉴权、授权与 ACL 方案
• Quota 机制
• 跨 IDC 的数据同步
• 监控告警
• 线上问题及解决方案

消息队列选型

当时主要考虑以下几个维度&＃xff1a;

• 社区活跃度
• 客户端支持
• 吞吐量

对比几个系统下来&＃xff0c;觉得 Kafka 比较符合我们的要求。现在有一个新的开源系统 Pulsar&＃xff0c;我觉得也可以尝试一下。

Kafka 设计上的亮点如下&＃xff1a;

Kafka 性能和吞吐都很高&＃xff0c;通过 Sendfile 和 Pagecache 来实现 Zero Copy 机制&＃xff0c;顺序读写的特性使得用普通磁盘就可以做到很大的吞吐&＃xff0c;相对来说性价比比较高。

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

Kafka 集群有两个管理角色&＃xff1a;

• Controller 主要是做集群的管理。
• Coordinator 主要做业务级别的管理。

这两种角色都由 Kafka 里面的某个 Broker 来担任&＃xff0c;这样 Failover 就很简单&＃xff0c;只需要选一个 Broker 来替代即可。

从这个角度来说 Kafka 有一个去中心化的设计思想在里面&＃xff0c; 但 Controller 本身也是一个瓶颈&＃xff0c;可以类比于 Hadoop 的 Namenode。

CAP 理论相信大家都有了解过&＃xff0c;分布式系统实现要么是 CP&＃xff0c;要么是 AP。

Kafka 实现比较灵活&＃xff0c;不同业务可以根据自身业务特点来对 Topic 级别做偏 CP 或偏 AP 的配置。

支持业务间独立重复消费&＃xff0c;并且可以做回放。

这个是 Kafka 的简要架构&＃xff0c;主要分为&＃xff1a;

• 生产端
• Broker 端
• 消费端

日志有三个层次&＃xff1a;

• 第一个层次 Topic
• 第二个层次 Partition(每个 Partition 是一个并行度)
• 第三个层次 Replica(Replica 表示 Partition 的副本数)

Kafka 在 360 商业化的现状

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

我们的物理机配置 24Core/10G 网卡/128G 内存/4T*12 HDD&＃xff0c;值得说一下的是我们采用了万兆网卡加普通磁盘 4T*12 的配置&＃xff0c;测下来磁盘吞吐和网络吞吐是能够匹配上的。

再者考虑到我们的数据量比较大&＃xff0c;SSD 盘没有特别大的且成本比较高。

磁盘的组织结构我们用的是 JBOD&＃xff0c;RAID10 也是很好的方案(磁盘成本会翻倍)。

我们目前的 Kafka 版本是 1.1.1&＃xff0c;推荐大家部署 0.11 以上的版本会好一些&＃xff0c;这个版本对协议做了很多优化&＃xff0c;对于后续的 2.x 版本都是兼容的。

这个是我们 Kafka 上下游相关的组件&＃xff0c;生产端主要是各种 Kafka Clients/实时服务/Flume/Logstash。

消费端分为实时&＃xff0c;离线(ETL)&＃xff0c;监控三部分。实时有 Spark/Flink/Storm 等主流框架, 离线部分我们基于 Flink 自研了一个统一落地框架 Hamal&＃xff0c;从 Kafka 消费一遍数据就可以落地到多个下游系统(HDFS、Hbase、Redis等)&＃xff0c;可以避免重复消费。

还有部分是监控的需求&＃xff0c;我们把 ES/InfluxDB 相关的日志打到 Kafka&＃xff0c;然后再消费出来通过 Grafana 展示&＃xff0c;但目前我们已经切到 Prometheus 上了。

Kafka Client 框架

为什么要做这个框架呢?之前有很多的业务部门用裸 API 自己去实现 Kafka Client 的逻辑。

但是会有很多问题&＃xff0c;有一些异常情况会 Catch 不全&＃xff0c;我们做这个框架是想把所有的细节屏蔽掉&＃xff0c;然后暴露出足够简单的接口。

这样可以减少业务犯错的可能性&＃xff0c;我们要确保极端的情况下比如网络或集群异常时的可用性&＃xff0c;如果网络或集群不可用&＃xff0c;数据会先落到本地&＃xff0c;等恢复的时候再从本地磁盘恢复到 Kafka 中。

我们实现了两个框架&＃xff1a;

• LogProducer&＃xff0c;支持 at least once。
• LogConsumer&＃xff0c;支持 at least once 和 exactly once 两种语意&＃xff0c;其中 exactly once 需要业务去实现 Rollback 接口。

LogProducer 框架的大体思路是通过内存队列将日志发送到 Kafka&＃xff0c;当 Kafka 或网络不可用的情况下会写本地磁盘&＃xff0c;同时会有一个线程去实时检测 Kafka 或者网络的可用情况&＃xff0c;如果恢复就会加载磁盘日志并发送到 Kafka。

我们还支持一种共享内存的策略来代替内存&＃xff0c;使用共享内存是为了减少重启过程中日志的丢失数。

LogConsumer 的框架实现&＃xff0c;通过 Blocking Queue 将 Consumer 线程和 Worker 线程解耦&＃xff0c;因为现实情况是消费逻辑很简单&＃xff0c;但是处理逻辑会很复杂。

这样就可以对 Consumer 线程和 Worker 线程做不同的配置&＃xff0c;同时通过 Blocking Queue 还可以实现反压机制。

比如 Worker 处理不过来了&＃xff0c;这时候 Blocking Queue 就会满&＃xff0c;反压到 Consumer 线程会停止消费。

同时我们在 Worker 线程接口里面会提供接口让用户提交到 global offsetmap。

如上图我们提供三个组合接口&＃xff0c;如果在业务处理与 Commit 中实现了业务端 Rollback 逻辑&＃xff0c; 那么就是 exactly once 语义&＃xff0c;默认是 at least once 语义。

数据高可用

之前讲过 Kafka 本身提供 Replica&＃43;ISR 的机制来保证数据高可用&＃xff0c;但我们觉得这个可能还不够&＃xff0c;所以我们还要支持 Rack Aware。

比如 Replica&＃61;3 的情况&＃xff0c;确保三个副本在不同的物理 Rack 上&＃xff0c;这样我们最多能容忍两个物理机架同时出问题而数据仍可用&＃xff0c;我们 Rack Aware 方案是与负载均衡方案一起做掉的&＃xff0c;具体后面会讲。

值得注意的是 Kafka 官方也支持 Rack Aware&＃xff0c;通过在 Broker 端配置 broker.rack 参数可实现。

但有一个限制&＃xff0c;必须为每个 Rack 分配数量相同的 Brokers&＃xff0c;否则会导致 Replica 分配倾斜&＃xff0c;实际情况是 IDC 的 Rack 是很多的&＃xff0c;分配到的物理机分布也可能很随机。

一个可以参考的解决思路是采用虚拟 Rack Group 的概念&＃xff0c;比如维护 3 个虚拟 Rack Group&＃xff0c;申请到的物理机加入到这 3 个 Group 中&＃xff0c;并确保 Rack Group 间分配的物理机数量一致。

当然 Rack Group 间物理机不应存在有相同物理 Rack 的情况。

负载均衡

Kafka 的负载均衡功能在 Confluent 商业版本才支持&＃xff0c;负载均衡本质上来说是 Replica 分配均匀问题。

我们一开始想通过经典一致性 Hash 来解决&＃xff0c;如下图&＃xff1a;

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

于是我们基于虚拟节点的一次性 Hash 的算法实现了一个方案&＃xff0c;如图所示&＃xff1a;相同的颜色对应同一个物理机&＃xff0c;Hash 环上的都是虚拟节点。

这里有四个物理节点&＃xff0c;其中 node4 是我们新加的节点。通过虚拟节点可以把物理节点的负载足够均衡地分散出去&＃xff0c;所以当我把 node4 加到 Hash 环上的时候&＃xff0c;分担了所有物理机的负载。

算法实现的步骤分为两个大的步骤&＃xff1a;

①新建 hash circle&＃xff1a;通过 vnode_str(比如 hostname-v0)做一个 MD5 的 Hash&＃xff0c;得到虚拟节点的 vnode_key&＃xff0c;再用 ring 字典来保存虚拟节点到物理节点的映射&＃xff0c;同时将 vnode_key 加入到 sorted_keys 的 list 中。

②在 Hash 环中分配 Replica&＃xff1a;将(topic_name&＃43;partition_num&＃43;replica_num)作为 Key 用相同的 MD5 Hash 算法得到 replica_key。

接着二分查找该 replica_key 在 sorted_keys 中的 Position, 最后用 Ring 字典来映射到物理机 Node&＃xff0c;至此 Replica 分配完成。

我们基于这个算法解决三个问题&＃xff1a;

• 添加物理节点只需迁移很小一部分数据。
• 对不同配置的物理机做权重设置&＃xff0c;可以支持异构集群的部署。
• 实现 Replica 的 Rack Aware&＃xff0c;物理节点上面会有 Rack 信息&＃xff0c;在为 Replica 分配物理节点的时候会记录已经分配的 Rack 信息。

如果有重复的情况&＃xff0c;就会把 vnode_key 找到 Position 的位置 &＃43;1 找下一个物理节点&＃xff0c;我们会确保三个 Replica 的物理 Rack 一定是不一样的(假如 Replica&＃61;3)。

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

Disk Rebalance&＃xff1a;这个 Feature 需要 Kafka1.1.0 版本之后才支持&＃xff0c;Kafka 提供了一些脚本和 API 可以做 Balance 操作, 其本质也是生成 Replica Plan 然后做 Reassign。

鉴权、授权和 ACL 方案

如果是新集群比较推荐基于 SASL 的 SCRAM 方案&＃xff0c;实施起来比较简单。

如果老集群想中途施行鉴权授权机制会比较困难&＃xff0c;需要推各个业务去修改配置&＃xff0c;同时切换的过程也很容易出问题。

下面介绍下我们实现的一个白名单机制来解决老集群的问题&＃xff0c;首先将老业务加入到白名单中&＃xff0c;让新业务通过工单流程来申请 Topics 和 Consumers 两种资源权限并加到白名单里&＃xff0c;定期监测非法(没有走工单)Topics&＃xff0c;Consumers 资源。

同时将这些资源都 Deny 掉&＃xff0c;这样就收紧了 Topics 和 Consumer 读写权限的口子&＃xff0c;同时原有业务不会有任何影响。

Quota 机制

Quota 主要是为了解决多个业务间资源抢占问题。Quota 类型有两种&＃xff1a;

• 一种是限制网络带宽。
• 一种是限制请求速率(限制 CPU)。

我们对业务做了三个优先级设置&＃xff1a;高&＃xff0c;中&＃xff0c;低优先级&＃xff0c;高优先级不做限制&＃xff0c;中优先级可容忍 lag&＃xff0c;低优先级极端情况可停掉&＃xff0c;通过工具可以批量限制某个优先级的所有业务&＃xff0c;可以确保高优先级业务及集群的安全。

跨 IDC 的数据同步

首先我们为什么要做跨 IDC 的数据同步?没做这个同步之前业务可能对数据的读写没有一个 IDC 的概念&＃xff0c;所以很容易就会有跨 IDC 的读写&＃xff0c;多个业务还可能有重复 Consume 和 Produce。

这就造成跨 IDC 网络的极大浪费&＃xff0c; 加上跨 IDC 的网络并不稳定&＃xff0c;有时候会有一些异常&＃xff0c;业务也不一定能很好处理。

为了解决以上问题&＃xff0c;我们统一做了跨 IDC 的数据同步服务&＃xff0c;首先我们约定业务只能做本 IDC 的读写&＃xff0c;不允许做跨 IDC 的读写&＃xff0c;如果有跨 IDC 的数据需求&＃xff0c;要向我们申请&＃xff0c;通过 Mirrormaker 去同步一份过来。

这样做有两个好处&＃xff1a;

• 一是屏蔽了异常对业务的影响。
• 二是节省了 IDC 之间的带宽(我们通过同步机制能保证这份数据只传输一份)。

我们还基于 Marathon/Mesos 对这个服务做了 Pass 化&＃xff0c;提高了服务的 SLA。

监控告警

我们的监控警告平台如下&＃xff1a;

• 基于 Jmx exporter&＃43;Promehteus&＃43;Grafana 来做图表展示&＃xff0c;在每个 Broker 上面部署 Jmx exporter&＃xff0c;Prometheus 会去 Pull 这些数据&＃xff0c;最后通过 Grafana 来展示。
• 基于 Kafka Manager 做瞬态指标的监控。
• 基于 Burrow 做 Consumer lag 的监控。
• 基于 Wonder 来做告警&＃xff0c;这个是 360 内部实现的一个组件&＃xff0c;类似 Zabbix。

线上问题及解决方案

磁盘故障&＃xff1a;我们通过 Smartctl 来监测&＃xff0c;首先状态是要 Passed 的&＃xff0c;其次我们会判断 197 Current_Pending_Sector 这个属性值不能大于 100&＃xff0c; 如果大于 100 这个磁盘可能有读写性能问题。

bootstrap.servers 性能瓶颈&＃xff1a;该参数可以配置多台 Broker&＃xff0c;这些 Broker 作为 Proxy 的角色为 Kafka Clients 提供 Lookup 服务。

如果集群规模很大&＃xff0c;Clients 很多的情况下&＃xff0c;这些 Proxy 角色的 Broker 的负载会很大&＃xff0c;为了解决这个问题&＃xff0c;我们对 bootstrap.servers 参数做了 VIP 配置。

每个 VIP 可以绑定任意多的 Brokers&＃xff0c;这样在客户端不需要修改配置的情况下可以对 Proxy 动态扩缩容。

Consumer 重启不消费&＃xff1a;业务反馈消费停止&＃xff0c;重启也不能够解决问题&＃xff0c;后来定位发现是早于 0.11 之前版本的 Bug&＃xff1a;

https://issues.apache.org/jira/browse/KAFKA-5413 

原因是 log cleaner 线程挂了导致 Compact 停止&＃xff0c;__consumer_offsets 这个 Topic 的量非常大&＃xff0c;broker reload 时间特别长&＃xff0c;这段时间是停止服务的。

解决方法有两个&＃xff1a;

• 一是升级到 Kafka 0.11&＃43; 版本
• 二是将 Offset 迁移到新的 Consumer Group 来解决(规避掉有问题的 Coordinator)。

严锁鹏

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