3.1.1创建并初始化消费者连接器

淩亂°似流年 2022-11-13 04:12 111阅读 0赞

> 消息系统由生产者、存储系统和消费者组成。上一章分析了生产者发送消息给服务端的过程，本章分析消费者从服务端存储系统读取生产者写入消息的过程。首先我们来了解消费者的一些基础知识。

1.  使用消费组实现消息队列的两种模式

作为分布式的消息系统，Kafka支持多个生产者和多个消费者，生产者可以将消息发布到集群中不同节点的不同分区上；消费者也可以消费集群中多个节点的多个分区上的消息。写消息时，多个生产者可以写到同一个分区。读消息时，如果多个消费者同时读取一个分区，为了保证将日志文件的不同数据分配给不同的消费者，需要采用加锁、同步等方式，在分区级别的日志文件上做些控制。

相反，如果约定“同一个分区只可被一个消费者处理”，就不需要加锁同步了，从而可提升消费者的处理能力。而且这也并不违反消息的处理语义：原先需要多个消费者处理，现在交给一个消费者处理也是可以的。图3-1给出了一种最简单的消息系统部署模式，生产者的数据源多种多样，它们都统一写人Kafka集群。处理消息时有多个消费者分担任务，这些消费者的处理逻辑都相同，每个消费者处理的分区都不会重复。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]  
实际应用中，Kafka集群的数据需要被不同类型的消费者使用，而不同类型的消费者处理逻辑不同。Kafka使用消费组的概念，允许一组消费者进程对消费工作进行划分。每个消费者都可以配置一个所属的消费组，并且订阅多个主题。Kafka会发送每条消息给每个消费组中的一个消费者迫二程（同一条消息广播给多个消费组，单播给同一组中的消费者）。被订阅主题的所有分区会平均地负载给订阅方，即消费组中的所有消费者。比如1个主题有4个分区，1个消费组有2个消费者，那么每个消费者都会分配到2个分区。

如图3-2所示，典型的Kafka集群部署方式会有多个消费组，并且每个消费组中也有多个消费者。这样既允许多种业务逻辑的消费组存在，也可以保证同一个消费组内的多个消费者协调士作，避免因一个消费组中只有一个消费者导致数据丢失。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
Kafka采用消费组保证了“一个分区只可被消费组中的一个消费者所消费”，这意味着：

(1)在一个消费组中，一个消费者可以消费多个分区。
    (2)不同的消费者消费的分区一定不会重复，所有消费者一起消费所有的分区。
    (3)在不同消费组中，每个消费组都会悄费所有的分区。
    (4)同一个消费组下消费者对分区是互斥的，而不同消费组之间是共享的。

比如，有两个消费者订阅了一个主题，如果这两个消费者在不同的消费组中，那么每个消费者都会获取到这个主题所有的记录·如果这两个消费者在同一个消费组中，那么它们会各向获取到一半的记录（两者的记录是对半分的，而且都不重复)。图3-3给出了多个消费者都在同一个消费组中（有图)，或者各向组成一个消费组（左图)的不同消费场景，这样Kafka也可以实现传统消息队列的发布一订阅  
模式和队列模式。

*  发布－订阅模式。同一条消息会被多个消费组消费，每个消费组只有一个消费者，实现广播。
 *  队列模式。只有一个消费组、多个消费者一条消息只被消费组的一个消费者消费，实现单播。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

1.  消费组再平衡实现故障容错

> 消费者是客户端的监务处理逻辑程序，因此要考虑消费者的故障容错。一个消费组有多个消费者，因此消费组需要维护所有的消费者。如果一个消费者看机了，分配给这个消费者的分区需要重新分配给相同组的其他消费者；如果一个消费者加入了同一个组，之前分配给其他消费组的分区需要分配给新加入的消费者。

