kafka之消费者基础知识一

发布时间:2024-01-05 15:00

一、消费者基础知识

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

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

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

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

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

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

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

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

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

(1)、发布-订阅模式。同一条消息会被多个消费组消费,每个消费组只有一个消费者,实现广播。

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

kafka之消费者基础知识一_第1张图片

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

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

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

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

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

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

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

kafka之消费者基础知识一_第2张图片

7、消费者保存消费进度

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

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

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

kafka之消费者基础知识一_第3张图片

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

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

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

(2)、线程数量少于分区的数量,有一些线程会消费多个分区的数据。

(3)、线程数量等于分区的数量,则正好一个钱程消费一个分区的数据。

kafka之消费者基础知识一_第4张图片

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

kafka之消费者基础知识一_第5张图片

10、消费者与ZK的关系

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

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

二、消费者启动和初始化

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

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

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

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

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

kafka之消费者基础知识一_第6张图片

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

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

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

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

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

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

topicRegistry: 消费者分配的分区,结构是“主题→(分区→分区信息)” 。

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

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

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

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

kafka之消费者基础知识一_第7张图片

(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: 表示消费者线程和队列的映射关系,因为每个线程对应一个队列。

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

kafka之消费者基础知识一_第8张图片

三、消费者再平衡操作

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

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

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

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

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

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

kafka之消费者基础知识一_第9张图片

再平衡示意图:

kafka之消费者基础知识一_第10张图片

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

(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 )分区分给第一个线程。

kafka之消费者基础知识一_第11张图片

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)应用程序从消息流里迭代获取消息。

kafka之消费者基础知识一_第12张图片

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

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

kafka之消费者基础知识一_第13张图片

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

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

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

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

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

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

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

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

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

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

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

kafka之消费者基础知识一_第14张图片

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

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

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

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

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

四、消费者拉取数据

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

kafka之消费者基础知识一_第15张图片

 2、拉取线程的拉取状态

kafka之消费者基础知识一_第16张图片

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

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

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

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

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

3、消费者拉取线程

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

kafka之消费者基础知识一_第17张图片

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

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

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

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

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

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

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

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

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

kafka之消费者基础知识一_第18张图片

五、消费者消费信息

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

kafka之消费者基础知识一_第19张图片

1、消费者迭代消费信息

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

拉取线程更新拉取偏移量,消费线程更新消费偏移量”,具体步骤如下:

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

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

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

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

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

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

kafka之消费者基础知识一_第20张图片

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

1、提交偏移量到ZK

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

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

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

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

偏移量的多种连接方案:

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

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

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

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

偏移量管理器( OffsetManager):每个消费组只连接一个节点是最好的,这个节点负责管理一个消费组所有消费者所有分区的偏移量。

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

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

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

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

kafka之消费者基础知识一_第21张图片

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

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

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

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

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

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

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

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

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

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

kafka之消费者基础知识一_第22张图片

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

(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 上。

kafka之消费者基础知识一_第23张图片

 4、缓存分区的偏移量

kafka之消费者基础知识一_第24张图片

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

kafka之消费者基础知识一_第25张图片

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

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

七、消费者API

1、api工作过程:

kafka之消费者基础知识一_第26张图片

kafka之消费者基础知识一_第27张图片

 

 

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

 

 

 

 

 

ItVuer - 免责声明 - 关于我们 - 联系我们

本网站信息来源于互联网,如有侵权请联系:561261067@qq.com

桂ICP备16001015号