Kafka实践、升级和新版本（0.10）特性预研

本文来自于网易云社区

一、消息总线MQ和Kafka （挡在请求的第一线）

1. 几个应用场景

case a：上游系统往下游系统推送消息，而不关心处理结果；

case b：一份新数据生成，需要实时保存到数据库，索引系统，统计系统等；

case c：调用一个耗时很长的接口，需要在任务完成的时候告知调用方；

这个时候消息总线（Message Queue）就可以发挥作用，它的特长是“解耦”：

case a：消息先推送到MQ，下游从MQ拿消息；

case b：新数据推送到MQ， 数据库、索引系统、统计系统可以各自开一个更新进程（独立的进程可以避免某个系统更新成功，某个系统更新失败的问题），从MQ拿数据更新。

case c：耗时很长的被调接口采用异步机制，收到请求后立即返回给调用方，然后提交具体任务。等真正完成后发消息给MQ， 调用方监控MQ获取完成通知。

MQ的性能一般比较强悍，还可用作“抵御请求流量冲击”，在高峰期的时候，请求都接受发送到MQ，后面根据实际处理能力，从MQ拿数据处理。因此只要满足可“解耦”的大前提，MQ就可以作为数据／请求的总入口，挡在请求的第一线，各个相关系统从MQ获取数据。

Kafka[1] 作为MQ中的佼佼者，有几个特性让人印象深刻：

1）数据多机备份，落地到磁盘，基于offset数据可以方便的指定消费。

2）基于内部协议，获取数据的消费进程天生支持 balance和failover。

3）集群的可靠性较高，维护成本还算低。

4）几乎主流的实时流计算框架（storm，spark，flink）都支持kafka。

2. Kafka的角色分布

Kafka的sever 叫broker，broker有两种角色：leader 和 follower。 leader 和 follower 是针对partition（kafka的逻辑结构见下文）来说的，producer和consumer只会和leader交互。当Kafka集群发生动荡，会重新挑选新的leader，producer 和 consumer 会自动兼容leader的变动，但当leader无法选出，会抛出异常。

3. Kafka的逻辑结构

broker -> topic -> partition -> message<time, key, value>

上面都是1对多的关系，操作数据只需要指定topic，partition的分配系统会帮你完成。

4. Kafka的物理结构

broker上会为拥有的（无论是作为leader还是follower）topic-partition创建目录 topic-partition-number,目录下的内容如截图：

一个segment由3种名字相同的文件组成，.index文件 ， .log 文件，.timeindex 文件

• 文件的名字（segment name）： 记录base offset，全局的offset，从0开始计数。

• .index file : 记录 <relative offset， physical position>，physical position对应的是.log文件物理偏移。

采用折半查找，基于消息的offset可以快速定位所在的文件， segment->index->log：

1）定位segment。

2）使用segment对应的index文件获取物理偏移。

3）使用segment对应的log文件，获取具体的数据。

.timeindex 文件是0.10版本新增的，结构如下：

• .timeindex file: <timestamp, offset>, 当新的msg 的时间大于当前segment的 largest time， 则往 timeindex file 添加一行数据  （0.10）

加入这个文件能优化  time based log rolling， time based log retention， search message by timestamp等操作（详见KIP-33 [2]）。

邮件-数据中心目前选用Kafka部署北京、杭州两地的集群，机器总规模大概20台，涉及的业务包括邮件服务化、严选大数据平台、严选CRM、严选商品搜索等。

二、Kafka producer&consumer  （Kafka的两板斧）

1. producer

0.10的producer新增较多的配置参数，和性能密切相关的配置包括：

• batch.size ：size based batching

• linger.ms：time based batching

• compression.type ： kafka支持 GZIP, Snappy 和 LZ4

• max.in.flight.requests.per.connection ：重试的时候会影响消息顺序

• acks：影响持久化

producer 的数据发送流程：

