3.6.1 消息消费主流程

清疚 2022-11-13 07:29 120阅读 0赞

### 3.6 消费者低级API示例 ###

上面基于消费者连接器的实现是高级API的使用方法，消费者拉取线程最终通过SiMpleConsul’ler和服务端进行数据传输，底层网络连接采用了同步类型的阻塞通道。虽然SiMpleConsul’ler从名称上看属于“简单”、低级API，但是高级API也会使用这个类。只不过高级API并不会把这个底层的类暴露给消费者客户端应用程序，而低级API则需要直接面对SiMpleConsul’ler类编程。

客户端使用低级API消费消息，需要手动指定固定的分区，并且不会利用“消费组向动分配分区”即“消费组再平衡”的功能。如果分区所属的消费者挂掉了，分区不会分配给其他的消费者。

低级API没有复杂的线程模型，没有基于ZK的消费者连接器，没有分区信息和队列，没有消息流和消费者迭代器等各个组件的协调工作。虽然低级API没有运用复杂的数据结构，矛干起来比较简单、原始，但它仍然满足了消息消费的完整流程：寻找分区的主副本，读取分区上一次的消费进度，拉取消息（构建拉取请求、发送请求），读取消息。这些步骤都可以基于Sil’lpleConsul’ler类来完成，因为它的底层提供了网络连接和请求的发送接收，而任何客户端要和服务端通信，网络连接处理都是必不可少的。

表3-4列举了消费者客户端应用程序使用低级API消费消息的主要步骤，这些步骤在高级API中都能找到对应的方法。在高级API中由消费者拉取线程负责调用SiMpleConsuMe,i;，所以下面的步骤都可以参考消费者拉取线程的对应方法。当然，消费者拉取钱程继承了抽象拉取线程类，如果要对应完整的拉取流程，则还要把抽象拉取线程类结合起来，因为主要的处理流程都定义在抽象拉取钱程类中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

在高级APl中除了拉取线程，还有消费线程即消费者迭代器，以及队列作为两个线程之间的数据中转站。而低级API没有使用队列，所以在拉取到消息之后，应该立即处理返回的消息集。其实’不管是低级API还是高级API，最终发送给服务端的拉取请求应该都是一样的。对于服务端而言，它不需要关心客户端采用哪种API发送拉取请求，而只要能够返回指定分区的消息、集给客户端就可以。如罔3-30所示，两种API者~使用SiMpleConsumer和服务端通信，只不过在这之前的拉取和之后的消费方封不相同。

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

如图3-31所示，消费者拉取分区消息，必须指定从分区的哪个位置开始读取。高级API拉取线程从ZK或者消费组的协调节点获取偏移量，并通过51.f"lpleConsuf"ler向分区的主副本节点拉取消息，偏移量的存储节点和分区的数据节点可能不是在同一个节点。低级API没有消费组的概念，获取分区最近的偏移量也是通过SUipleConsuf"ler发送OffsetRequest给分区的主副本节点，然后还是通过51.f"lpleConsuf"ler拉取消息，说明偏移量和消息都在同一个节点。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]

在低级API巾为了保证程序的健壮性，不仅仅要满足消息的整个消费流程能够正常走通，还要处理一些异常情况。比如拉取请求发送给分区的主副本但是分区的主副本暂时不可用怎么办？如果拉取结果有错误，服务端返回的错误码是“偏移盘超过范围”，是否需要重置读取的偏移量？还有，异常情况下代表和l服务端连接的51.f"lpleConsuf"ler对象要不要进行切换？下面我们分析低级API的实现。

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

#### 3.6.1 消息消费主流程 ####

下面的示例中，run()方法指定了消费者要消费的分区编号和初始连接的种子节点(seedBrokers)。由于消费者客户端只允许和分区主副本所在的节点通信，所以要先找出分区的主副本，然后创建S"i.1’1pleConsuf11er对象代表消费者客户端和主副本节点的网络连接。

