- wpBeta
-
* 回复内容中包含的链接未经审核,可能存在风险,暂不予完整展示!
consumer group是kafka提供的可扩展且具有容错性的消费者机制。组内可以有多个消费者或消费者实例(consumer instance),它们共享一个公共的ID,即group ID。组内的所有消费者协调在一起来消费订阅主题(subscribed topics)的所有分区(partition)。
consumer group下可以有一个或多个consumer instance,consumer instance可以是一个进程,也可以是一个线程
group.id是一个字符串,唯一标识一个consumer group
consumer group下订阅的topic下的每个分区只能分配给某个group下的一个consumer(当然该分区还可以被分配给其他group)
Coordinator一般指的是运行在broker上的group Coordinator,用于管理Consumer Group中各个成员,每个KafkaServer都有一个GroupCoordinator实例,管理多个消费者组,主要用于offset位移管理和Consumer Rebalance。
对于每个Consumer Group,Coordinator会存储以下信息:
consumer group如何确定自己的coordinator是谁呢? 简单来说分为两步:
消费者在消费的过程中需要记录自己消费了多少数据,即消费位置信息。在Kafka中这个位置信息有个专门的术语:位移(offset)。
(1)、很多消息引擎都把这部分信息保存在服务器端(broker端)。这样做的好处当然是实现简单,但会有三个主要的问题:
1. broker从此变成有状态的,会影响伸缩性;
2. 需要引入应答机制(acknowledgement)来确认消费成功。
3. 由于要保存很多consumer的offset信息,必然引入复杂的数据结构,造成资源浪费。
而Kafka选择了不同的方式:每个consumer group管理自己的位移信息,那么只需要简单的一个整数表示位置就够了;同时可以引入checkpoint机制定期持久化,简化了应答机制的实现。
(2)、Kafka默认是定期帮你自动提交位移的(enable.a**.commit = true),你当然可以选择手动提交位移实现自己控制。
(3)、另外kafka会定期把group消费情况保存起来,做成一个offset map,如下图所示:
上图中表明了test-group这个组当前的消费情况。
老版本的位移是提交到zookeeper中的,目录结构是:/consumers/<group.id>/offsets/<topic>/<partitionId>,但是zookeeper其实并不适合进行大批量的读写操作,尤其是写操作。
__consumers_offsets topic配置了compact策略,使得它总是能够保存最新的位移信息,既控制了该topic总体的日志容量,也能实现保存最新offset的目的。compact的具体原理请参见: Log Compaction
至于每个group保存到__consumers_offsets的哪个分区,如何查看的问题请参见这篇文章: Kafka 如何读取offset topic内容 (__consumer_offsets)
offset提交消息会根据消费组的key(消费组名称)进行分区. 对于一个给定的消费组,它的所有消息都会发送到唯一的broker(即Coordinator)
Coordinator上负责管理offset的组件是 Offset manager 。负责存储,抓取,和维护消费者的offsets. 每个broker都有一个offset manager实例. 有两种具体的实现:
ZookeeperOffsetManager: 调用zookeeper来存储和接收offset(老版本的位移管理)。
DefaultOffsetManager: 提供消费者offsets内置的offset管理。
通过在config/server.properties中的offset.storage参数选择。
DefaultOffsetManager
除了将offset作为logs保存到磁盘上,DefaultOffsetManager维护了一张能快速服务于offset抓取请求的 consumer offsets表 。这个表作为缓存,包含的含仅仅是”offsets topic”的partitions中属于leader partition对应的条目(存储的是offset)。
对于DefaultOffsetManager还有两个其他属性: “offsets.topic.replication.factor和”offsets.topic.num.partitions”,默认值都是1。这两个属性会用来自动地创建”offsets topic”。
offset manager接口的概要:
什么是rebalance?
rebalance本质上是一种协议,规定了一个consumer group下的所有consumer如何达成一致来分配订阅topic的每个分区。比如某个group下有20个consumer,它订阅了一个具有100个分区的topic。正常情况下,Kafka平均会为每个consumer分配5个分区。这个分配的过程就叫rebalance。Kafka新版本consumer默认提供了两种分配策略:range和round-robin。
rebalance的触发条件有三种:
组成员发生变更(新consumer加入组、已有consumer主动离开组或已有consumer崩溃了——这两者的区别后面会谈到)
订阅主题数发生变更——这当然是可能的,如果你使用了正则表达式的方式进行订阅,那么新建匹配正则表达式的topic就会触发rebalance
订阅主题的分区数发生变更
refer
http://w*.cnb***.com/huxi2b/p/6223228.html
http://zqhxuyuan.github.io/2016/02/18/Kafka-Consumer-Offset-Manager/
http://w*.cnb***.com/byrhuangqiang/p/6384986.html