KafkaConsuner 指定开始消费的位置

这篇具有很好参考价值的文章主要介绍了KafkaConsuner 指定开始消费的位置。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

KafkaConsumer 指定消费位置的基础

一个 Topic 对应着磁盘上的几个重要的文件:

  • .log:数据文件,存储了该 topic 中的所有消息。
  • .index:索引文件,对数据文件中的消息进行索引。关键就是可以按照 offset 来索引
  • .timeindex:时间索引文件,类似于索引文件,但按照消息的时间戳进行索引。它保存着 timestamp-offset 的关系,所以可以使用时间戳找到对应的 offset ,然后再找到对应的 offset 位置,这样就能找到对应的文件开始的位置了。
  • leader-epoch-checkpoint:该文件的职责是存储leader对于topic的进度,以便在leader重启时可以重新定位到上一次成功提交的offset位置。

在我们探讨的功能中,index和timeindex两个文件扮演着关键角色,它们与消费者开始消费的位置密切相关。这是我们今天要讲解的核心基础,没有这两个文件,我们无法进行后续的操作。

具体的做法

方法名 用法 业务意义
subscribe() consumer.subscribe(Arrays.asList(“topic1”, “topic2”)); 订阅一个或多个主题,按照策略指定当前消费者实例消费那个分区的数据
assign() TopicPartition partition0 = new TopicPartition(“test”, 0);
consumer.assign(Arrays.asList(partition0));
指定分配的TopicPartition,手工指定当前消费者实例消费那个分区的数据
assignment() Set partitions = consumer.assignment(); 获取当前已分配的TopicPartition列表
beginningOffsets() Map<TopicPartition, Long> beginningOffsets = consumer.beginningOffsets(Arrays.asList(partition0)); 获取指定TopicPartition的起始偏移量信息

说明:

  • subscribe()assign() 两个方法的使用是互斥的,只能使用其中之一。
  • assignment() 方法只能在 consumer 已经订阅主题或手动分配分区之后使用,否则返回空列表。

在执行subscribe()assign() 之后,assignment() 才能获取到偏移的位置,否返回一个 size() = 0 的 map。更加让人不能接收的是执行subscribe() 之后,仍然assignment() 不能获取 TopicPartition 信息。需要在执行了 poll() 方法之后才行。这就比较扯了。我本来想指定一个开始位置,但是还没有指定位置呢?先让我消费出来点数据,这不是多余吗?

在 FlinkKafkaConsumer 类中,使用了 assign() + seek 的方式,指定了消费者的消费位置,这样以来, __consumer_offsets 就有用了,如果 FlinkKafkaConsumer 不是从 checkpoint 消费,则可以指定 Consumer 的 startMap 模式,在不同的模式下,可以从最早、最新、上次消费的位置来开始消费。
FlinkKafkaConsumer 中的 startMode 参数表示消费者的启动模式,具体有以下几种:

startMode 中文名称 业务意义
GROUP_OFFSETS 消费者组偏移量 从消费者组中最近提交的偏移量开始消费,保证消费者组的偏移量和Kafka中的主题分区的偏移量保持一致,避免出现重复消费或者漏消费的情况
EARLIEST 最早的 从最早的消息开始消费,即消费者会从最早的偏移量开始消费Kafka中的消息
LATEST 最新的 从最新的消息开始消费,即消费者会从最新的偏移量开始消费Kafka中的消息
TIMESTAMP 时间戳 从指定的时间戳开始消费Kafka中的消息
SPECIFIC_OFFSETS 指定偏移量 从指定的偏移量开始消费Kafka中的消息

一般情况下,我们会使用 GROUP_OFFSETS 模式来消费 Kafka 中的消息,因为这样可以保证消费者组的偏移量和 Kafka 中的主题分区的偏移量保持一致,避免出现重复消费或者漏消费的情况。如果需要从最早或者最新的消息开始消费,可以选择 EARLIEST 或者 LATEST 模式。如果需要从指定的时间戳或者偏移量开始消费,可以选择 TIMESTAMP 或者 SPECIFIC_OFFSETS 模式。

关键代码

以下是加上注释的代码:

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

public class KafkaConsumerAt {
    private static Logger logger = LoggerFactory.getLogger(KafkaConsumerAt.class);