1. 每个topic的partition有个队列，从队列中获取就绪发送的batch。

2. 发送到相同leader broker的batch聚合在一起。

3. 构建request发送数据。

batch 只要满足以下任一条件，即可认为就绪：

1. batch.size 条件满足。

2. linger.ms 条件满足，收集等待更多的数据可以提升吞吐量。

3. 其它发送到相同leader broker 的batch已经满足条件。

4. flush（） 或者 close（）调用。

batch的size越大，意味着可以获得更好的压缩比率=>更大的吞吐量， 但也会导致 延时较高， 需要根据应用场景做调配。

producer acks 参数的影响：

max.in.flight.requests.per.connection 

• 表示将grouped batch 发送到leader broker，还未resp的请求数

• 设置为 1: request确定resp后，逐个发送，但会降低吞吐量。

• 设置为>1: 提升吞吐量，但在producer失败重试时，会导致message在borker端乱序。

2. producer的性能测试

在线上的备机(40 core, 96GB)测试了一把producer的性能，环境如下：

borker： 一台server上启动3个broker。（只是测试目的，线上环境应该一台server部署一个broker）

topic： partition个数为6， replicator为3

使用自带的性能测试脚本kafka-producer-perf-test.sh，设置 

batch.size = 16KB， 

linger.ms=5，

max.in.flight.requests.per.connection=1，

compression.type=gzip 

acks=-1

总共发送100W消息，每个消息的size为1000byte，结果如下：

32941 records/sec (31.42 MB/sec)

32.53 ms avg latency, 412.00 ms max latency, 

6 ms 50th, 186 ms 95th, 239 ms 99th, 335 ms 99.9th

最基础的参数调配，producer性能还是挺给力的。基于这批参数，可以对吞吐量、延时、持久化做取舍，增加partition的个数也会提升吞吐量。

3. consumer

kafka consumer balance

kafka以group的概念来组合 consumer， 以partition粒度对组内的consumer做rebalancing，封装的消费API可以自动帮你完成balance。实现consumer间的分配调用有两种方式：

• old consumer ： 基于zookeeper实现。消费过的offset信息要更新到zookeeper，可能存在zookeeper的性能瓶颈，0.9以下版本都是这种消费模式。

• new consumer：基于kafka group protocol 支持，使用client-side执行分配策略，在 group coordinator（broker） 维护 partition 到 consumer 的分配关系。将groupId映射到 内部topic  __consumer_offsets 的某个partition， 该partition 的leader 作为 group coordinator，同时该partition也存放offset信息。

消息处理的语意支持

在消息处理上，kafka 只支持 at least once 和 at most once，0.10版本目前不支持 exactly once，因为重复数据可能出现在produce阶段，数据重复处理可能出现在consumer阶段：

 在produce阶段，如果borker回传确认消息的时候挂掉，触发produer重发消息，导致重复数据，这时候如果业务是幂等的，可以忽略该问题。

在consume阶段，依靠可外部系统可以保证数据只处理一次，例如将 partition-offset-result 一起存放到结果中，失败时只要定位上次的offset，可以保证消费阶段的exactly once。

KIP-98 [3]（Exactly Once Delivery and Transactional Messaging）将会支持exactly once，发布在0.11版本。

自动提交offset

配置以下几个参数，可以开始自动提交offset：

enable.auto.commit=true 

auto.commit.interval.ms=5000

若enable.auto.commit设置为true，consumer在获取下一批数据的时候，自动提交上批数据的offset （获得的语意是 at least once）

consumer#poll()方法触发两件事：

1） 如果 enable.auto.commit 设置为true，距离上次提交时间间隔大于auto.commit.interval.ms，则提交上一批数据的offset。

2） 获取当前一批的数据返回。

auto.commit.interval.ms控制offset提交的频率， 值越大consumer rebalance 后重复处理的数据可能越多。

手动提交offset

有两个方法可以手动提交：

