Kafka系列：消息发送时，网络“偷偷”帮忙做的那点事儿

前言

上篇文章讲述了消息从生产到写入到 Broker 的 partition 上背后发生的故事，并提出了消息发送的网络模型的问题。本篇文章我们来尝试揭开其背后的神秘面纱，耐心看完你一定会有所收获。（本文来自公众号：z小赵）

文章概览Sender 线程的建连准备阶段和发送网络请求两阶段。Selector 选择器处理网络请求过程。Sender 线程的两阶段

上篇文章结尾提到了三个重要的方法，分别是 ready()、send()、poll()。其中 ready()和 send()可以理解为第一阶段，即建连准备阶段；poll()可以理解为第二阶段，即发送网络请求阶段。接下来对这两阶段做深入研究。

阶段流程说明

ready()阶段： 遍历节点列表，查询当前是否已建立连接，若已完成建连，则认为该节点可用；若还未建连，则判断该节点是否可以被连接，若是则建连。对于不可建连和正在建连的节点暂时还不能参与网络数据传输请求。send()阶段： 通过 ready()阶段拿到了已经完成建连的节点，然后遍历节点，判断当前节点是否可以被发送数据，若可以则将当前节点对应的 RequestChannel 加入到 InFlightRequest 双端队列中去。为什么要将 RequestChannel 加入到一个双端队列中去呢？因为服务端为了保证服务端性能，一个服务端在同一时刻只能被一个客户端请求连接，如果上一个客户端请求还未完成，则不允许新的客户端请求连接。当客户端请求接收到服务端响应后，将对应的客户端请求从 InFlightRequest 队列中移除。poll()阶段： 通过 ready()和 send()两阶段，完成了数据准备和可用节点检查。在上一篇中我们介绍到客户端是按照 Broker 分组，每组建立一个网络连接请求，每个网络连接请求管理多个网络连接通道，从而形成了一个连接同时与多个 Broker 进行网络数据传输。poll()方法采用了选择器(Selector)模式来处理这种网络模型，其底层是使用 Java 的 NIO 来实现的。

简单介绍下 Java NIO 的几个组件，想要深入了解的同学通过 Google 去了解。

SocketChannel： 客户端网络连接通道，在此通道上可进行数据的读写操作，比如将数据写入到通道中和将数据从通道中读取出来操作。Selector： 选择器，通道是需要注册到 Selector 选择器上的，同时在注册后会返回一个选择器建，Selector 会通过选择器键来监听读写事件。SelectorKey： 选择器键，通道注册到选择器上，同时返回了选择器键，从而使得选择器键和通道建立了关系。

以上三者之间的关系如下：

当有读写请求发生时，Selector 可以通过 SelectorKey 拿到对应的 SocketChannel，从而在 SocketChannel 上进行数据的读写请求。

Selector 选择器的实现原理

关于 Selector 选择器，我们从两个方面来介绍其背后发生了那些故事，分别是 建连过程和读写操作流程。

Selector 建连过程分析

Kafka建连过程

从上图可以看出，首先打开一个客户端连接 SocketChannel，然后对 Socket 设置一些参数，比如写入数据大小、接受数据大小、TCP 延迟等等参数。然后使用 SocketChannel 尝试建立连接。建连完成后将 SocketChannel 注册到 Selector 选择器上，并返回 SelectorKey。最后将 SocketChannel 包装成 KafkaChannel，并使用 SelectorKey 与 KafkaChannel 进行关联；为啥会出现 KafkaChannel 了呢？因为 Kafka 框架为了屏蔽 SocketChannel 内部的细节操作，所有就对 SocketChannel 进行了一层包装方便 Kafka 客户端操作。

附上源码供大家参考研究

Selector 选择器读写操作流程

读写流程图