    public static void main(String[] args) throws InterruptedException {
        Properties p = new Properties();
        p.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG , "your.kafka.address:9092");
        p.setProperty(ConsumerConfig.GROUP_ID_CONFIG , "test-text-001");
        p.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"5"); // 设置一次拉取的最大记录数
        p.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG , "org.apache.kafka.common.serialization.StringDeserializer");
        p.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG , "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(p);

        String topic = "test_create_timestamp" ;

        // 获取指定主题的所有分区信息
        List<PartitionInfo> partitionInfos = consumer.partitionsFor(topic);
        TopicPartition topicPartition = null ;
        Map<TopicPartition, Long> tpl = new HashMap<>();
        for (PartitionInfo partitionInfo : partitionInfos) {
            topicPartition = new TopicPartition(partitionInfo.topic() , partitionInfo.partition());
            tpl.put(topicPartition , 0L); // 将所有分区的起始偏移量设置为0
        }

        List<TopicPartition> topicPartitions = Collections.singletonList(topicPartition);
        // 分配consumer消费者到指定分区进行消费,这里是最后一个分区
        consumer.assign(topicPartitions); 
        tpl.forEach((key,value)->{
        // 将所有分区的偏移量设置为起始偏移量,因为 value 的值在上一步中已经设置为 0 了。
            consumer.seek(key , value); 
        });

        ConsumerRecords<String,String> records = consumer.poll(Duration.ofSeconds(1L)); // 消费消息

        for (ConsumerRecord<String, String> record : records) {
            logger.info("key:{} value:{} offset:{} timestamp:{} timestampType:{}", record.key(), record.value(), record.offset(), record.timestamp(), record.timestampType()); // 输出消费的消息
        }
        consumer.close(); // 关闭消费者
    }
}

KafkaConsumer#partitionsFor方法用于获取指定主题的所有分区信息。它会返回一个List,每个PartitionInfo对象包含分区的详细信息,比如分区编号、分区的主题、分区的副本列表等等。

以下是一个简单的使用案例:

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

List<PartitionInfo> partitions = consumer.partitionsFor("my-topic");
for (PartitionInfo partition : partitions) {
    System.out.println(partition);
}

在上面的示例中,我们首先创建了一个KafkaConsumer对象,并设置了所需的属性。然后调用partitionsFor方法获取名为"my-topic"的主题的分区信息,最后打印出每个分区的详细信息。

KafkaConsumer#seek方法用于将消费者的偏移量(offset)设置到指定的分区和偏移量。它可以用于重置消费者在某个分区的消费进度,例如重新开始消费、跳过某些消息或者定位到特定的偏移量处继续消费等。

以下是一个简单的使用案例:

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

TopicPartition partition = new TopicPartition("my-topic", 0); // 设置要操作的分区
consumer.assign(Arrays.asList(partition)); // 订阅指定分区

consumer.seek(partition, 100); // 将偏移量设置为100

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); // 消费消息
for (ConsumerRecord<String, String> record : records) {
    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

在上面的示例中,我们首先创建了一个KafkaConsumer对象,并设置了所需的属性。然后订阅名为"my-topic"的主题中的分区0,并将偏移量设置为100。最后调用poll方法消费消息,并打印出每条消费的消息的偏移量、键和值。

.timeindex 文件,可以由 timestamp 找到对应的 offset 。
所以把 KafkaConsumer 提供给我们了 offsetsForTimes(Map<TopicPartition,Long>)
此方法可以根据指定的 partition 的 timestamp 找到对应的 offset , 然后使用 seek() 来指定开始消费的位置。

下面是代码:文章来源地址https://www.toymoban.com/news/detail-775494.html


public class ConsumerFromTimestamp{
	

	public static Map<TopicPartition,Long> fetchOffsetsWithTimestamp(Collection<KafkaTopicPartition> partitions , long timestamp , Properties properties){

		Map<TopicPartition, Long> partitionOffsetsRequest = new HashMap(partitions.size());

		Map<TopicPartition,Long> rs = new HashMap<>();

		for(KafkaTopicPartition partition : partitions){
			partitionOffsetsRequest.put(new TopicPartition(partition.topic(),partition.partition()),timestamp);
		}
		KafkaConsumer<?, ?> consumer = new KafkaConsumer(properties);
		Set<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(partitionOffsetsRequest);
		for(Entry<TopicPartition, OffsetAndTimestamp> partitionToOffset : offsets.entrySet()){
			rs.put(new KafkaTopicPartition(((TopicPartition)partitionToOffset.getKey()).topic(), ((TopicPartition)partitionToOffset.getKey()).partition()), partitionToOffset.getValue() == null ? null : ((OffsetAndTimestamp)partitionToOffset.getValue()).offset());
		}
		return rs ;
	}

} 

到了这里,关于KafkaConsuner 指定开始消费的位置的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包