3.2.4    客户端分析

3.2.4.1 Consumer客户端

Consumer客户端，使用的是Scala的客户端。

多硬盘，单分区下，JProfile如下：

400分区如下：

3000分区如下：

3000分区的时候，可以看到，97.8的cpu时间被用来执行了parseFull这个函数，这个函数是创建Streams的时候，对从zk获取的json文件进行解析，每次只执行一次。

但是由于分区越多，这个文件越大越复杂，导致执行时间非常长，影响了Consumer的TPS。

由于这个函数只执行一次，不会大量被执行，所以当Consumer运行时间越长，那么影响会越小。

3.2.4.2 Producer Scala客户端

最开始使用Scala客户端的测试，发现多分区下，大量的时间被消耗在日志打印参数的format函数上：

25分区：

这里的代码，会在分个消息发送的时候，对分区的MAP循环执行如下的函数，这个代码里面会多次拼接字符串很耗时。

通过屏蔽日志打印函数，避免执行format，可以大大提升TPS：

    partitionMetadata.map { m =>

      m.leader match {

        case Some(leader) =>

          //debug("Partition [%s,%d] has leader %d".format(topic, m.partitionId, leader.id))

          new PartitionAndLeader(topic, m.partitionId, Some(leader.id))

        case None =>

          //debug("Partition [%s,%d] does not have a leader yet".format(topic, m.partitionId))

          new PartitionAndLeader(topic, m.partitionId, None)

      }

    }.sortWith((s, t) => s.partitionId < t.partitionId)

那两句debug信息打印注释掉，能大大的提升TPS。

估计还有其他的瓶颈点，之后我们将测试客户端切换到了java客户端。

3.2.4.3 Producer Java客户端

多硬盘一分区下，客户端的情况（ACK=1）：

多硬盘5000分区下，Producer客户端（ACK=1）：

多硬盘5000分区下，Producer客户端（ACK=0）

我们可以看到，分区数量越大，send函数耗时越长，5000分区下，耗时大半。

ACK=0的时候，客户端是不需要等待server的响应的，可以说基本上客户端的TPS和服务器端没有关系，但是，TPS照样降低，send照样消耗大量的时间。

所以，我们判断，多硬盘下，当分区达到5000以上的时候，瓶颈在客户端代码上，可以通过优化客户端代码来提升性能。

4     分析结论

4.1   结论

一、单Broker的分区，主要受以下几个方面的瓶颈限制：

1、 OS限制：Kafka代码中没有对分区做限制，但是分区数量受到OS的文件句柄的数量限制。可以用ulimit –a查看。需要调整该数目到一个较大的值。

2、 文件系统（磁盘刷盘）：单硬盘下，磁盘IO和util会随着分区数目增大而增大，导致Kafka的性能会下降。

主要原因：

刷盘是一个很昂贵的操作，虽然OS提供了Group Commit和排序等优化机制，但是当分区数量太多之后，无法完全消除影响。每个分区写入是磁盘顺序写，但是多个分区同时顺序写入在操作系统层面变为了随机写入。

另外，当Pagecache的dirty数据达到一定限制之后，刷盘操作会阻塞应用的write操作，也会带来影响。

3、 客户端瓶颈：

通过追加硬盘可以解决单硬盘下的文件系统瓶颈。但是当达到5000以上，客户端会成为瓶颈。

这个是由于发送消息的时候需要获取Topic下的分区信息而产生的。

实际生产中，这个限制的影响应该不大，因为不会在1个Topic下划分太多的分区，而是会有多个Topic，每个Topic下有一些分区。测试的时候为了测试方便，采取了一种比较极端的测试方法（单个Topic下挂上数千的分区）。

当有需要的时候，也可以考虑优化。

建议分区数量：

1、单台服务器单硬盘下，推荐值1000，最好不超过2000。

2、单台服务器多硬盘下（7个），推荐值3000，最好不要超过5000。（当分区被分散到多个Topic的时候，这个值会更高，预计至少可以达到7000以上）。

二、集群（多节点）的分区限制：

理论上是随着节点增加而线性增加。但是实际还是受到Zookeeper的影响。节点最大支持数量估计在1w（zookeeper的限制）左右，那么整个集群分区的支持估计在百万左右（但是实际上受网络速度的影响，分区越大，存放在zookeeper的数据量越多等各种因素影响肯定会有个折扣）。

预计10w左右的分区支持问题不大。

1）        参考LinkedIn的Kafka集群部署规模：

http://www.slideshare.net/miguno/apache-kafka-08-basic-training-verisign

2)   在这篇文章中，提到曾经在单broker上使用了上万的分区。

http://www.confluent.io/blog/how-to-choose-the-number-of-topicspartitions-in-a-kafka-cluster/

4.2   社区意见参考

这里，kafka的committer有很明确的答复,基本上和我们验证的结果一致：

http://www.quora.com/How-many-topics-can-be-created-in-Apache-Kafka

有几个主要观点：

1、 Kafka的分区数量两个因素的影响：文件系统和zookeeper。

2、 实际应用中，分区数量不应该成为一个问题。分区的数量在实际中应该随着消费者数量扩展，不应该根据Data的特征来扩展。

4.3  解决方案

针对目前的几个瓶颈的解决方案如下：

1、 OS限制：通过调整Server的支持文件数目的句柄来解决。

2、 文件系统（磁盘刷盘）：通过追加硬盘或追加Broker Node的方式来解决。

单个硬盘支持的分区数量推荐1000。多个硬盘：N*1000。

3、 客户端：当一个Topic下有过多的分区的时候（>3000），客户端预计会成为瓶颈。