从上图可以看出，以写操作为例，客户端轮询到写请求时，首先获取写请求对应的 SelectorKey，从而拿到对应的 KafkaChannel；然后将要发送的数据写入到 KafkaChannel 中；然后通过传输协议将数据交由底层的 SocketChannel；最后由 SocketChannel 将数据发送给 Broker，完成数据的发送请求。该过程中需要注意一个问题，Broker 在同一时间只能处理一个客户端请求，如果当前客户端请求还没被被处理完，下一个请求是不能被发送给服务端的。

总结

以上即为数据从客户端发送到服务端背后相关的网络操作故事；到此，关于生产者客户端的相关操作暂且分析到这里，关于客户端幂等性、消息重发等问题我们在后面专门用篇幅来讲解。下篇文章我们来分析一下消费者端消费消息背后的一些故事，敬请期待。

kafka系列：一文读懂消费者背后的那点"猫腻"

前言

经过前几篇文章的介绍，大致了解了生产者背后的运行原理。消息有生产就得有人去消费，今天我们就来介绍下消费端消费消息背后发生的那点事儿。

文章概览消费者与消费组的“父子关系”。Repartition 触发时机。消费者与 ZK 的关系。消费端工作流程。消费者的三种消费情况。消费者与消费组的“父子关系”

消费者消费组关系图

Kafka 消费端确保一个 Partition 在一个消费者组内只能被一个消费者消费。这句话该怎么理解呢？

在同一个消费者组内，一个 Partition 只能被一个消费者消费。在同一个消费者组内，所有消费者组合起来必定可以消费一个 Topic 下的所有 Partition。在同一个消费组内，一个消费者可以消费多个 Partition 的信息。在不同消费者组内，同一个分区可以被多个消费者消费。每个消费者组一定会完整消费一个 Topic 下的所有 Partition。消费组存在的意义

了解了消费者与消费组的关系后，有朋友会比较疑惑消费者组有啥实际存在的意义呢？或者说消费组的作用是什么？

作者对消费组的作用归结了如下两点。

在实际生产中，对于同一个 Topic，可能有 A、B、C 等 N 个消费方想要消费。比如一份用户点击日志，A 消费方想用来做一个用户近 N 天点击过哪些商品；B 消费方想用来做一个用户近 N 天点击过前 TopN 个相似的商品；C 消费方想用来做一个根据用户点击过的商品推荐相关周边的商品需求。对于多应用场景，就可以使用消费组来隔离不同的业务使用场景，从而达到一个 Topic 可以被多个消费组重复消费的目的。消费组与 Partition 的消费进度绑定。当有新的消费者加入或者有消费者从消费组退出时，会触发消费组的 Repartition 操作（后面会详细介绍 Repartition）；在 Repartition 前，Partition1 被消费组的消费者 A 进行消费，Repartition 后，Partition1 消费组的消费者 B 进行消费，为了避免消息被重复消费，需要从消费组记录的 Partition 消费进度读取当前消费到的位置（即 OffSet 位置），然后在继续消费，从而达到消费者的平滑迁移，同时也提高了系统的可用性。Repartition 触发时机

使用过 Kafka 消费者客户端的同学肯定知道，消费者组内偶尔会触发 Repartition 操作，所谓 Repartition 即 Partition 在某些情况下重新被分配给参与消费的消费者。基本可以分为如下几种情况。

消费组内某消费者宕机，触发 Repartition 操作，如下图所示。

消费者宕机情况

消费组内新增消费者，触发 Repartition 操作，如下图所示。一般这种情况是为了提高消费端的消费能力，从而加快消费进度。

新增消费者情况

Topic 下的 Partition 增多，触发 Repartition 操作，如下图所示。一般这种调整 Partition 个数的情况也是为了提高消费端消费速度的，因为当消费者个数大于等于 Partition 个数时，在增加消费者个数是没有用的（原因是：在一个消费组内，消费者:Partition = 1:N，当 N 小于 1 时，相当于消费者过剩了），所以一方面增加 Partition 个数同时增加消费者个数可以提高消费端的消费速度。

新增Partition个数情况

消费者与 ZK 的关系

