1、Consumer与Consumer Group

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)

图片.png

2、Coordinator介绍

Coordinator一般指的是运行在broker上的group Coordinator，用于管理Consumer Group中各个成员，每个KafkaServer都有一个GroupCoordinator实例，管理多个消费者组，主要用于offset位移管理和Consumer Rebalance。

Coordinator存储的信息

对于每个Consumer Group，Coordinator会存储以下信息：

1. 对每个存在的topic，可以有多个消费组group订阅同一个topic(对应消息系统中的广播)
2. 对每个Consumer Group，元数据如下：
   订阅的topics列表
   Consumer Group配置信息，包括session timeout等
   组中每个Consumer的元数据。包括主机名，consumer id
   每个正在消费的topic partition的当前offsets
   Partition的ownership元数据，包括consumer消费的partitions映射关系

如何确定consumer group的coordinator

consumer group如何确定自己的coordinator是谁呢？ 简单来说分为两步：

1. 确定consumer group位移信息写入__consumers_offsets这个topic的哪个分区。具体计算公式：
   __consumers_offsets partition# = Math.abs(groupId.hashCode() % groupMetadataTopicPartitionCount) 注意：groupMetadataTopicPartitionCount由offsets.topic.num.partitions指定，默认是50个分区。
2. 该分区leader所在的broker就是被选定的coordinator

Coordinator工作过程

3、offset位移管理

消费者在消费的过程中需要记录自己消费了多少数据，即消费位置信息。在Kafka中这个位置信息有个专门的术语：位移(offset)。
(1)、很多消息引擎都把这部分信息保存在服务器端(broker端)。这样做的好处当然是实现简单，但会有三个主要的问题：
1. broker从此变成有状态的，会影响伸缩性；
2. 需要引入应答机制(acknowledgement)来确认消费成功。
3. 由于要保存很多consumer的offset信息，必然引入复杂的数据结构，造成资源浪费。
而Kafka选择了不同的方式：每个consumer group管理自己的位移信息，那么只需要简单的一个整数表示位置就够了；同时可以引入checkpoint机制定期持久化，简化了应答机制的实现。
(2)、Kafka默认是定期帮你自动提交位移的(enable.auto.commit = true)，你当然可以选择手动提交位移实现自己控制。
(3)、另外kafka会定期把group消费情况保存起来，做成一个offset map，如下图所示：

图片.png

上图中表明了test-group这个组当前的消费情况。

offset管理

老版本的位移是提交到zookeeper中的，目录结构是：/consumers/<group.id>/offsets/<topic>/<partitionId>，但是zookeeper其实并不适合进行大批量的读写操作，尤其是写操作。

图片.png

图片.png

图片.png

图片.png

__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接口的概要：

图片.png

图片.png

图片.png

3、Consumer Reblance

什么是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
订阅主题的分区数发生变更

图片.png

图片.png

Paste_Image.png

Paste_Image.png

Paste_Image.png

Paste_Image.png

Paste_Image.png

Paste_Image.png

Paste_Image.png

refer
http://www.cnblogs.com/huxi2b/p/6223228.html
http://zqhxuyuan.github.io/2016/02/18/Kafka-Consumer-Offset-Manager/
http://www.cnblogs.com/byrhuangqiang/p/6384986.html