一旦有消费者加入或退出消费组，导致消费组成员列表发生变化，消费组中所有的消费者就要执行再平衡（rebalance）工作。如果订阅主题的分区有变化，所有的消费者也都要再平衡。如图3-4所示，在加入一个新的消费者后，需要为所有的消费者重新分配分区，因此所有消费者都会执行再平衡。

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]  
消费者再平衡前后分配到的分区会完全不同，那么消费者之间如何确保各向消费消息的平滑过渡呢？假设分区Pl原先分配给消费者Cl，再平衡后被分配给消费者C2。如果再平衡前消费者Cl保存了分区Pl的消费进度，再平衡后消费者C2就可以从保存的进度位置继续读取分区凹，保证分区Pl不管分配给哪个消费者，消息都不会丢失，实现了消费者的故障容错。

1.  消费者保存消费进度

生产者的提交日志采用递增的偏移量，连同消息内容一起写入本地日志文件。生产者客户端不需要保存偏移量相关的状态，消费者客户端则要保存消费消息的偏移盘即消费进度。消费进度表示消费者对一个分区已经消费到了哪里。

由于消费者消费消息的最小单元是分区，因此每个分区都应该记录消费进度，而且消费进度应该面向消费组级别。假设面向的是消费者级别，再平衡前分区Pl只记录到消费者Cl中，再平衡后分区Pl属于消费者C2。但是这样一来，分区Pl和消费者C2之前没有记录任何信息，就无法做到无缝迁移。而如果针对消费组，因为消费者Cl和消费者C2属于同一个消费组，再平衡前记录分区Pl到消费组l，再平衡后消费者C2可以正常地读取消费组l的分区Pl进度，还是可以准确还原出这个分区在消费组l中的最新进度的。总结下，虽然分区是以消费者级别被消费的，但分区的消费进度要保存成消费组级别的。

`消费者对分区的消费进度通常保存在外部存储系统中`，比如ZK或者Kafka的内部主题  
(\_consumer＿offsets）。这样分区的不同拥有者总是可以读取同一个存储系统的消费进度，即使消费者成员发生变化，也不会影响消息的消费和处理。如图3-5所示，消费者消费消息时，需要定时将分区的最新消费进度保存到ZK中。当发生再平衡时，消费者拥有的新分区消费进度都可以从ZK中读取出来，从而恢复到最近的消费状态。

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

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

*  线程数量多于分区的数量，有部分钱程无法消费该主题下任何一条消息。
 *  线程数量少于分区的数量，有一些线程会消费多个分区的数据。
 *  线程数量等于分区的数量，则正好一个钱程消费一个分区的数据。

图3-6展示了上面这3种场景，正常情况下采用第二种是最好的，这种方案既不会有第一种的资源浪费现象存在，也不会像第三种那样每个线程只负责一点点工作。通过让一个线程消费多个分区，可以最大限度地利用每个线程的处理能力。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]

一个消费者线程消费多个分区，可以保证消费同一个分区的消息一定是有序的，但并不保证消费者接收到多个分区的消息完全有序。如图3-7所示，消费者分配了分区PO和分区Pl，虽然消费者收到的消息整体上不是有序的，但是针对同一个分区的消息是有序的。比如图3-7(左)中分区内的消息顺序(1)(2)(3)对应的消费者读取顺序也一定是(1)(2)(3),图3-7(右)中分区PO的消息顺序(1)(3)(5)对应的消费者读取顺序也一定是(1)(3)(5)。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]

1.  消费者与ZK的关系

消费者除了需要保存消费进度到ZK中，它分配的分区也是从ZK读取的。ZK不仅存储了Kafka的内部元数据，而且记录了消费组的成员列表、分区的消费进度、分区的所有者。表3-1总结了消息代理节点、主题、分区、消费者、偏移量（offset）、所有权（ownership）在ZK中的注册信息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]  
消费者要消费哪些分区的消息、自消费组来决定，因为消费组管理所有的消费者，所以它需要知道集群中所有可用的分区和所有存活的消费者，才能执行分区分配算法，而这些信息都需要保存到ZK中。每个消费者都要在ZK的消费组节点下注册对应的消费者节点，在分配到不同的分区后，才会开始  
各自拉取分区的消息。

