kafka使用中的若干知识点_topic的end和offset

最近在使用flink的时候，因为对kafka的基础知识了解不深入，导致了一些问题，所以趁着这个机会看了一下kafka消费端的代码，整理处理。

版本

• 0.8.2版本

  • producer 都以异步批量的方式发送消息到 broker 节点。
  • consumer 的消费偏移位置 offset 由原来的保存在 zookeeper 改为保存在 kafka 本身。

• 0.9版本

  • 新的 consumer api ag: 心跳线程与用户线程进行分离

• 0.11 版本

  • 支持 Exactly-Once 语义即精准一次语义，主要是实现了Producer端的消息幂等性，以及事务特性。

• 1.0.0版本

  • 增强 stream api
  • 引入了线程协议的变化。通过遵循下面建议的滚动升级计划，您可以保证在升级期间不会出现停机。

对于版本的问题，一般还是根据客户端和服务端保持一致的版本，如果实在有问题，尽量保证高版本客户端兼容低版本的服务端。
里面几个比较重要的版本就是0.9版本，几乎各大厂商都是0.9版本以上，使用了新的消费客户端。
对于flink而言，如果要保证Exactly-Once, 尽量是使用0.11版本以上，如果只是客户端的一致性，就不用太过于考虑版本了。

消费者消费topic三种方式

KafkaConsumer.subscribe()

• 指定对应的topic.
• consumer自动分配partition,内部算法保障topic-partition以最优的方式均匀分配给group下不同的consumer.
• consumer自动上下线/group的扩缩容可以自动通过reblance机制均衡消费，不过再均衡期间，消费者是无法拉取消息的
• offset的管理通过group management保证。
• 高级api
• offset支持自动/手动提交
  在使用subscribe的时候，需要关注的点包括配置，自动/手动提交，rebalance监听器。

绝大多数的应用场景，用户不需要关心consumer的上下线，自动重连，消费迁移等。

KaflaConsumer.assign();

• 指定topic对应的分区。
• 用户直接手动消费具体分区,具体offset. 这些在启动之前必须要初始化完成。
• 没有kafka group management机制，offset的管理需要消费者自行管理。
• 没有消费者数量变化的时候reblance行为的发生，consumer的自动上下线/group的扩缩容全部需要重启应用来完成。

为了避免reblance造成的数据重复读取或者丢失的情况，用户手动消费具体的分区，负载均衡全部需要用户自己控制。

SimpleConsumer.fetch()

• 不需要group来管理offset, offset的管理全部依赖第三方。可以自定义每次读取的offsetBegin, offsetEnd. seek方法提供了这个功能。
  

对于有追踪之前的数据，多次在不同的位点回溯消费等批量获取数据。而且大多数场景下需要exactly-once保证消息的不丢失和唯一性。

poll()方法

subscribe(), assign()在拉取数据的时候使用的是相同的方法体，关于客户端的部分参数是相同的逻辑。关于poll源码的具体逻辑可以参考这篇博文：添加链接描述

三种提交方式

自动提交

enable.auto.commit=true, 消费者会自动把从poll()方法接收到的最大偏移量提交上去。提交的时间间隔auto.commit.interval.ms控制，默认5s。自动提交是在轮询中进行。

手动提交

auto.commit.offset=false, 自行决定何时提交偏移量，分为同步提交commitSync()和commitAsync(); commitSync模式在成功提交或者碰到无法恢复的错误之前，会一直重试。而commitAsync不会, 只会试用future等待超时时间过去，之所以不会被重试，对于异步提交而言，是没有顺序的，如果同时存在多个异步提交，进行重试可能会导致位移覆盖，就会导致消息重复消费。

同步与异步组合

在触发关闭消费者或者均衡前的最后一次提交，如果异步没有提交成功就需要用同步的方式进行轮询尝试提交。

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(100);
        for (ConsumerRecord<String, String> record : records) {
            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }
        consumer.commitAsync();
    }
} catch (Exception e) {
    System.out.println("commit failed");
} finally {
    try {
        consumer.commitSync();
    } finally {
        consumer.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

提交特定的偏移量

上面讲的commitSync(), commitAsync()都是提交最后的偏移量，当然也可以通过提交特定偏移量的offset, 这个经常用在低级api里面。

void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback);
void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout);
1
2

