消费者实现逻辑-kafka知识体系(四）

发布时间：2025-11-05 12:14:47 来源：云智核 作者：人工智能

上篇文章分享kafka broker 的消费现逻实现原理、数据的实知识存储结构和消息持久化相关的东西，那消息存储完了之后，体系怎么被消费端消费呢，消费现逻本文来聊一聊Kafka 消费端的实知识那些事儿。

1)拉取机制

Kafka生产端是体系推的机制即Push，消费端是消费现逻拉的机制即Pull。

2)Pull的实知识优缺点

优点是消费端可以自己控制消息的读取速度和数量;

缺点是不知道服务端有没有数据，所以要一直pull或隔一定时间pull，体系可能要pull多次并等待。消费现逻

3)消息投递语义：

Kafka默认保证at-least-once delivery，实知识容许用户实现at-most-once语义，体系exactly-once的消费现逻实现取决于目的存储系统。

4)分区分配策略

RangeAssignor：按照分区范围分配，实知识当前默认策略;

RoundRobinAssignor：轮询的体系方式分配;

StickyAssignor：Kafka 0.11版本引入，根据更多指标比如负载，尽可能均匀。

这些前面的文章中也有提到。b2b信息网

消费者组

Consumer Group 是 Kafka 提供的可扩展且具有容错性的消费者机制。Kafka 仅仅使用 Consumer Group 这一种机制，却同时实现了传统消息引擎系统的两大模型：消息队列模型、发布 / 订阅模型。

理想情况下，Consumer 实例的数量应该等于该 Group 订阅主题的分区总数。

【消费者和消费组】

Kafka消费者是消费组的一部分，当多个消费者形成一个消费组来消费主题时，每个消费者会收到不同分区的消息。假设有一个T1主题，该主题有4个分区;同时我们有一个消费组G1，这个消费组只有一个消费者C1。那么消费者C1将会收到这4个分区的消息，如下所示：

Kafka一个很重要的特性就是，只需写入一次消息，可以支持任意多的应用读取这个消息。换句话说，每个应用都可以读到全量的消息。为了使得每个应用都能读到全量消息，企商汇应用需要有不同的消费组。对于上面的例子，假如我们新增了一个新的消费组G2，而这个消费组有两个消费者，那么会是这样的：

这里值得我们注意的是：

但是每个 消费者组 消费的数据是 互不干扰 的，也就是说，每个 消费组 消费的都是 完整的数据 。

而 不能拆给多个消费者 消费，也就是说如果你某个 消费者组内的消费者数 比 该 Topic 的分区数还多，那么多余的消费者是不起作用的免费信息发布网

消费者分区分配的过程

那么我们现在就来看看分配过程是怎么样的。

1.确定 群组协调器

每当我们创建一个消费组，kafka 会为我们分配一个 broker 作为该消费组的 coordinator(协调器)

2.注册消费者 并选出 leader consumer

当我们的有了 coordinator 之后，消费者将会开始往该 coordinator上进行注册，第一个注册的 消费者将成为该消费组的 leader，后续的 作为 follower

3.当 leader 选出来后，

他会从coordinator那里实时获取分区 和 consumer 信息，并根据分区策略给每个consumer 分配 分区，并将分配结果告诉 coordinator。

4.follower 消费者将从 coordinator 那里获取到自己相关的分区信息进行消费，

对于所有的 follower 消费者而言，他们只知道自己消费的分区，并不知道其他消费者的存在。

5.至此，消费者都知道自己的消费的分区，

分区过程结束，当发生 分区再均衡 的时候，leader 将会重复分配过程

具体的流程图可以翻阅前面的文章。

关于位移

【位移 offset】

【位移的保存】

其实Consumer 端应用程序在提交位移时，其实是向 Coordinator 所在的 Broker 提交位移。同样地，当 Consumer 应用启动时，也是向 Coordinator 所在的 Broker 发送各种请求，然后由 Coordinator 负责执行消费者组的注册、成员管理记录等元数据管理操作。

老版本的 Consumer Group 把位移保存在 ZooKeeper 中，新版本的 Consumer Group 中，Kafka 社区重新设计了 Consumer Group 的位移管理方式，采用了将位移保存在 Kafka内部主题的方法,也就是__consumer_offsets，俗称位移主题。至于为什么放弃kafka 保存位移请看我前面的文章《基础概念、架构和新版的升级Kafka知识体系1》。

【位移主题的数据格式】

key

value

【位移的提交】

1. 自动提交

最简单的提交方式是让消费者自动提交偏移量，如果 enable.auto.commit 被设为 true，那么每过 5s，消费者会自动把从 poll() 方法接收到的最大偏移量提交上去。

可能造成的问题：数据重复读

假设我们仍然使用默认的 5s 提交时间间隔，在最近一次提交之后的 3s 发生了再均衡，再均衡之后，消费者从最后一次提交的偏移量位置开始读取消息。这个时候偏移量已经落后了 3s，所以在这 3s内到达的消息会被重复处理。可以通过修改提交时间间隔来更频繁地提交偏移量，减小可能出现重复消息的时间窗，不过这种情况是无法完全避免的。

2. 手动提交

2.1 同步提交

同步存在的问题

2.2 异步提交

手动提交有一个不足之处，在 broker 对提交请求作出回应之前，应用程序会一直阻塞，这样会限制应用程序的吞吐量。我们可以通过降低提交频率来提升吞吐量，但如果发生了再均衡，会增加重复消息的数量。

这时可以使用异步提交，只管发送提交请求，无需等待 broker 的响应。它之所以不进行重试，是因为在它收到服务器响应的时候，可能有一个更大的偏移量已经提交成功。

假设我们发出一个请求用于提交偏移量2000，这个时候发生了短暂的通信问题，服务器收不到请求，自然也不会作出任何响应。与此同时，我们处理了另外一批消息，并成功提交了偏移量3000。如果commitAsync()重新尝试提交偏移量2000，它有可能在偏移量3000之后提交成功。这个时候如果发生再均衡，就会出现重复消息。

异步存在的问题

综合异步和同步来提交

同时使用了 commitSync() 和 commitAsync()。对于常规性、阶段性的手动提交，我们调用 commitAsync() 避免程序阻塞，而在 Consumer 要关闭前，我们调用 commitSync() 方法执行同步阻塞式的位移提交，以确保 Consumer 关闭前能够保存正确的位移数据。

关于再均衡Rebalance

分区的所有权从一个消费者转移到另一个消费者，这样的行为被称为再均衡(Rebalance)。再均衡非常重要，为消费者组带来了高可用性和伸缩性，可以放心的增加或移除消费者。以下是触发再均衡的三种行为：

分区再均衡 期间该 Topic 是不可用的，所以Rebalance 实在是太慢了!!!

这里再补充一下生产环境中因为不正确的配置引起的不需要的分区再均衡。

正常集群变动不再考虑范围内：

1.防止 因为未能及时发送心跳，导致Consumer 超时被踢出消费者组。

这里可以设置 session.timeout.ms超时时间 和 heartbeat.interval.ms 心跳间隔一般可以把 超时时间设置为 心跳间隔的 3倍。

2.Consumer消费时间过长导致的。

Consumer端如果无法在规定时间内消费完 poll 来的消息，那么就认为该消费者有问题，从而该消费者会自主离组，所以我们可以设置 max.poll.interval.ms比处理时间略长。

3.从第二点我们还可能引申一点就是，如果集群经常发生 分区在均衡，

那么你可能需要去观察下消费者执行任务的耗时，特别注意观察下 GC 的占用时间。

往往线上出问题也是因为配置不合理导致的。

相关文章