通常，客户端代码并不直接完成上面那些复杂的操作步骤，而是由服务端暴露出一个API接口，让客户端可以透明地和集群交互。这个API接口实际上属于客户端进程范畴，用来和管理员以及数据存储节点通信。Kafka提供了两种层次的客户端API：如果消费者不太关心消息偏移量的处理，可以使用高级API；如果想自定义消费逻辑，可以使用低级API。

*  高级API。消费者客户端代码不需要管理偏移量的提交，并且采用了消费组的向动负载均衡功能，确保消费者的增减不会影响消息的消费。高级API提供了从Kafka消费数据的高层抽象。
 *  低级API。通常针对特殊的消费逻辑，比如消费者只想消费某些特定的分区。低级API的客户端代码需要自己实现一些和Kafka服务端相关的底层逻辑，比如选择分区的主剧本、处理主副本的故障转移等;

--------------------

### 3.1 消费者启动和初始化 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]  
消费者的配置信息要指定连接的ZK集群以及消费组编号。消费者客户端会通过消费者连接器(ConsumerConnector）连接ZK集群，获取分配的分区，创建每个主题对应的消息流（KafkaStream),最后迭代消息流，读取每条消息，并完成具体的业务处理逻辑（这里只是简单地打印出收到的每条信息）。

为了方便理解上面这些对象的产生意义，我们以JDBC连接查询数据库为例来展开介绍。客户端查询数据库返回结果集（ResultSet），通过迭代结果集，就可以返回数据库中的每条记录。而Kafka的消费者客户端做的事情实际上与此类似，前者查询数据库，后者查询消息系统；前者迭代结果集的每条记录，后者迭代ConsumerIterator的每条消费记录。相关代码如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 9]  
如图3-8所示，消费者客户端读取消息会创建消息流，然后从消息流中读取消息。那么为消费者分也配分区，以及拉取分区消息一定发生在这两个步骤之间，而实际上这些工作都会在消费者连接器中完成。消费者客户端通过消费者连接器读取消息的具体步骤如下 。

(1)消费者的配置信息指定订阅的主题和主题对应的线程数，每个线程对应一个消息流。
    (2)Consumer对象通过配置文件创建基于ZK的消费者连接器。
    (3)消费者连接器根据主题和线程数创建多个消息流。
    (4)在每个消息流通过循环消费者迭代器（ConsumerIterator)读出消息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]  
上面的消费者示例只为订阅的一个主题指定了一个消费线程，而一个消费者进程允许为不同的主题设置不同的消费者线程数量。比如，我们可以为重要的主题设置更多的线程数以加快消费进度，不同主题之间互相隔离，不会互相影响。而且即使是相同的主题，不同的线程之间也不像Java的多线程那样需要加锁等重量级的同步方案。因为消费者采用多线程访问的分区都是隔离的，所以不会1-H现同一个分区被不同线程同时访问的情况。前面我们说过“一个分区只会被分配给消费组中的一个消费者”，实际上从线程模型来看，更准确地讲应该是“一个分区只会被分配给一个消费者线程”。下面的代码为一个消费者设置了多个消费者线程，每个消息流使用线程池的方式运行：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 11]  
![在这里插入图片描述][20210325200732711.png]

消费者连接器不仅要处理分区分配和拉取消息这些和消费逻辑相关的任务，而且要在线程模型方面保证不同线程的处理逻辑都一致。如图3-9(左)所示－个消费者进程订阅的主题设置了3个线程。而图3-9(有)巾有3个消费者进程，但每个消费者订阅的主题只有一个线程。这两种方式的效果应该E…是一样的：郁能够消费所有的分区，只不过对于外部ZK而言，一个消费者和3个消费者在消费组看来…·消费组成员列表是不同的。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 12]

