从未如此简单：10分钟带你逆袭Kafka！(6)

Offset Commit：Consumer 从 Broker 中取一批消息写入 Buffer 进行消费，在规定的时间内消费完消息后，会自动将其消费消息的 Offset 提交给 Broker，以记录下哪些消息是消费过的。当然，若在时限内没有消费完毕，其是不会提交 Offset 的。

Kafka的工作原理和过程

①消息写入算法

消息发送者将消息发送给 Broker, 并形成最终的可供消费者消费的 log，是已给比较复杂的过程：

Producer 先从 Zookeeper 中找到该 Partition 的 Leader。

Producer将消息发送给该 Leader。

Leader 将消息接入本地的 log，并通知 ISR 的 Followers。

ISR 中的 Followers 从 Leader 中 Pull 消息, 写入本地 log 后向 Leader 发送 Ack。

Leader 收到所有 ISR 中的 Followers 的 Ack 后，增加 HW 并向 Producer 发送 Ack，表示消息写入成功。

②消息路由策略

在通过 API 方式发布消息时，生产者是以 Record 为消息进行发布的。

Record 中包含 Key 与 Value，Value 才是我们真正的消息本身，而 Key 用于路由消息所要存放的 Partition。

消息要写入到哪个 Partition 并不是随机的，而是有路由策略的：

若指定了 Partition，则直接写入到指定的 Partition。

若未指定 Partition 但指定了 Key，则通过对 Key 的 Hash 值与 Partition 数量取模，该取模。

结果就是要选出的 Partition 索引。

若 Partition 和 Key 都未指定，则使用轮询算法选出一个 Partition。

③HW 截断机制

如果 Partition Leader 接收到了新的消息， ISR 中其它 Follower 正在同步过程中，还未同步完毕时 leader 宕机。

此时就需要选举出新的 Leader。若没有 HW 截断机制，将会导致 Partition 中 Leader 与 Follower 数据的不一致。

当原 Leader 宕机后又恢复时，将其 LEO 回退到其宕机时的 HW，然后再与新的 Leader 进行数据同步，这样就可以保证老 Leader 与新 Leader 中数据一致了，这种机制称为 HW 截断机制。

④消息发送的可靠性

生产者向 Kafka 发送消息时，可以选择需要的可靠性级别。通过 request.required.acks 参数的值进行设置。

0 值：异步发送。生产者向 Kafka 发送消息而不需要 Kafka 反馈成功 Ack。该方式效率最高，但可靠性最低。

其可能会存在消息丢失的情况：

在传输过程中会出现消息丢失。

在 Broker 内部会出现消息丢失。

会出现写入到 Kafka 中的消息的顺序与生产顺序不一致的情况。

1 值：同步发送。生产者发送消息给 Kafka，Broker 的 Partition Leader 在收到消息后马上发送成功 Ack(无需等等 ISR 中的 Follower 同步)。

生产者收到后知道消息发送成功，然后会再发送消息。如果一直未收到 Kafka 的 Ack，则生产者会认为消息发送失败，会重发消息。

该方式对于 Producer 来说，若没有收到 Ack，一定可以确认消息发送失败了，然后可以重发。

但是，即使收到了 ACK，也不能保证消息一定就发送成功了。故，这种情况，也可能会发生消息丢失的情况。

-1 值：同步发送。生产者发送消息给 Kafka，Kafka 收到消息后要等到 ISR 列表中的所有副本都 同步消息完成后，才向生产者发送成功 Ack。

如果一直未收到 Kafka 的 Ack，则认为消息发送 失败，会自动重发消息。该方式会出现消息重复接收的情况。

⑤消费者消费过程解析

生产者将消息发送到 Topitc 中，消费者即可对其进行消费，其消费过程如下：

Consumer 向 Broker 提交连接请求，其所连接上的 Broker 都会向其发送Broker Controller 的通信 URL，即配置文件中的 Listeners 地址。

当 Consumer 指定了要消费的 Topic 后，会向 Broker Controller 发送消费请求。

Broker Controller 会为 Consumer 分配一个或几个 Partition Leader，并将该 Partition 的当前 Offset 发送给 Consumer。

Consumer 会按照 Broker Controller 分配的 Partition 对其中的消息进行消费。

当 Consumer 消费完该条消息后，Consumer 会向 Broker 发送一个消息已经被消费反馈，即该消息的 Offset。

在 Broker 接收到 Consumer 的 Offset 后，会更新相应的 __consumer_offset 中。

以上过程会一直重复，知道消费者停止请求消费。

Consumer 可以重置 Offset，从而可以灵活消费存储在 Broker 上的消息。

⑥Partition Leader 选举范围

当 Leader 宕机后，Broker Controller 会从 ISR 中挑选一个 Follower 成为新的 Leader。

如果 ISR 中没有其他副本怎么办?可以通过 unclean.leader.election.enable 的值来设置 Leader 选举范围。

False：必须等到 ISR 列表中所有的副本都活过来才进行新的选举。该策略可靠性有保证，但可用性低。

True：在 ISR 列表中没有副本的情况下，可以选择任意一个没有宕机的主机作为新的 Leader，该策略可用性高，但可靠性没有保证。

⑦重复消费问题的解决方案

同一个 Consumer 重复消费：当 Consumer 由于消费能力低而引发了消费超时，则可能会形成重复消费。

在某数据刚好消费完毕，但是正准备提交 Offset 时候，消费时间超时，则 Broker 认为这条消息未消费成功。这时就会产生重复消费问题。其解决方案：延长 Offset 提交时间。

不同的 Consumer 重复消费：当 Consumer 消费了消息，但还没有提交 Offset 时宕机，则这些已经被消费过的消息会被重复消费。其解决方案：将自动提交改为手动提交。

⑧从架构设计上解决 Kafka 重复消费的问题

我们在设计程序的时候，比如考虑到网络故障等一些异常的情况，我们都会设置消息的重试次数，可能还有其他可能出现消息重复，那我们应该如何解决呢?下面提供三个方案：

方案一：保存并查询

给每个消息都设置一个独一无二的 uuid，所有的消息，我们都要存一个 uuid。

我们在消费消息的时候，首先去持久化系统中查询一下看这个看是否以前消费过，如没有消费过，在进行消费，如果已经消费过，丢弃就好了。

下图表明了这种方案：

（编辑：ASP站长网）