consumer.commitSync()  // retry 直到成功 或者 抛出异常

consumer.commitAsync()  // 不会retry， 提供callback包含提交的结果

4. consumer的性能测试

测试机器和producer相同，测试的topic partition个数有6个（消费的线程数最大开到6个），结果如下：

start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec

2017-07-03 14:29:18:164, 2017-07-03 14:29:22:408, 476.8372, 112.3556, 500000, 117813.3836

每秒能消费的消息数量 11W+ ， 数据量每秒 112MB。

消息队列用于构建异步、解耦的架构，会更关注producer的性能能否顶住大量的请求，而在consumer阶段，处理消息的耗时（应用自身的业务逻辑）一般会远远大于consumer自身的耗时。kafka 的producer会等待数据从leader到follower的落地（ack=all），而consumer没有这些耗时，只需要批量的从leader获取数据和更新消费Offset，所以性能会比producer好很多。

三、Kafka的大版本升级和兼容性 （部署的系统总会要升级的）

1. Kafka 版本升级

当前版本：0.8.1.1

升级版本：0.10.2

需要升级的几大因素：

1. 升级后有权限控制ACL

可以精确控制每个topic 的produce权限和consume权限。

2. 消息offset数据的存储改变

消费的offset存放位置在低版本中在zookeeper， 存在一定的性能瓶颈和隐患，新版本中已经放到 internal topic， 性能有较大提升。

3. kafka stream的引入

kafka stream 是轻量级的流处理Client lib， 无任何其它依赖。

它使得Apache Kafka可以拥有流处理的能力，通过使用Kafka Stream API进行业务逻辑处理最后写回Kakfa或者其他系统中。

2. 升级目标

1. 不影响线上任务，逐台重启。

2. 能兼容新旧的producer和consumer，不影响旧代码的运行。

3. 旧的数据生成消费机制能顺利的过渡到新模式下，不会对数据造成重复处理或丢失。

3. 升级过程记录

两个重要的参数设置：

inter.broker.protocal.version

更新中：0.8.1.1

全部broker更新完成后设置为 0.10.2 ，再逐台重启服务。

log.message.format.version

更新中：0.8.2

所有consumer都更新后（new consumer），再设置为 0.10.2

如果设置为 0.8.1.1 ，获取消息的时候会做转换（兼容低版本的message格式），没有zero-copy， CPU负载上升5倍， 建议设置为 0.8.2 or 0.9.0。

因为broker是向下兼容的，升级过程中必需先成功升级所有的broker。Client（producer 和 consumer）在broker完成升级之后再升级。但在实际情况中，client涉及的程序很多，升级进度必然会远远落后于broker端，所以我们考虑跨版本的情况（broker的版本高于client）。

4. 无压缩跨版本的性能测试

broker: 0.10.2

producer: 0.8.1.1

./bin/kafka-producer-perf-test.sh --messages 500000 --message-size 50000 --topic test-update-perf --threads 8 --broker-list * —show-detailed-stats --csv-reporter-enabled --metrics-dir ./perf-test --reporting-interval 3000

结果如下：

compression, message.size, batch.size, total.data.sent.in.MB, MB.sec, total.data.sent.in.nMsg, nMsg.sec

0, 50000, 200, 23841.86, 56.4680, 500000, 1184.2196

QPS 1184 nMsg.sec，每个broker 进程 CPU load < 50%。

old consumer：0.8.1.1

./bin/kafka-consumer-perf-test.sh --topic test-update-perf --zookeeper 10.100.96.94:2183 --threads 1 --group perf-consumer-t4 --message-size 50000 --messages 10

QPS大概在 3000 nMsg.sec，使用old consumer（0.8.1） 消费 new broker（0.10.2）的数据， broker需要做消息格式的转换， 实际观测 broker 进程的CPU负载不高 < 20%。

5. 跨版本压缩数据的测试

broker: 0.10.2

old producer： 0.8.1.1

