Kafka Consumer

本文转发自技术世界，原文链接　http://www.jasongj.com/2015/08/09/KafkaColumn4

本文主要介绍了Kafka High Level Consumer，Consumer Group，Consumer Rebalance，Low Level Consumer实现的语义，以及适用场景。以及未来版本中对High Level Consumer的重新设计–使用Consumer Coordinator解决Split Brain和Herd等问题。

Kafka架构：

一个典型的kafka集群中包含若干producer（可以是web前端产生的page view，或者是服务器日志，系统CPU、memory等），若干broker（Kafka支持水平扩展，一般broker数量越多，集群吞吐率越高），若干consumer group，以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置，选举leader，以及在consumer group发生变化时进行rebalance。producer使用push模式将消息发布到broker，consumer使用pull模式从broker订阅并消费消息。

一、Pull vs Push

Producer Producer通过主动Push的方式将消息发布到Broker

Consumer Consumer通过pull从Broker消费数据

>>Push

• 优势：延时低
• 劣势：可能造成Consumer来不及处理消息，网络拥塞；

>>Pull

• 优势：Consumer按实际处理能力获取相应量的数据，Broker实现简单；
• 劣势：如果处理不好，实时性相对不足（kafka使用long polling）

二、Topic vs Partition

 

三、  High Level Consumer

1. 很多时候，客户程序只是希望从Kafka读取数据，不太关心消息offset的处理。
2. 同时也希望提供一些语义，例如同一条消息只被某一个Consumer消费（单播）或被所有Consumer消费（广播）。
3. 因此，Kafka Hight Level Consumer提供了一个从Kafka消费数据的高层抽象，从而屏蔽掉其中的细节并提供丰富的语义。

四、  Consumer Group

1. High Level Consumer将从某个Partition读取的最后一条消息的offset存于Zookeeper中(Kafka从0.8.2版本开始同时支持将offset存于Zookeeper中与将offset存于专用的Kafka Topic中)。
2. 这个offset基于客户程序提供给Kafka的名字来保存，这个名字被称为Consumer Group。
3. Consumer Group是整个Kafka集群全局的，而非某个Topic的。
4. 每一个High Level Consumer实例都属于一个Consumer Group，若不指定则属于默认的Group。

　　

Zookeeper中Consumer相关节点如下图所示

很多传统的Message Queue都会在消息被消费完后将消息删除，一方面避免重复消费，另一方面可以保证Queue的长度比较短，提高效率。

1. 而Kafka并不删除已消费的消息，知识相应的Offset加1，为了实现传统Message Queue消息只被消费一次的语义；
2. Kafka保证每条消息在同一个Consumer Group里只会被某一个Consumer消费；
3. 与传统Message Queue不同的是，Kafka还允许不同Consumer Group同时消费同一条消息，这一特性可以为消息的多元化处理提供支持。

1. Kafka的设计理念之一就是同时提供离线处理和实时处理；
2. 根据这一特性，可以使用Storm这种实时流处理系统对消息进行实时在线处理，同时使用Hadoop这种批处理系统进行离线处理；
3. 还可以同时将数据实时备份到另一个数据中心（可使用kafka的MIrror Maker将消息从一个数据中心镜像到另一个数据中心），只需要保证这三个操作所使用的Consumer在不同的Consumer Group即可。

下图展示了Kafka在LinkedIn的一种简化部署模型。

 

五、High Level Consumer Rebalance

（本节所讲述Rebalance相关内容均基于Kafka High Level Consumer）
　　Kafka保证同一Consumer Group中只有一个Consumer会消费某条消息，实际上，Kafka保证的是稳定状态下每一个Consumer实例只会消费某一个或多个特定Partition的数据，而某个Partition的数据只会被某一个特定的Consumer实例所消费。也就是说Kafka对消息的分配是以Partition为单位分配的，而非以每一条消息作为分配单元。这样设计的劣势是无法保证同一个Consumer Group里的Consumer均匀消费数据，优势是每个Consumer不用都跟大量的Broker通信，减少通信开销，同时也降低了分配难度，实现也更简单。另外，因为同一个Partition里的数据是有序的，这种设计可以保证每个Partition里的数据可以被有序消费。
　　如果某Consumer Group中Consumer（每个Consumer只创建1个MessageStream）数量少于Partition数量，则至少有一个Consumer会消费多个Partition的数据，如果Consumer的数量与Partition数量相同，则正好一个Consumer消费一个Partition的数据。而如果Consumer的数量多于Partition的数量时，会有部分Consumer无法消费该Topic下任何一条消息。
　　1、如果topic1有0，1，2共三个Partition，当group1只有一个Consumer(名为consumer1)时，该 Consumer可消费这3个Partition的所有数据。

