kafka-consumer-offset位移-创新互联

在霍林郭勒等地区，都构建了全面的区域性战略布局，加强发展的系统性、市场前瞻性、产品创新能力，以专注、极致的服务理念，为客户提供成都网站设计、成都网站建设、外贸网站建设网站设计制作定制网站,公司网站建设,企业网站建设,高端网站设计,网络营销推广,外贸网站制作,霍林郭勒网站建设费用合理。

1 offset的默认维护位置

1.1 消费offset案例

2 自动提交offset

3 手动提交offset

3.1 原理

3.2 代码示例

3.2.1 同步提交

3.2.2 异步提交(生产常用)

4 指定offset消费

5 指定时间消费

6 漏消费和重复消费分析

6.1 重复消费

6.2 漏消费

6.3 消费者事务

7 数据积压

1 offset的默认维护位置

_consumer_offsets主题里面采用key和 value的方式存储数据。key是 group.id+topic+分区号，value 就是当前offset的值。每隔一段时间，kafka 内部会对这个topic进行compact(压缩)，也就是每个group.id+topic+分区号就保留最新数据。

Kafka0.9版本之前，consumer黑认将offset保存在Zookeeper中。0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为_consumer_offsets。

将offset信息存储在zk中的不足：如果将offset信息存储在zk中，那么所有的consumer都会访问zk，会消耗大量的网络资源，消费速度慢。

1.1 消费offset案例

思想:_consumer_offsets为Kafka中的 topic，那就可以通过消费者进行消费。
在配置文件 config/consumer.properties中添加配置exclude.internal.topics = false，默认是 true，表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为false。修改以后执行分发命令：xsync consumer.properties。

采用命令行方式，创建一个新的topic。

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2

启动生产者往atguigu生产数据。

[atguigu@hadoop102 kafka] $ bin/kafka-console-producer.sh --topic atguigu --bootstrap-server hadoop102:9092

启动消费者消费atguigu数据。
```
[atguigu@hadoop104 kafka]$ bin/kafka-console-consumer.sh bootstrap-server hadoop102:9092--topic atguigu --group test
```
注意:指定消费者组名称，更好观察数据存储位置(key是 group.id+topic+分区号)。

查看消费者消费主题_consumer_offsets。

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --topic _consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

2 自动提交offset

为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能。自动提交offset的相关参数:

enable.auto.commit:是否开启自动提交offset功能，默认是true
auto.commit.interval.ms:自动提交offset的时间间隔，默认是5s

消费者配置代码：

//配置是否是自动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
//提交时间间隔，单位是ms
properties.put(ConsumerConfig.AUTO_COMNIT_INTERVAL_NS_CONFI6,1000);

3 手动提交offset 3.1 原理

虽然自动提交offset十分简单便利，但由于其是基于时间提交的，开发人员难以把握offset提交的时机。因此Kafka还提供了手动提交offset的API。
手动提交offset的方法有两种:分别是commitSync（同步提交）和commitAsync(异步提交）。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交;不同点是，同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试(由不可控因素导致，也会出现提交失败）﹔而异步提交则没有失败重试机制，故有可能提交失败。

commitSync（同步提交)﹔必须等待offset提交完毕，再去消费下一批数据。
commitAsync(异步提交）:发送完提交offset请求后，就开始消费下一批数据了

3.2 代码示例 3.2.1 同步提交

//手动提交属性配置
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ,false);
//消费代码逻辑
XXX
XXX
XXX
//手动提交代码(处理完数据以后，这里为了方便，只展示关键代码)
//手动提交offset
kafkaConsumer.commitsync();

3.2.2 异步提交(生产常用)

//手动提交属性配置
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ,false);
//消费代码逻辑
XXX
XXX
XXX
//手动提交代码(处理完数据以后，这里为了方便，只展示关键代码)
//手动提交offset
kafkaConsumer.commitAsync();

4 指定offset消费

auto.offset.reset = earliest | latest | none 默认是latest。
当Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除)，该怎么办?

earliest:自动将偏移量重置为最早的偏移量，--from-beginning。
latest（默认值):自动将偏移量重置为最新偏移量。
none:如果未找到消费者组的先前偏移量，则向消费者抛出异常。

任意指定offset位移开始消费。

//1创建消费者
KafkaConsumerkafkaConsumer = new KafkaConsumer<>(properties);
// 2订阅主题
ArrayListtopics = new ArrayList<>(;topics.add( "first");
kafkaConsumer.subscribe(topics);

//指定位置进行消费
setassignment = kafkaConsumer.assignment();//获取所有分区信息
//保证分区分配方案已经制定完毕，因为由于leader消费者制定分配方案会消耗一定时间，有可能此时获取不到分区信息,所以加一层分区空间判断
while (assignment.size() == 0){
    //促使获取的分区数量不为0
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}

//遍历所有分区，指定消费的offset
for (TopicPartition topicPartition : assignment) {
    kafkaConsumer.seek(topicPartition, 100);
}

// 3消费数据
while (true){

5 指定时间消费

需求:在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。例如要求按照时间消费前一天的数据，怎么处理?

//1创建消费者
KafkaConsumerkafkaConsumer = new KafkaConsumer<>(properties);
// 2订阅主题
ArrayListtopics = new ArrayList<>(;topics.add( "first");
kafkaConsumer.subscribe(topics);

//指定位置进行消费
setassignment = kafkaConsumer.assignment();//获取所有分区信息
//保证分区分配方案已经制定完毕，因为由于leader消费者制定分配方案会消耗一定时间，有可能此时获取不到分区信息,所以加一层分区空间判断
while (assignment.size() == 0){
    //促使获取的分区数量不为0
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}
//希望把时间转换为对应的offset
HashMaptopicPartitionLongHashMap = new HashMap<>();
//封装对应集合
for (TopicPartition topicPartition : assignment) {
    //希望获取当前系统时间一天前的数据。
    topicPartitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
}
NaptopioPartitionffsetAndrtimestampMep = karfiaConsumer.offsetsForTines(topicPartitionL ongHashiap);


//遍历所有分区，指定消费的offset
//指定消费的offset
for (TopicPartition topicPartition : assignment) {
    OffsetAndTimestamp offsetAndTimestamp = topicPartition0ffsetAndTimestampHap.get(topicPartition);
    kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
}

// 3消费数据
while (true){

6 漏消费和重复消费分析 6.1 重复消费

场景1:重复消费。自动提交offset引起。