kafka的消费者

消费组和消费者概念

24154608_yl1d.jpg.png

• 每个consumer客户端被创建时,会向zookeeper注册自己的信息;
• 同一个Consumer Group中的Consumers，Kafka将相应Topic中的每个消息只发送给其中一个Consumer。
• Consumer Group中的每个Consumer读取Topic的一个或多个Partitions，并且是唯一的Consumer；
• 一个Consumer group的多个consumer的所有线程依次有序地消费一个topic的所有partitions,如果Consumer group中所有consumer总线程大于partitions数量，则会出现空闲情况;

举例：

• kafka集群中创建一个topic为report-log 4 partitions 索引编号为0,1,2,3， 假如有目前有三个消费者node。
• 如果每个consumer创建一个consumer thread线程,各个node消费情况如下，node1消费索引编号为0,1分区，node2费索引编号为2,node3费索引编号为3。
• 如果每个consumer创建2个consumer thread线程，各个node消费情况如下(是从consumer node先后启动状态来确定的)，node1消费索引编号为0,1分区；node2费索引编号为2,3；node3为空闲状态。
  总结：从以上可知，Consumer Group中各个consumer是根据先后启动的顺序有序消费一个topic的所有partitions的。
  如果Consumer Group中所有consumer的总线程数大于partitions数量，则可能consumer thread或consumer会出现空闲状态。

消费者均衡算法

当一个group中,有consumer加入或者离开时,会触发partitions均衡.均衡的最终目的,是提升topic的并发消费能力。

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

实例：

1. topic1,具有如下partitions: P0,P1,P2,P3
2. 消费group中,有如下consumer: C0,C1
3. 首先根据partition索引号对partitions排序: P0,P1,P2,P3
4. 根据(consumer.id + '-'+ thread序号)排序: C0,C1
5. 计算倍数: M = [P0,P1,P2,P3].size / [C0,C1].size,本例值M=2(向上取整
6. 然后依次分配partitions: C0 = [P0,P1],C1=[P2,P3],即Ci = [P(i * M),P((i + 1) * M -1)]