Kafka介绍及升级经验分享

http://blog.talkingdata.net/?p=3165

• 背景

当时的现状：开始使用Kafka的时候，使用的版本是0.7.2，当时的目的是为了替代kestrel，主要是使用Kafka来做消息传输中间件。Kafka解决了我们当时使用Kestrel遇到的最大的三个问题：吞吐量、数据量、一份数据多次消费。

• 为什么要升级

相比其它开源项目，Kafka的升级比较麻烦，其根本原因主要是作为消息传输中间件，涉及的系统多。既然升级麻烦，而且Kafka 0.7在这一年多来运行稳定，性能优异，那么我们为什么要升级呢？

其实之所以决定升级Kafka，主要有两个原因，一是因为Kafka 0.7的scala版本是2.8，目前使用的多数开源框架（Spark、Play等）都是基于2.10的scala了，而scala 2.10和2.8两个版本是不兼容的。二，是因为之前曾发生过Kafka服务器RAID卡损坏的故障，期待Kafka 0.8的Replication功能。

• 升级中遇到的问题及解决方案

1. 配置问题：消费者找不到Broker

2. 问题描述：消费者在消费数据的时候，连不上Broker

3. 问题原因：Broker在正常启动之后会在zookeeper中注册自己的信息。消费者会根据这里面的host和port去连接broker，host是在server.properties配置host.name配置的值，这个值如果不配置，那么在Zookeeper中存放的就是Kafka这台服务器的主机名而不是ip，所以消费者才会连不上。而在Kafka0.7中，类似的配置叫hostname，这个值如果不配置，它会调用InetAddress.getLocalHost()去获取，获取的值不一定是你想要的，在我们当时它获取恰好是Kafka服务器的ip。

4. 解决方案（其中一种）

修改server.properties中host.name的配置，把它改成ip。

修改消费者所在机器的hosts文件，加入Kafka主机名与ip的映射。

使用DNS（推荐）

• 数据去哪了

1. 问题描述：

生产者生产10000条数据后停止，这时启动消费者，发现消费者多不到任何东西，而且zookeeper中的offset居然和生产者生产了的offset一样。如果这时启动生产者继续发送数据，消费者从第10001条数据开始读取。之前的10000条数据都不见了。

2. 问题原因：Kafka的官方wiki的原话：

In 0.8, wehave moved to logical offsets from physical offsets. This means that theoffsets are not compatible. When you try to consume using the 0.7 offsets, youwould hit “OffsetOutOfRangeException”. The default behavior of theconsumer when this happens is based on the config value of “auto.offset.reset”.If it is set to “smallest”, the consumer will start consuming fromthe beginning. If it is set to “largest”, the consumer will startconsuming from the end.

3. 解决方案：给消费者增加auto.offset.reset配置，auto.offset.reset=smallest

• 给”消息”减减肥

1. 问题描述：生产者或消费者抛出MessageSizeTooLargeException异常

2. 问题原因：这个异常的命名还是很直白的，消息太大了，去官网找找配置就解决了，比较郁闷的就是同样的消息大小，在Kafka0.7没有配置相应的参数也不报错。

3. 解决方案：

如果是生产者报错，修改Kafka Broker的配置，在server.properties中配置message.max.bytes，默认是1M（约）。

如果是消费者报错，修改消费者中增加fetch.message.max.bytes的配置，这个配置的值要大于Broker的message.max.bytes配置。

• 性能问题：ACK参数配置

1. 问题描述：生产者上线后，吞吐量下降了1倍。

2. 问题原因：

首先检查Kafka Broker，发现的不管是网络、IO、CPU等都没有出现瓶颈，并且增加生产者线程或者生产者实例可以解决问题，假设生产者写Kafka的速度是10000条每秒，那么再部署一个生产者，两者写入速度均可以达到10000条，遂断定问题出在生产者本身，通过jstack可以发现，线程都在做写Kafka的操作，那么写Kafka究竟和0.7有什么不一样呢？

Kafka0.8有Replication功能，消息写入Kafka中后，Followers会创建副本，生产者有个配置叫request.required.acks，当时配置的是1，生产者会等至少1个Followers创建完副本之后才算发送成功，平均响应时间变长，所以速度变慢。

3. 解决方案（其中一种）

增加生产者线程数或者生产者实例，系统的相应时间增加，但是系统的并发数并没有到达上限，并且Kafka Broker可以平行扩展。设置request.required.acks=0，这样做会有丢失数据的风险。

• Producer锁

1. 问题描述：生产者有很多的线程状态都是BLOCKED，导致系统性能大幅度下降。

2. 问题原因：根据源代码可以看出，生产者发送时是有锁的，但这个锁每个Producer对象各自持有各自的。

3. 解决方案：对于不同线程使其持有不同的producer对象。

• 坑：文件分段大小配置有bug

1. 问题描述：

系统上线后我们遇到了文件句柄数过多的问题，如果配置的分段文件大小一样，0.8会比0.7多4倍的文件数目，所以我们当时决定增加分段文件的大小，必须是一个Int，于是改成Int.Max，结果数据整个offset错乱，文件损坏，集群不可用了。

2. 问题原因：

private def maybeRoll(messagesSize: Int): LogSegment = {

val segment = activeSegment

if (segment.size > config.segmentSize – messagesSize ||

segment.size > 0 && time.milliseconds – segment.created > config.segmentMs – segment.rollJitterMs ||

segment.index.isFull) {

……

roll()

} else {

segment

}

}

3. 解决方案：配小一些或者升级到0.8.2

• 神一般的错误提示

1. 问题描述：Consumer消费时出现Iterator is in failed state的错误提示，错误量很多。

2. 问题原因：

这个错误并不是真正的错误，是因为MessageSizeTooLargeException导致的，发生MessageSizeTooLargeException异常会导致Iterator is in failed state错误的发生，但是MessageSizeTooLargeException只会打印一次，而那个错误会随着读取方法的调用不停的打，完全被带跑偏了。

3. 解决方案：解决MessageSizeTooLargeException即可。