start.time, end.time, compression, message.size, batch.size, total.data.sent.in.MB, MB.sec, total.data.sent.in.nMsg, nMsg.sec

2017-06-09 18:17:09:810, 2017-06-09 18:17:40:309, 2, 50000, 200, 23841.86, 781.7259, 500000, 16393.9801

QPS 16393 nMsg.sec, 每个broker 进程 CPU load < 50%

old consumer：0.8.1.1

QPS大概40000 nMSg.sec，使用 old consumer 消费压缩数据，broker 的CPU 负载没有很高 <20%。

因为跨版本调用涉及到消息格式转换，更关注的是带来的CPU负载，性能这块可以参见上文的producer、consumer性能测试相关。

四、Kafka Stream 初窥 （实时计算我有我的路）

1. Kafka Stream简介

目前支持实时流计算的有一些框架，如： Spark （micro batch），Storm， Flink，Samza；这些系统都需要部署集群。Kafka stream是实时流计算库，依靠现有的kafka集群，提供分布式的、高容错的、抽象DSL的实时流计算实现：

1. 除了Kafka Stream Client lib以外无外部依赖，轻量的嵌入到Java应用中。

2. 严格区分Event time（数据产生的时间）和Process Time（系统处理的时间），可处理乱序到达的数据。

3. 支持本地状态故障转移，以实现非常高效的有状态操作，如join和window函数。

4. 提供必要的流处理原语、hige-level Stream DSL和low-level Processor API。

2. Kafka Stream和其它流计算框架的对比

一般的流计算框架用job或者topology来描述任务，它们关注的是：

• 对于提交的任务，高效的将机器资源分配给任务执行。

• 任务必须打包自己的业务实现代码（包括依赖包，配置等）提交到集群，集群负责部署到工作节点。

• 管理任务的执行情况，保证任务之间的资源隔离性。

kafka stream更关注于问题本身：

• 当有新的实例加入执行或者某个执行实例挂掉，就会触发工作实例的balance（原理和kafka consumer的balance一样）。只需要启动一个java进程就可以加入计算任务，kafka stream的实例很适合用容器部署。

• kafka stream的输入和输出都是kafka的topic，同时可维护本地状态（key-value 的表，默认是以 RocksDB实现）支持aggregations和join的计算。状态结果以topic作为Write-ahead logging，所以即使某个实例失败，状态也可以在其它实例恢复。一般实时计算的结果如果要查询，需要将结果数据写入到外部系统（如HBase，Redis），但基于kafka stream的本地状态存储，可以直接向各个实例查询 [4]，节省数据的再落地成本。

3. 什么时候选用Kafka Stream

当你的环境中已经存在Kafka作为数据入口，后面当然也可以接 Spark、Storm、Flink等实时流处理框架。如果你的实时流计算只是做一些数据转换清洗、按key聚合计算、需要读取多个topic的数据join，不妨先看下kafka stream的流式计算支持，它能以更轻便的方式实现计算需求。如果你的数据来源涉及DB，HDFS，计算过程中涉及机器学习、NoSQL， 则需要考虑 Spark、Storm、Flink。

REF:

[1]  http://kafka.apache.org/intro.html

[2]  https://cwiki.apache.org/confluence/display/KAFKA/KIP-33+-+Add+a+time+based+log+index

[3]  https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging

[4] https://www.confluent.io/blog/unifying-stream-processing-and-interactive-queries-in-apache-kafka/

本文来自网易实践者社区，经作者潘胜一授权发布

网易云免费体验馆，0成本体验20+款云产品！

更多网易研发、产品、运营经验分享请访问网易云社区。

相关文章：
【推荐】 八月暑期福利，10本Python热门书籍免费送！
【推荐】 如何实现最佳的跨平台游戏体验？Unity成亮解密实时渲染
【推荐】 如何从“点子”落地到“执行”？—完整解析1个手游传播类mini项目的进化