--------------------

#### 3.1.1创建并初始化消费者连接器 ####

消费者(Consumer)和消费者连接器接口(ConsumerConneetor)在同一个文件中，消费者只需要消费者配置(ConsumerConfig)就可以创建消费者连接器。消费者连接器接口的主要方法有：createmessageStreams()方法，它创建消息流并返回给客户端应用程序，这样客户端就会使用消息流读取消息；commitOffsets()方法，它会提交分区的偏移量元数据到ZK或者Kafka的内部主题中。相关代码如下：

![在这里插入图片描述][20210325201054802.png]

默认的消费者连接器实现类是ZookeeperConsumerConnector。除了上面暴露给客户端使用的两个方法，消费者连接器还会协调下面的各个组件来读取消息。

*  listeners。注册主题分区的更新、会话超时、消费者成员变化事件，触发再平衡。默认的消费者连接器实现类是ZookeeperConsuMerConnector。除了上面暴露给客户端使用的两个方法，消费者连接器还会协调下面的各个组件来读取消息。
 *  zkUtills。从ZK中获取主题、分区、消费者列表，为再平衡时的分区分配提供决策。
 *  topicRegistry。消费者分配的分区，结构是“主题→(分区→分区信息)”。
 *  fetcher。消费者拉取线程的管理类，拉取线程会向服务端拉取分区的消息。
 *  topicThreadldAndQueues。消费者订阅的主题和线程数，每个线程对应一个队列。
 *  offsetsChannel。偏移量存储为Kafka内部主题时，需要和管理消费组的协调者通信。

图3-10展示了消费者连接器中这些组件如何协调完成消息的消费。其中，监听器(I)是消息消费事件的导火索，一旦触发了再平衡，需要从ZK中读取所有的分区和已注册的消费者(2)。然后通过分区分配算法，每个消费者都会分配到不同的分区列表(3)。接着拉取线程开始拉取对应的分区消息(4)，并将拉取到的消息放到每个线程的队列中(5），最后消费者客户端就可以从队列中读取出消息了。另外，为了及时保存消费进度，我们还需要将偏移量保存至offsetsChannel通道对应的节点中(6)。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 13]

为了保证再平衡时各项准备工作都已就绪，创建消费者连接器时，需要执行以下初始化方法。

(1)确保连接上ZK，因为消费者要和ZK通信，包括保存消费进度或者读取分区信息等。
    (2)创建管理所有消费者拉取线程的消费者拉取管理器(ConsumerFetcherManager)。
    (3)确保连接上偏移量管理器(OffsetManager)，消费者保存消费进度到内部主题时和它通信。
    (4)调度定时提交偏移量到ZK或者Kafka内部主题的线程。

--------------------

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/da3c38560dec474bbb640203ea251a0f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/15f1fa34a407407c83f3eb4305fa465c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/1bdcf612775446518a96a3fc1d1757c7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/654e921f4c2f4d15874518329467f9bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/5f5d81cb6ee64ff7bc43294f79487b4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/d8cfcf2155704eeb915279ebbd3dee91.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 6]: /images/20221022/5b27ab10570f47dca779dde6ca395a60.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 7]: /images/20221022/44afde88a90f4933981ef9b5e226669a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 8]: /images/20221022/4c69154e6cc041f2b993d23a4ab7cbaa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 9]: /images/20221022/0487e63ce3804df5b738830d65d9a196.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 10]: /images/20221022/34c2dd9b31b54da980d266756ab494a2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 11]: /images/20221022/26fd2c193ae74424a7baa20d2b042999.png
[20210325200732711.png]: /images/20221022/22f4525b3f8945c28af52a3e53264c08.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 12]: /images/20221022/672aac91db7a433da3c5dc64d50fe4e1.png
[20210325201054802.png]: /images/20221022/7862822625d44f20b1aa54880a3f2837.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 13]: /images/20221022/c92444e135c844ffbaace7237e591db7.png