配置说明

enable.auto.commit

在subscribe模式下，可以设置true/false, 在assign模式下只能为false.

• commitSync 同步提交，会根据consumer.position()得到position的值进行提交，该方法使用顺序同步阻塞消费者线程直到offset提交完成(在规定的时间内轮询请求)，并且返回消息。
• commitAsync 异步提交，使用的是future模式+回调，不会阻塞消费者线程，可能在提交offset的结果还未返回，消费者线程就可以开始继续拉去数据，异步提交可以让消费者的消费能力得到增强。

如果不考虑数据丢失的情况下，可以使用自定提交的模式。
如果对于数据做了幂等的操作，数据可重复读对结果并没有影响，一般使用enable.auto.commit.
如果要求数据只处理一次，只能用手工提交，同步+异步的模式以及两阶段提交的才能够保证数据提交+offset提交是原子操作。
1
2
3

timeout.ms

KafkaComsumer.poll()方法, 等待拉取的时常。如果等于0，不管buffer中有没有可用的数据，如果有直接拿到返回，没有返回empty. 如果大于0，并且buffer中没有可用的数据，会一直轮询直到对应的时间超时为止。

这个配置有一个妙用，可以帮助我们确定数据是否在正常消费，比如对于一个均匀的消息报，如果超过timeout.ms还没有收到消息，是否可以认为上游消息报发送不及时或者挂掉。合理设置timeout.ms的大小，对于我们排查问题有一定的帮助。
1

max.poll.record / max.partition.fetch.bytes

max.partition.fetch.bytes 表示consumer每次发起fetch请求读取到的分配的每个分区的数据最大值。并且读取到的数据放入到本地缓存。
max.poll.record 表示每次poll的时候从缓存中拉取到所分配所有分区的数据最大记录数。

这就会有一个很有意思的问题，比如每次fetch的数据是100条，每次poll的数据是10条，那么需要执行10次poll才能把一次fetch的数据读完。如果每次fetch的数据是10条，max.poll.record=50，那么每次poll的数据就是10. 所以一般而言 max.partition.fetch.bytes / per recore size >= max.poll.record 比较好，才会让线程发挥最大的价值。
1

heartbeat.interval.ms / session.timeout.ms

heartbeat.interval.ms 心跳间隔。心跳是在consumer与coordinator之间进行的，心跳用来保持consumer的会话，并且在有consumer加入或者离开group时帮助进行rebalance。
session.timeout.ms, Consumer session 过期时间， 本身是个逻辑值，一般是需要和heartbeat.interval.ms配合使用。
heartbeat线程 每隔heartbeat.interval.ms向coordinator发送一个心跳包，证明自己还活着。只要 heartbeat线程 在 session.timeout.ms 时间内 向 coordinator发送过心跳包，那么 group coordinator就认为当前的kafka consumer是活着的

通常设置heartbeat.interval.ms <  session.timeout.ms / 3。以防止突然网络卡顿的现象。
1

max.poll.interval.ms

max.poll.interval.ms subscribe模式下，两次Poll之间最大的延迟时间，也是rebalance的timeout，max.poll.interval.ms = record的处理时间 + poll的时间 + commit的时间
如果两次poll时间超过当前时间，就会认为当前consumer挂了，就会导致重新reblance。

必须尽力保证一次poll的消息很快的完成，无论我们的业务代码做了什么操作(耗时的，短时的)。
如果我们业务代码是耗时的，需要适当的增大max.poll.interval.ms， 并且减少max.poll.record。对于耗时的业务代码，也可以使用异步线程去调用，以免超时造成reblance.

1
2
3

关于reblance

