Kafka 在华泰证券的探索与实践

（1）高可用双活架构
如图 3 所示，Kafka 高可用特性依赖于 zookeeper 来实现，由于 zookeeper 的 paxos 算法特性，故 zookeeper 采用同城三中心部署方式，保证 zookeeper 本身高可用，通常其中两个数据中心部署偶数台机器，另一数据中心部署单台机器。
Kafkabroker 跨数据中心两节点部署，所有 topic 的 partition 保证在两中心都有副本。如果单数据中心出现问题，另一个中心能自动进行接管，业务系统可以无感知切换。
由于Kafka的高带宽需求，主机采用万兆网卡，并且在网卡做 bond0 以保证网卡高可用，跨数据中心之间的网络通信采用独立的万兆波分通道。

（2）参数调优
• 首先我们在 JVM 层面做了很多尝试。对 Kafka 服务启动参数进行调优，使用 G1 回收器。kafka 内存配置一般选择 64G，其中 16G 给 Kafka 应用本身，剩余内存全部用于操作系统本身的 page cache.
• 此外为了保证核心系统的达到最佳的读写效果，我们采用 SSD 硬盘，并做了 raid5 冗余,来保证硬盘的高效 IO 读写能力。
• 其次我们通过调整 broker 的 num.io.threads，num.network.threads, num.replica.fetchers 等参数来保证集群之间快速复制和吞吐。
（3）数据一致性保证
Kafka 有自己一套独特的消息传输保障机制(at least once)。当 producer 向 broker 发送消息时，由于副本机制（replication）的存在，一旦这条消息被 broker 确认，它将不会丢失。但如果 producer 发送数据给 broker 后，遇到网络问题而造成通信中断，那 producer 就无法判断该条消息是否已经被确认。这时 producer 可以重试，确保消息已经被 broker 确认，为了保证消息的可靠性，我们要求业务做到：

• 保证发送端成功
当 producer 向 leader 发送数据时，可以通过 request.required.acks 参数来设置数据可靠性的级别：

• 保证消费者消费成功(at least once)
我们要求消费者关闭自动提交(enable.auto.commit:false)，同时当消费者每次 poll 处理完业务逻辑后必须完成手动同步提交（commitSync），如果消费者在消费过程中发生 crash，下次启动时依然会从之前的位置开始消费，从而保证每次提交的内容都能被消费。

• 消息去重
考虑到 producer,broker,consumer 之间都有可能造成消息重复，所以我们要求接收端需要支持消息去重的功能，最好借助业务消息本身的幂等性来做。其中有些大数据组件，如 hbase，elasticsearch 天然就支持幂等操作。

场景事例： 行情数据 hbase 存储
在华泰内部使用 kafka 来缓存一段时间的行情数据，并做相应处理为了保证 kafka 中数据的完整性，发送端API参数配置:

在接收端，启动专门的消费者拉取 kafka 数据存入 hbase。hbase 的 rowkey 的设计主要包括 SecurityId（股票id）和 timestamp（行情数据时间）。消费线程从 kafka 拉取数据后反序列化，然后批量插入 hbase，只有插入成功后才往 kafka 中持久化 offset。这样的好处是，如果在中间任意一个阶段发生报错，程序恢复后都会从上一次持久化 offset 的位置开始消费数据，而不会造成数据丢失。如果中途有重复消费的数据，则插入 hbase 的 rowkey 是相同的，数据只会覆盖不会重复，最终达到数据一致。
所以，从根本上说，kafka 上的数据传输也是 数据最终一致性的典型场景。

（4）ACL安全
目前华泰内部通过配置 allow.everyone.if.no.acl.found参数（:true）让 Kafka 集群同时具备ACL和非ACL的能力，避免资源的浪费。我们选用 SASL 作为 Kafka 鉴权方式，因为 SASL 虽然简单，但已满足需求，而 Kerberos 使用过重，过度复杂组件会给 Kafka 带来更多不确定的因素，如示例所示，根据部门划分来分配用户。
示例：
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
ser_dep1=“ password 1”
user_dep2=“ password 2”
user_dep3=“ password 3”;
};
服务启动后，通过 Kafka 的 command line 接口，配置基于用户、ip、topic、groupid 等的 acl 权限来保证各业务之间的隔离。