正常拉取消息没有错误的话，会从拉取响应中获取分区的消息集。消息集包含不止一条消息，我们循环处理每条消息，并增加读取偏移盐(readOffset)，减少fllaxReads计数器。如果没有达到读取盘(即还需要继续读取)，则会继续以新的读取偏移量构建拉取请求，再次拉取分区的消息。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]  
注意：低级API更新读取偏移量(readOffset)和高级API的拉取偏移量(fetchOffset)类似。高级API的拉取线程以最新的parti.ti.onMap创建拉取请求，拉取到消息后再次更新parti.t"i.onMap的分区拉取状态。不管哪种API拉取都妥指定读耳又位直，拉取过一次后就不应该重复拉取已经拉取过的消息。

如果拉取发生错误，一定不会执行for循环部分的读取消息，因为错误情况下fetchResponse不会有消息集结果。如果说出错了还能够处理消息的话，读取偏移量会被更新，而这部分数据实际上是有问题的，只有正常情况才可以更新读取偏移量。错误有几种情况，如果分区的主副本节点没有问题，仅仅是客户端发送的拉取请求中的读取偏移量超过服务端的范围，此时应该重置读取偏移量到服务端最近的偏移量。比如服务端的最大偏移量是10，而客户端却请求了20，应该重置为10。

错误也可能是主副本节点出现故障无法连接，此时应该调用fi.ndNewleader()连接新的主副本。注意，客户端没有权利选举分区的主副本，这是服务端的业务范畴：服务端的所有副本会在主副本出现故障时，选择一个新的主副本(服务端负责选举，客户端只负责查询)。选举主副本的过程肯定需要一段时间，客户端如果没有查询到新的主副本，就要过会儿再重试。一旦主副本发生故障，在拉取消息之前创建的Si.MpleConsuMer就要及时关闭，待选举出新的主副本后，创建新的Si.MpleCo『1Sl』开始新的？肖息拉取。相关代码如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]

如图3-32所示，考虑了主副本发生切换，低级API的整个工作过程比较复杂。正常情况下会走“粗线”的消息消费路线，主副本出现异常时走“虚线”连接新的主副本，，当结束时走“点线”关闭Si.MpleConsul’ler结束流程，具体步骤如下。

(1)第一次fi.ndleader()获取分区的主副本[A]，并创建Si.MpleConsuMer[BJ。
    (2)通过Si.MpleConsuMer[B]获取读取偏移量，构建拉取请求，并向主副本[A]发送请求。
    (3)拉取发生错误时，关闭Si.叩leConsul'ler[BJ，重置consul'ler＝null。
    (4)重新获取分区的leadBroker＝主副本[C]，并继续循环，因为拉取出错，并不会消费消息。
    (5)consuMer为空，使用步骤(4)的主副本[C]，创建新的Si.MpleConsuMer[DJ。
    (6)步骤(2)获取了读取偏移盘，而且上一次拉取出错并没有更新，使用读取偏移量构建拉取请求。
    (7)通过Si.MpleConsuMer[DJ向主副本[C]发送拉取请求。
    (8)拉取正常，消费消息集的每条消息，更新读取偏移量和减少MaxReads计数器。
    (9)使用上一步最新的读取偏移量创建新的拉取请求，并通过Si.MpleConsuMer向分区的主副本节点发送拉取请求，如此循环下去，直到计数器等于日，拉取流程结束。

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

上面分析了消费者低级API的整体流程，但是具体的一些细节还需要深入研究，比如找出分区的主副本、读取分区最近的偏移量、发送拉取请求并消费拉取结果。下面我们一－来分析这些实现细节。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/4133e3d95b784262b8669d0ce73d1b86.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/2edc6d4382c84be5828a8f6b3906359b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 2]: /images/20221022/0523ee2ad1464b9aa48d03d1b6617088.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 3]: /images/20221022/75e1067c2a2d43148e09e7d91792b52f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 4]: /images/20221022/605afbad60ee42b392a64e0a24d261d1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 5]: /images/20221022/fbafbe5b3aa44ada9c70dc95003782a7.png