　　2、增加一个Consumer(consumer2)后，其中一个Consumer（consumer1）可消费2个Partition的数据（Partition 0和Partition 1），另外一个Consumer(consumer2)可消费另外一个Partition（Partition 2）的数据。

　　3、再增加一个Consumer(consumer3)后，每个Consumer可消费一个Partition的数据。consumer1消费partition0，consumer2消费partition1，consumer3消费partition2。

　　4、再增加一个Consumer（consumer4）后，其中3个Consumer可分别消费一个Partition的数据，另外一个Consumer（consumer4）不能消费topic1的任何数据。

　　5、此时关闭consumer1，其余3个Consumer可分别消费一个Partition的数据。

　　6、接着关闭consumer2，consumer3可消费2个Partition，consumer4可消费1个Partition。

　　7、再关闭consumer3，仅存的consumer4可同时消费topic1的3个Partition。

Consumer Rebalance的算法如下：

• 将目标Topic下的所有Partirtion排序，存于P_T
  
• 对某Consumer Group下所有Consumer排序，存于于C_G，第ii个Consumer记为C_i
• N=size(P_T)/size(C_G)N，向上取整
• 解除C_i对原来分配的Partition的消费权（i从0开始）
• 将第i∗N到（i+1）∗N−1个Partition分配给C_i

　目前，最新版（0.8.2.1）Kafka的Consumer Rebalance的控制策略是由每一个Consumer通过在Zookeeper上注册Watch完成的。每个Consumer被创建时会触发Consumer Group的Rebalance，启动流程如下：

• High Level Consumer启动时将其ID注册到其Consumer Group下，在Zookeeper上的路径为/consumers/[consumer group]/ids/[consumer id]
• 在/consumers/[consumer group]/ids上注册Watch
• 在/brokers/ids上注册Watch
• 如果Consumer通过Topic Filter创建消息流，则它会同时在/brokers/topics上也创建Watch
• 强制自己在其Consumer Group内启动Rebalance流程

　　在这种策略下，每一个Consumer或者Broker的增加或者减少都会触发Consumer Rebalance。因为每个Consumer只负责调整自己所消费的Partition，为了保证整个Consumer Group的一致性，当一个Consumer触发了Rebalance时，该Consumer Group内的其它所有其它Consumer也应该同时触发Rebalance。

Consumer Rebalance算法缺陷：

• Herd effect
  　　任何Broker或者Consumer的增减都会触发所有的Consumer的Rebalance
• Split Brain
  　　每个Consumer分别单独通过Zookeeper判断哪些Broker和Consumer 宕机了，那么不同Consumer在同一时刻从Zookeeper“看”到的View就可能不一样，这是由Zookeeper的特性决定的，这就会造成不正确的Reblance尝试。
• 调整结果不可控
  　　所有的Consumer都并不知道其它Consumer的Rebalance是否成功，这可能会导致Kafka工作在一个不正确的状态。

      根据Kafka社区wiki，Kafka作者正在考虑在还未发布的0.9.x版本中使用中心协调器(Coordinator)。大体思想是为所有Consumer Group的子集选举出一个Broker作为Coordinator，由它Watch Zookeeper，从而判断是否有Partition或者Consumer的增减，然后生成Rebalance命令，并检查是否这些Rebalance在所有相关的Consumer中被执行成功，如果不成功则重试，若成功则认为此次Rebalance成功（这个过程跟Replication Controller非常类似）。具体方案将在后文中详细阐述。

六、Low Level Consumer

使用Low Level Consumer (Simple Consumer)的主要原因是，用户希望比Consumer Group更好的控制数据的消费。比如：

• 同一条消息读多次
• 只读取某个Topic的部分Partition
• 管理事务，从而确保每条消息被处理一次，且仅被处理一次

　　与Consumer Group相比，Low Level Consumer要求用户做大量的额外工作。

• 必须在应用程序中跟踪offset，从而确定下一条应该消费哪条消息
• 应用程序需要通过程序获知每个Partition的Leader是谁
• 必须处理Leader的变化

　　使用Low Level Consumer的一般流程如下

• 查找到一个“活着”的Broker，并且找出每个Partition的Leader
• 找出每个Partition的Follower
• 定义好请求，该请求应该能描述应用程序需要哪些数据
• Fetch数据
• 识别Leader的变化，并对之作出必要的响应