如果有这种需求，可以考虑优化客户端来解决这里的瓶颈。

5     备注

5.1   参考资料

5.1.1    Cloudera配置Kafka的建议

http://blog.cloudera.com/blog/2015/07/deploying-apache-kafka-a-practical-faq/

非常好的实际经验。

http://blog.cloudera.com/blog/category/kafka/

是否应当为Kafka Broker使用 固态硬盘 (SSD)

实际上使用SSD盘并不能显著地改善 Kafka 的性能，主要有两个原因：

·         Kafka写磁盘是异步的，不是同步的。就是说，除了启动、停止之外，Kafka的任何操作都不会去等待磁盘同步（sync）完成；而磁盘同步(syncs)总是在后台完成的。这就是为什么Kafka消息至少复制到三个副本是至关重要的，因为一旦单个副本崩溃，这个副本就会丢失数据无法同步写到磁盘。

·         每一个Kafka Partition被存储为一个串行的WAL（Write Ahead Log）日志文件。因此，除了极少数的数据查询，Kafka中的磁盘读写都是串行的。现代的操作系统已经对串行读写做了大量的优化工作。

如何对Kafka Broker上持久化的数据进行加密 

目前，Kafka不提供任何机制对Broker上持久化的数据进行加密。用户可以自己对写入到Kafka的数据进行加密，即是，生产者(Producers)在写Kafka之前加密数据，消费者(Consumers)能解密收到的消息。这就要求生产者(Producers)把加密协议(protocols)和密钥(keys)分享给消费者(Consumers)。

另外一种选择，就是使用软件提供的文件系统级别的加密，例如Cloudera Navigator Encrypt。Cloudera Navigator Encrypt是Cloudera企业版(Cloudera Enterprise)的一部分，在应用程序和文件系统之间提供了一个透明的加密层。

Apache Zookeeper正成为Kafka集群的一个痛点(pain point)，真的吗？

Kafka高级消费者(high-level consumer)的早期版本(0.8.1或更早)使用Zookeeper来维护读的偏移量(offsets，主要是Topic的每个Partition的读偏移量)。如果有大量生产者(consumers)同时从Kafka中读数据，对Kafka的读写负载可能就会超出它的容量，Zookeeper就变成一个瓶颈(bottleneck)。当然，这仅仅出现在一些很极端的案例中(extreme cases)，即有成百上千个消费者(consumers)在使用同一个Zookeeper集群来管理偏移量(offset)。

不过，这个问题已经在Kafka当前的版本(0.8.2)中解决。从版本0.8.2开始，高级消费者(high-level consumer)能够使用Kafka自己来管理偏移量(offsets)。本质上讲，它使用一个单独的Kafka Topic来管理最近的读偏移量(read offsets)，因此偏移量管理(offset management)不再要求Zookeeper必须存在。然后，用户将不得不面临选择是用Kafka还是Zookeeper来管理偏移量(offsets)，由消费者(consumer)配置参数 offsets.storage 决定。

Cloudera强烈推荐使用Kafka来存储偏移量。当然，为了保证向后兼容性，你可以继续选择使用Zookeeper存储偏移量。(例如，你可能有一个监控平台需要从Zookeeper中读取偏移量信息。) 假如你不得不使用Zookeeper进行偏移量(offset)管理，我们推荐你为Kafka集群使用一个专用的Zookeeper集群。假如一个专用的Zookeeper集群仍然有性能瓶颈，你依然可以通过在Zookeeper节点上使用固态硬盘(SSD)来解决问题。

Kafka是否支持跨数据中心的可用性

Kafka跨数据中心可用性的推荐解决方案是使用MirrorMaker。在你的每一个数据中心都搭建一个Kafka集群，在Kafka集群之间使用MirrorMaker来完成近实时的数据复制。

使用MirrorMaker的架构模式是为每一个”逻辑”的topic在每一个数据中心创建一个topic：例如，在逻辑上你有一个”clicks”的topic，那么你实际上有”DC1.clicks”和“DC2.clicks”两个topic(DC1和DC2指得是你的数据中心)。DC1向DC1.clicks中写数据，DC2向DC2.clicks中写数据。MirrorMaker将复制所有的DC1 topics到DC2，并且复制所有的DC2 topics到DC1。现在每个DC上的应用程序都能够访问写入到两个DC的事件。这个应用程序能够合并信息和处理相应的冲突。

另一种更复杂的模式是在每一个DC都搭建本地和聚合Kafka集群。这个模式已经被Linkedin使用，Linkedin Kafka运维团队已经在 这篇Blog 中有详细的描述(参见“Tiers and Aggregation”)。

5.1.2    大数据领域的架构图

http://www.blogbus.com/navigating-logs/272257444.html

5.1.3    源代码High level分析

http://ju.outofmemory.cn/entry/124628

5.1.4    Kafka的配置推荐

http://bbs.chinacloud.cn/archiver/showtopic-29836.aspx

http://liyonghui160com.iteye.com/blog/2163899

5.1.5    Zookeeper的讨论

http://bbs.chinacloud.cn/archiver/showtopic-29836.aspx

 重要的记录： 

5.1.6    Kafka跨集群同步方案

http://jingyan.baidu.com/article/8065f87fea4d3a233124989f.html

http://tangzhaohui.net/post/524

5.1.7    实践部署与使用apache kafka

http://blog.csdn.net/zhongwen7710/article/details/41252649

5.1.8    从Apache Kafka 重温文件高效读写

http://calvin1978.blogcn.com/articles/kafkaio.html