kafka之消费者基础知识一

一、消费者基础知识

1、读消息时，如果多个消费者同时读取一个分区，为了保证将日志文件的不同数据分配给不同的消费者，需要采用加锁、同步等方式，在分区级别的日志文件上做些控制。“同一个分区只可被一个消费者处理”，就不需要加锁同步。

2、每个消费者处理的分区都不会重复；

3、同一条消息广播给多个消费组，单播给同一组中的消费者。被订阅主题的所有分区会平均地负载给订阅方，即消费组中的所有消费者。

4、关于“一个分区只可被消费组中的一个消费者所消费”的解析：

 (1 ）在一个消费组中，一个消费者可以消费多个分区。

(2 ）不同的消费者消费的分区一定不会重复，所有消费者一起消费所有的分区。

(3 ）在不同消费组中，每个消费组都会消费所有的分区。

(4 ）同一个消费组下消费者对分区是互斥的，而不同消费组之间是共享的。

5、使用消费组实现消息队列的两种模式：

（1）、发布－订阅模式。同一条消息会被多个消费组消费，每个消费组只有一个消费者，实现广播。

（2）、队列模式。只有一个消费组、多个消费者一条消息只被消费组的一个消费者消费，实现单播。

6、消费组再平衡实现故障容错，发生再平衡的各种情况：

所谓的再平衡，指的是在kafka consumer所订阅的topic发生变化时发生的一种分区重分配机制。

（1）、有消费者加入或退出消费组，导致消费组成员列表发生变化，消费组中所有的消费者就要执行再平衡（ rebalance ） 工作。

（2）、订阅主题的分区有变化（发生了新增分区的行为），所有的消费者都要再平衡。

（3）、订阅的topic发生变化，采用的是正则表达式（test-*）的形式添加的主题形式

注意： 再平衡操作针对的是消费组中的所有消费者， 即所有消费者都妥执行重新分配分区的动作。

如下图，增多一个C3消费者，重新分配分区。

7、消费者保存消费进度

        消费进度表示消费者对一个分区已经消费到了哪里；生产者客户端不需要保存偏移量相关的状态，消费者客户端则要保存消费消息的偏移盘即消费进度。

注：虽然分区是以消费者级别被消费的，但分区的消费进度要保存成消费组级别的。

8、消费者对分区的消费进度通常保存在外部存储系统中，比如ZK 或者Kafka 的内部主题( consumer_offsets ）。

        消费者保存消费进度的另一个原因是： 消费者消费消息是主动从服务端拉取数据，而不是由服务端向消费者推送数据。如果由服务端推送数据给消费者，消费者只负责接收数据，就不需要保存状态了。但后面这种方法会严重影响服务端的性能，因为要在服务端记录每条消息分配给哪个消费者，还要记录消费者消费到哪里了。

9、一个分区只能属于一个消费者线程，将分区分配给消费者有以下几种场景：

（1）、线程数量多于分区的数量，有部分钱程无法消费该主题下任何一条消息。

（2）、线程数量少于分区的数量，有一些线程会消费多个分区的数据。

（3）、线程数量等于分区的数量，则正好一个钱程消费一个分区的数据。

注：一个消费者线程消费多个分区，可以保证消费同一个分区的消息一定是有序的，但并不保证消费者接收到多个分区的消息完全有序。

10、消费者与ZK的关系

（1）、ZK不仅存储了Kafka的内部元数据，而且记录了消费组的成员列表、分区的消费进度、分区的所有者。

（2）、消费组决定消费者要消费哪些分区的信息。每个消费者都要在ZK的消费组节点下注册对应的消费者节点，在分配到不同的分区后，才会开始各自拉取分区的消息。

二、消费者启动和初始化

1、消费者执行过程：消费者客户端会通过消费者连接器（ConsumerConnector）连接ZK集群，获取分配的分区， 创建每个主题对应的消息流（KafkaStream),最后迭代消息流，读取每条消息。

(1 ）消费者的配置信息指定订阅的主题和主题对应的线程数， 每个线程对应一个消息流。

(2) Consumer对象通过配置文件创建基于ZK的消费者连接器。

(3）消费者连接器根据主题和线程数创建多个消息流。

(4）在每个消息流通过循环消费者迭代器（ConsumerIterator）读出消息。

2、消费者连接器(默认实现类是ZookeeperConsumerConnector)，接口方法如下:

createMessageStreams()方法，它创建消息流并返回给客户端应用程序， 这样客户端就会使用消息流读取消息；

commitOffsets() 方法，它会提交分区的偏移量元数据到ZK 或者Kafka的内部主题中。

需要以下组件辅助来读取信息：

listeners: 注册主题分区的更新、会话超时、消费者成员变化事件，触发再平衡。

zkUtils: 从ZK 中获取主题、分区、消费者列表，为再平衡时的分区分配提供决策。

topicRegistry: 消费者分配的分区，结构是“主题→（分区→分区信息）” 。

fetcher: 消费者拉取线程的管理类，拉取线程会向服务端拉取分区的消息。

topicThreadIdAndQueues: 消费者订阅的主题和线程数，每个线程对应一个队列。

offsetsChannel : 偏移量存储为Kafka 内部主题时，需要和管理消费组的协调者通信。

消费者连接器中这些组件协调完成消息的消费过程:

(I )、注册监听器，是消息消费事件的导火索。

(2) 、一旦触发了再平衡，需要从Z K 中读取所有的分区和已注册的消费者 。

(3)、然后通过分区分配算法，每个消费者都会分配到不同的分区列表 。

(4)、接着拉取线程开始拉取对应的分区消息(从kafka集群拉取)。

(5)、将拉取到的消息放到每个线程的队列中，最后消费者客户端就可以从队列中读取 消息了。

(6)、为了及时保存消费进度，我们还需要将偏移量保存至offsetsChannel通道对应的节点中。

创建消费者连接器时需要初始化的以下方法:

(1)、确保连接上ZK ， 因为消费者要和ZK通信，包括保存消费进度或者读取分区信息等。

(2)、创建管理所有消费者拉取线程的消费者拉取管理器ConsumerFetcherManager。

(3 ）确保连接上偏移量管理器 OffsetManager，消费者保存消费进度到内部主题时和它通信。

(4 ）调度定时提交偏移量到ZK或者Kafka 内部主题的线程。

3、消费者客户端的线程模型

createMessageStreams()的 consume()的操作逻辑:

(1 ）根据客户端传入的topicCountMap构造对应的队列和消息流，消息流引用了队列。

(2 ）在ZK 的消费组父节点下注册消费者子节点。

(3 ）执行初始化工作，触发再平衡，为消费者分配分区，拉取线程会拉取消息放到队列中。

(4）返回消息流列表，队列rp 有数据时，客户端就可以从消息流中迭代读取消息。

消费者线程模型主要概念有：消费者线程、队列、消息流。模型相关的变量：

topicCountMap : 设置主题及其对应的线程个数， 每个钱程都对应一个队列和一个消息流。

consumerId:即“消费者编号”，用“消费组名称＋随机值”表示，指定消费者在消费组中的唯一编号。

CoonsumerThreadId：即“消费者线程编号”，用“ 消费者编号＋线程编号”表示。

consumerThreadIdsPerToicMap: 表示每个主题和消费者线程编号集合的映射关系。

topicThreadIds: 表示所有的消费者线程编号集合，相同主题的线程会在同一个数组里。

topicThreadIdAndQueues: 表示消费者线程和队列的映射关系，因为每个线程对应一个队列。

客户端的关注点是·每个线程都对应一个队列，每个队列都对应了一个消息流，只要队列中有数据，就能从消息流中迭代读取出消息。

三、消费者再平衡操作

1、触发消费者连接器执行再平衡操作有两种方式：外部事件和直接触发。

直接触发:会在消费者启动时执行，即重新初始化消费者时，直接调用syncedRebalance （） 方法强制触发一次再平衡。

外部事件:会通过下面3种监昕器和线程检查的方式触发再平衡。

(1)、ZKSessionExpireListener：当新的会话建立或者会话超时需要重新注册消费者，并调用syncedRebalance()触发再平衡。

(2)、ZKTopicPartitionChangeListener：当主题的分区数量变化时，通过rebalanceEvenTriggered触发再平衡。

(3)、ZKRebalanceListener：当消费组成员变化时， 通过rebalanceEvenTriggered发再平衡。

再平衡示意图：

消费者执行再平衡和提交偏移量都直接和协调者交互，具体步骤如下：

(I ）每个消费者触发再平衡时都和协调者联系，由协调者执行全局的分区分配。

(2）协调者分配完成后，将分区分配给每个消费者。

(3）每个消费者收到任务列表后，启动拉取钱程，拉取对应分区的消息，并更新拉取状态。

(4）消费者周期性提交分区的偏移量给协调者，协调者将分区偏移量写到内部主题。

ZKRebalancerListener.rebalance()执行再平衡的步骤：

        (1) 关闭数据拉取线程，清空队列和消息流，提交偏移量。

(2）释放分区的所有权，删除ZK中分区和消费者的所有者关系。

(3）将所有分区重新分配给每个消费者，每个消费者都会分到不同的分区。

(4）将分区对应的消费者所有者关系写入ZK ， 记录分区的所有权信息。

(5）重新启动消费者的拉取线程管理器，管理每个分区的拉取线程。

拉取线程和分区所有权的关闭和开启顺序为： 停止拉取钱程→释放分区的所有权→添加分区的所有权→启动拉取线程。

2、分区所有权

       分区的所有权记录在ZK的/consumers/[group_id]/owner/[topic]/[partition_id]-->consumer_thread_id节点，表示“主题一分区”会被指定的消费者线程所消费，或者说分区被分配给消费者、消费者拥有了分区。要释放分区的所有权，只需要删除分区对应的ZK节点；要重建分区的所有权，数据源中除了包含分区，还要有消费者线程编号。

注：ZK中不仅记录了消费者和分区的所有权映射关系，而且记录了消费组的消费者列表、主题的分区列表，这些信息为消费者分配分区提供了数据来源。

3、为消费者分配分区

从ZK中查询出所有的分区以及所有的消费者成员列表。

分区分配算法(针对的是消费组的所有消费者)：有2个消费者， 每个消费者都有2个线程， 一共有5个可用的分区。每个消费者线程（ 一共4个线程）都可以获取至少l个分区（ 514=1 ） ， 剩余l个（ 5%4=1 ）分区分给第一个线程。

rebalance()的分区分配的一些步骤：

( 1 ）构造消费者的分配上下文，得到订阅主题的分区和所有的消费者线程信息。

(2） 分区分配算法计算每个消费者的分区和消费者线程的映射关系。

(3 ）从步骤（2）的全局结果中获取属于当前消费者的分区和消费者线程。

(4） 读取当前消费者拥有的分区在ZK中的最新消费进度， 即它所拥有分区的偏移量。

(5 ） 构造PartitionTopicInfo， 加入到表示消费者的主题注册信息的topicRegistry中。

(6） 更新topicRegistry ，后面的拉取线程会使用该数据结构。

4、创建分区信息对象

       从ZK中读取出的分区的偏移量， 会被用来构造分区信息对象PartitionTopicInfo。分区信息对象的主要内容有：分区，表示拉取线程的“目标” ；队列，作为消息的“存储”介质； 偏移量， 作为拉取“状态” 。消费者的拉取线程会以最新的“状态”拉取“目标”的数据填充到“存储”队列中。

拉取偏移量（ fetchOffset ）：表示要从哪里开始拉取。

消费偏移量（ consumedOffset ）：表示消费到了哪里。

队列从创建到填充数据，再到数据被消费的过程，具体步骤如下：

(1 ）连接器根据订阅信息生成队列和消息流的映射，并且队列也会传给消息流。

(2） 为消费者分配分区时，会从ZK中读取分区消费到的最新位置。

(3 ） 根据偏移量创建分区信息， 队列也会传给分区信息对象。

(4）分区信息被用于消费者的拉取线程。

(5）拉取线程从服务端的分区拉取消息。

(6）消费者拉取到消息后， 会将最新的偏移量更新到ZK 。

(7 ）拉取线程将拉取到的消息填充到队列里。

(8 ）消息流可以从队列里获取消息。

(9）应用程序从消息流里迭代获取消息。

分区信息中队列的数据来源路线图：

 分区信息和客户端线程模型的关系图 ：

注意： 拉取线程和分区并不存在直接关联，而是通过负责管理所有拉取线程的消费者拉取线程管理器进行关联。

5、关闭和更新拉取线程管理器

       关闭拉取管理器时提交偏移量和写文件的过程类似： 每条数据迄加到文件中并没有立即刷写到磁盘， 而是先写到磁盘缓存中， 然后定时地刷写到磁盘上， 最后在关闭文件时也要刷写一次磁盘。如果没有最后一次的强制和l 写， 有可能会导致仍然还在磁盘缓存中的数据丢失。

6、分区信息对象的偏移量

分区信息对象的偏移量在拉取线程的具体操作步骤:

(1 ）关闭拉取线程时提交consumedOffset偏移量到ZK 。

(2 ）重新启动拉取线程时读取ZK 中的偏移量。

(3 ）将ZK 的偏移量作为刚开始的fetchedOffset 。

(4 ）客户端读取到消息后会更新consumedOffset 。

(5 ）在这之后每次拉取使用的fetchedOffset 都来向于最新的consumedOffset.

(6 ）客户端进程定时提交偏移i注和(1)类似，也是取consumedOffset 写到Z K 中。

消费者客户端使用消费者连接器的主要工作，具体步骤如下:

( 1 ）创建队列和消息流，前者用于保存消费者拉取的消息，后者会读取消息。

(2 ）注册各种事件的监昕器，当事件发生时，消费组所有消费者成员都会再平衡。

(3 ）再平衡会为消费者重新分配分区，并构造分区信息加入topicRegistry。

(4 ）拉取线程获取topicRegistry中分配给消费者的所有分区信息开始工作。

四、消费者拉取数据

1、消费者的拉取管理器（ConsumerFetcherManager）:管理了当前消费者的所有拉取线程。

 2、拉取线程的拉取状态

       拉取管理器的LeaderFinderThread后台线程会将分区添加到对应的拉取钱程。具体过程为： 后台线程选择有主副本的分区后，将分区添加到所属的拉取线程；如果拉取线程不存在，就会创建拉取线程，再将分区添加到拉取线程中。抽象的拉取线程中有一个表示分区及其对应分区拉取状态的partitionMap变量，它的含义和抽象拉取管理器中表示分区及其对应分区信息的partitionMap不同。具体步骤如下：

( 1 ）后台线程调用抽象拉取管理器的addFetcherPartitions() 方法。

(2) addFetcherPartitions() 方法调用createFetcherThread()抽象方法。

(3 ）消费者拉取管理器的createFetcherThread()创建具体的消费者拉取线程。

(4 ）消费者拉取管理器调用抽象拉取线程的addPartitons()将分区添加到步骤（3 ）的拉取钱程。

3、消费者拉取线程

分区信息的队列保存拉取的消息：

从分配分区给消费者，到拉取线程拉取消息返回给消费者的具体步骤如下：

(1)、再平衡操作将分区分配给消费者，读取ZK 的偏移量作为分区信息的拉取偏移量。

(2）分区信息的队列用来存储结果数据，拉取偏移量作为拉取线程初始的拉取位置。

(3） 拉取线程拉取分区的数据，初始时从拉取偏移量开始拉取消息。

(4) partitionMap表示分区的最新拉取状态，每次拉取数据后都要更新拉取状态。

(5 ）拉取线程创建拉取请求，并通过SimpleConsumer发送请求和接收响应结果。

(6 ）拉取钱程技取到分区消息后， 将分区数据的消息集填充到分区信息对象的队列。

(7 ） 创建消费者连接对象时，会创建队列和消息流， 一个队列关联了一个消息流。

(8 ）消费者客户端从消息流中迭代读取结果数据，实际上就是从队列中拉取消息。

五、消费者消费信息

一个队列包含多个数据块，每个数据块对应一个分区的消息集， 一个消息集包含多条消息。

1、消费者迭代消费信息

       消费者的“拉取线程” 拉取消息后会更新“拉取状态” ，对应的“消费线程”获取消息后也要更新相关的“消费状态” 。（准确地说，消费消息的对象是一个迭代器而不是钱程。这里为了和拉取线程相对应，故叫作消费线程。） 拉取状态对应分区信息、对象的拉取偏移量 fetchedOffset ， 表示消费者已经拉取的分区位置； 消费状态对应了消费偏移量consumededOffset ，表示消费者已经消费完成的偏移量。

“拉取线程更新拉取偏移量，消费线程更新消费偏移量”，具体步骤如下：

(1 ） 消费者的拉取线程从服务端拉取分区的消息。

(2 ）拉取到分区消息后，就更新分区信息对象的拉取偏移量。

(3 ）将分区数据的消息集封装成数据块。

( 4 ） 客户端循环迭代数据块的消息集。

(5） 消费完一条消息后，就更新分区信息对象的消费偏移量。

(6）消息流中的每一条消息返回给消费者客户端应用程序。

六、消费者提交分区偏移量

1、提交偏移量到ZK

(1)、分区的来源是再平衡后分配给当前消费者的topicRegistry， 消费者负责哪些分区，相应地就应该提交哪些分区的偏移量。

(2)、偏移量表示分区的消费进度，来自于分区信息对象的consumedOffset变量，这个变量会在迭代消费者迭代器的每一条消息时更新。

注意： 分区信息主要包含3 个变量：队列用来存储拉取到的消息， fetchedOffset用来确定拉取的起始位直， consumedOffset表示消费过的位直，用来记录分区的消费进度。

2、提交偏移量到内部主题

偏移量的多种连接方案:

(I ）如果不同分区的偏移量写到了不同的节点，消费者分配了多个分区，当要读取不同分区的偏移量时，就得连接不同的节点才可以获得完整的数据。

(2）如果能让所有分区的偏移量数据只保存在一个节点，消费者就只需要同一个节点通信。但因为消费者和分区的关系是变化的， 即使保证这一次分区在一个节点上， 也无法保证下一次仍然在同一个节点。

(3 ）如果消费组所有消费者所有分区的偏移量都保存在一个节点，就可以解决第二种方式的问题。

(4） 实际上， 消费者的分区偏移量要保存在哪个节点，跟消费者所属的消费组有关系。只要保证消费组级别的偏移i量在一个节点上， 即使消费者和分区的关系发生变化， 也能够保证消费者访问新分配的分区时， 只需要访问一个节点。

偏移量管理器（ OffsetManager）：每个消费组只连接一个节点是最好的，这个节点负责管理一个消费组所有消费者所有分区的偏移量。

客户端需要确定服务端节点的几个场景：

(1)、生产者发送消息时，直接在客户端决定消息要发送给哪个分区，这一步不向服务端发送请求。

(2)、消费者拉取管理器的LeaderFinderThread 线程向服务端发送主题元数据请求，获取包含了主副本等信息的所有分区元数据，消费者拉取线程才能确定要连接哪些服务端节点。

(3)、提交偏移茸虽然有点像生产者的发送消息，都是写数据，但也需要和消费者的LeaderFinderThread 一样，获取分区的主副本作为偏移ill:管理器，才能确定提交到哪个节点。

3、服务端处理提交偏移量的请求

消费者发送提交偏移量和获取偏移量都会被j服务端的KafkaApis处理，服务端处理

这两个请求的具体步骤如下：

( 1) KafkaApis将提交偏移量请求的处理交给消费组的协调者GroupCoordinator。

(2 ）消费组的协调者再交给消费组的元数据管理类GroupMetadataManager去处理。

(3 ）延迟的存储对象DelayedStore会调用副本管理器的appendMessages()存储消息。

(4）副本管理器将消息追加到底层文件系统的日志文件中，这样分区的偏移量就抒储到服务端了。

(5 ）分区和l对应的偏移量会在消息存储成功后，被缓存至服务端的消费组元数据管理类。

(6 ）服务端处理客户端的获取分区偏移盐请求， 会首先从缓存中获取。

(7 ）如果缓存中没有分区的偏移量， 就从日志文件中读取。

消费者提交偏移量的过程:

(1)、消费者分配到分区， 比如消费者l(C1）分配到主题（test1）的分区PO和分区Pl 。

(2 ）分区PO 的主副本是消息代理节点1 ( Broker1 ），分区Pl 的主副本是消息代理节点2 ( B rok e r2 ),消费者创建拉取线程拉取分区消息。

(3 ）消费者拉取到每个分区的消息后，客户端迭代每条消息，会更新分区信息对象的消费进度。

( 4 ）消费者定时提交分区偏移量， 连接消费组的协调节点，消费组l 对应内部主题的P 1 即Broker2 。

(5 ）消费者l 将向己负责的分区（即PO 和Pl ）偏移革A提交到协调节点B ro k e r2 上。

 4、缓存分区的偏移量

主题元数据（ TopicMetadata）和消费组的协调者（GroupCoordinator）因为在每个服务端节点保存的数据都一样，可以请求任何一个节点，所以是所有节点共享的缓存

注意：只要消费组所有消费者都提交了分区的消费进度，再平衡时无论怎么重新

分配分区，任何一个消费者都可以查询到任意一个分区的最新消费进度。

七、消费者API

1、api工作过程：

 

 

来源于《Kafka技术内幕：图文详解Kafka源码设计与实现.郑奇煌》学习笔记