Rebalance 本质上是一种协议，规定了一个 Consumer Group 下的所有 consumer 如何达成一致，来分配订阅 Topic 的每个分区。例如：某 Group 下有 20 个 consumer 实例，它订阅了一个具有 100 个 partition 的 Topic 。正常情况下，kafka 会为每个 Consumer 平均的分配 5 个分区。这个分配的过程就是 Rebalance。
Rebalance 发生时，Group 下所有 consumer 实例都会协调在一起共同参与，kafka 能够保证尽量达到最公平的分配。但是 Rebalance 过程对 consumer group 会造成比较严重的影响。在 Rebalance 的过程中 consumer group 下的所有消费者实例都会停止工作，等待 Rebalance 过程完成。

reblance触发的条件

• 组成员个数发生变化。例如有新的 consumer 实例加入该消费组或者离开组
• 订阅的 Topic 个数发生变化。
• 订阅 Topic 的分区数发生变化

造成reblance的一些因素

• session.timeout 设置的太短，比如网络出现问题，心跳阻塞，超过session.timeout， group协调器就会任务consumer已经挂了，就会导致reblance. 所以以heartbeat.interval.ms=3000ms为例，实际场景中建议session.timeout设置为30000ms左右最好。不过这个只适合心跳突然阻塞，如果网络长时间阻塞必然会rebalance.
• 消费能力不足，如果某个poll拉取的数据消耗的时间超过max.poll.interval.ms,也会导致rebalance, 所以对于处理耗时比较长的记录，可以增大max.poll.interval.ms，并且减少max.poll.records。
• 服务端网络IO负载较高，commit 超时，这个时候就会产生消费能力不足的假象，导致rebalance.

一致性的解决方案

at most once

提交的偏移量大于客户端处理的最后一个消息的偏移量,在发生消费关闭/reblance时候，那么处于两个偏移量之间的消息将会丢失

出现at most once的一些场景

• enable.auto.commit=true，并且auto.commit.interval.ms为一个较小的时间间隔, 如果消费失败/消费关闭/重平衡，但是已经提交offset
• enable.auto.commit=false, 同步/异步提交，如果消费记录同步/异步提交之后, 提交成功，但是消费记录失败。

at least once

提交的偏移量小于消费者实际处理的最后一个消息的偏移量，处于两个偏移量之间的消息会被重复处理

• enable.auto.commit=true, 并且auto.commit.interval.ms为一个较大的时间间隔，消费已经完成，但是offset还没有来得及提交。
• enable.auto.commit=false, 同步/异步提交，如果消费记录在同步/异步提交之前, 提交成功，但是消费记录成功，但是提交失败。

exactly once

如果要实现这种方式，必须自己控制消息的offset，自己记录一下当前的offset，对消息的处理和offset的移动必须保持在同一个事务中，例如在同一个事务中，把消息处理的结果存到mysql数据库同时更新此时的消息的偏移，而且在很多业务中需要考虑记录的顺序执行。因为kafka本身是分区顺序保证，所以需要从业务层面去考虑这个问题

• enable.auto.commit=false, 保存ConsumerRecord中的offset到数据库或者文件，assign模式消费。

使用经验

在我的经验中，kafka消费的三种模式都有用到，

• 大部分的场景，一般使用订阅的模式，以at-least-once的形式消费记录。
• 在实时数据中，对于可以忽略部分数据，以订阅模式at-most-once的形式消费。
• flink实时数据消费kafka，通常使用kafka-connector，以assign指定分区的形式进行消费，可以做到at-least-once, exactly-once等。
• 对于批处理消费kafka，很多时候需要考虑回刷到上一次的消费位点，这种一般会直接获取topic的offset-begin, offset-end, 在第三方保存位点。批量进行消费。

每种模式都有自己的使用场景，在使用的时候我们需要考虑的点

• 是否需要管理group以及offset.
• 如果进行提交，手工，还是自动
• 客户端挂掉的场景下，如何不丢失数据，如何保证幂等。

引用

https://www.cnblogs.com/yoke/p/11405397.html
https://cloud.tencent.com/developer/article/1676951
https://blog.csdn.net/pml18710973036/article/details/87207071
https://blog.csdn.net/weixin_39662432/article/details/111665050
https://blog.csdn.net/weixin_43956062/article/details/106781317?spm=1001.2014.3001.5501
http://kafka.apache.org/0110/documentation.html#consumerconfigs