众所周知，ZK 不仅保存了消费者消费 partition 的进度，同时也保存了消费组的成员列表、partition 的所有者。消费者想要消费 Partition，需要从 ZK 中获取该消费者对应的分区信息及当前分区对应的消费进度，即 OffSert 信息。那么 Partition 应该由那个消费者进行消费，决定因素有哪些呢？从之前的图中不难得出，两个重要因素分别是：消费组中存活的消费者列表和 Topic 对应的 Partition 列表。通过这两个因素结合 Partition 分配算法，即可得出消费者与 Partition 的对应关系，然后将信息存储到 ZK 中。Kafka 有高级 API 和低级 API，如果不需要操作 OffSet 偏移量的提交，可通过高级 API 直接使用，从而降低使用者的难度。对于一些比较特殊的使用场景，比如想要消费特定 Partition 的信息，Kafka 也提供了低级 API 可进行手动操作。

消费端工作流程

在介绍消费端工作流程前，先来熟悉一下用到的一些组件。

KakfaConsumer：消费端，用于启动消费者进程来消费消息。ConsumerConfig：消费端配置管理，用于给消费端配置相关参数，比如指定 Kafka 集群，设置自动提交和自动提交时间间隔等等参数，都由其来管理。ConsumerConnector：消费者连接器，通过消费者连接器可以获得 Kafka 消息流，然后通过消息流就能获得消息从而使得客户端开始消费消息。

以上三者之间的关系可以概括为：消费端使用消费者配置管理创建出了消费者连接器，通过消费者连接器创建队列（这个队列的作用也是为了缓存数据），其中队列中的消息由专门的拉取线程从服务端拉取然后写入，最后由消费者客户端轮询队列中的消息进行消费。具体操作流程如下图所示。

消费端工作流程

我们在从消费者与 ZK 的角度来看看其工作流程是什么样的？

消费端与ZK之间的工作流程

从上图可以看出，首先拉取线程每拉取一次消息，同步更新一次拉取状态，其作用是为了下一次拉取消息时能够拉取到最新产生的消息；拉取线程将拉取到的消息写入到队列中等待消费消费线程去真正读取处理。消费线程以轮询的方式持续读取队列中的消息，只要发现队列中有消息就开始消费，消费完消息后更新消费进度，此处需要注意的是，消费线程不是每次都和 ZK 同步消费进度，而是将消费进度暂时写入本地。这样做的目的是为了减少消费者与 ZK 的频繁同步消息，从而降低 ZK 的压力。

消费者的三种消费情况

消费者从服务端的 Partition 上拉取到消息，消费消息有三种情况，分别如下：

至少一次。即一条消息至少被消费一次，消息不可能丢失，但是可能会被重复消费。至多一次。即一条消息最多可以被消费一次，消息不可能被重复消费，但是消息有可能丢失。正好一次。即一条消息正好被消费一次，消息不可能丢失也不可能被重复消费。1.至少一次

消费者读取消息，先处理消息，在保存消费进度。消费者拉取到消息，先消费消息，然后在保存偏移量，当消费者消费消息后还没来得及保存偏移量，则会造成消息被重复消费。如下图所示：

先消费后保存消费进度

2.至多一次

消费者读取消息，先保存消费进度，在处理消息。消费者拉取到消息，先保存了偏移量，当保存了偏移量后还没消费完消息，消费者挂了，则会造成未消费的消息丢失。如下图所示：

先保存消费进度后消费消息

3.正好一次

正好消费一次的办法可以通过将消费者的消费进度和消息处理结果保存在一起。只要能保证两个操作是一个原子操作，就能达到正好消费一次的目的。通常可以将两个操作保存在一起，比如 HDFS 中。正好消费一次流程如下图所示。

正好消费一次

总结

本文讲解了消费组与消费者之间的关系，及 Repartition 的触发时机，然后讲述了消费端的基本工作流程，最后提出了一条消息被重复消费的几种情况。下篇文章我们来讲讲消息在服务端是怎么存储的，敬请期待。（本文来自公众号：z小赵）

感谢大家支持， 多多转发关注不迷路~~~~