kafka consumer
这种消费方式由broker主动推送消息给消费者,消费者被动接收消息。缺点: consumer 消费能力不强的情况下可能出现拒绝服务、以及因网络问问题产生的网络拥塞的情况;
消费者主动轮询broker是否有数据可以消费,拉取消息的速率完全由consumer自己掌握,但是可能会出现broker没有数据,消费者陷入无限循环当中;解决的办法是,在kafka consumer消费数据时传入一个时长参数 timeout,防止cpu空转
一个consumer group 中有多个consumer, 一个topic 中会有多个partition;所以会出现消费者消费分区数据时,partition分配的问题,即确定哪个partition 由哪个consumer来消费?
如下图所示: 假如同一个主题:
另外 consumer group 可以对多个主题进行消费: 看如下场景: 2个主题 T1、T2
在这种场景下分区策略伪代码如下所示:
def get_partition_index():
return map(TopicAndPartition:List, hash()) mod num(consumers)
来确定分配给消费者组中消费者的partition index 这种方式消费者组会将所有topic 中的 partition 当作一个整体来轮询分配。分配主体是消费者组; 适用于 消费者组中所有 消费者 订阅的都是同一个主题 的场景。
range 分配主体是被消费的broker的单个主题: consumer group 中的单个consumer 被分配的 可消费 partition 个数差距越来越大。要点: 按主题来区分的。
range策略详述,我们根据一个场景来深入理解一下: 如下图所示:
当前某主题有8个partition,某消费者组中消费者的个数是3个,那么最终的分配结果如下图所示:
具体算法过程也很简单,简述一下:
1. 计算n = num(topic partitions)/num(consumers of consumer group) = 8/3 = 2
2. 计算m = num(topic partitions)%num(consumers of consumer group) = 2
3. 分配规则为消费组中的前m个消费者,每个消费者可以分配到的分区数为n+1 = 2+1=3, 剩余的消费者可消费的分区数为n
range 分配策略是有一些问题的:加入新的主题,但是消费者组中的消费者数量不变,那么头部的消费者就会被分配更多的partition,造成分配不均的问题。
当消费者组中的消费者数量发生变化的时候,就会调用Range策略。
如下图所示,开启了4个窗口: (顺时针描述) 左上: 1个producer console - 生产消息至first topic(first topic 有3个partition,每个partition有2个replication)分区,具体消息进入哪个分区 ,属于无指定partition,无key,有value的情况,可以参考 producer 这一节来理解。右上: 1个consumer console(其实是一个消费者组,只不过只有1个消费者) - 消费 first topic 分区(leader) 中的消息 右下: 1个consumer console (其实是一个消费者组,只不过只有1个消费者) - 消费 first topic 分区(leader) 中的消息 左下: 1个zkcli console, 从中我们可以查看到具体的contoller、brokers、consumers、config 等相关信息
通过producer 发送消息 + consumer 消费消息, 在zkCli 中查看具体的消费者消费消息的offset变化 命令为:
./zkCli.sh -server localhost:2181
get /consumers/$consumer_group/offsets/$topic/$partition
所以老版本消费者消费消息的offset 记录方式为 [consumergroup]+[topic]+[partition index]
采用这样的方式记录offset,当consumer group 中下线、上线新的consumer时,消费过的消息就不会被重新消费。
下一节我们继续学习新版本 bootstrap-server上保存消费者消费过的消息的offset机制!