【Spring连载】使用Spring访问 Apache Kafka(七)----Topic/Partition初始偏移量&定位到特定的偏移量

发布时间:2024年01月17日

【Spring连载】使用Spring访问 Apache Kafka(七)----Topic/Partition初始偏移量&定位到特定的偏移量

一、Topic/Partition的初始偏移量Initial Offset

有几种方法可以设置分区的初始偏移量。
手动分配分区时,可以在配置的TopicPartitionOffset参数中设置初始偏移量(如果需要)(请参阅消息监听器容器)。你也可以在任何时候定位到特定的偏移量。
当你使用broker分配分区的组管理时:

  • 对于新的group.id,初始偏移量由consumer 属性auto.offset.reset确定(earliest 或 latest)。
  • 对于已存在的组ID,初始偏移量是该组ID的当前偏移量。但是,你可以在初始化期间(或此后的任何时间)定位到特定偏移量。

二、定位到特定的偏移量Seeking to a Specific Offset

为了seek,监听器必须实现ConsumerSeekAware,它有以下方法:

void registerSeekCallback(ConsumerSeekCallback callback);

void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

void onPartitionsRevoked(Collection<TopicPartition> partitions)

void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

registerSeekCallback在容器启动和分配分区时调用。在初始化后的任意时间进行seek时,应使用此回调。你应该保存一个对回调的引用。如果在多个容器(或ConcurrentMessageListenerContainer)中使用同一个监听器,则应将回调存储在ThreadLocal或由监听器Thread作为key的其他结构中。
使用组管理时,分配分区时会调用onPartitionsAssigned。例如,您可以使用此方法,通过调用回调来设置分区的初始偏移量。您还可以使用此方法将此线程的回调与分配的分区相关联(请参阅下面的示例)。你必须使用回调参数,而不是传递到registerSeekCallback的回调。即使使用手动分区分配,也会调用此方法。
onPartitionsRevoked在容器停止或Kafka撤消分配时调用。你应该放弃此线程的回调,并删除与已撤销分区的任何关联。
回调具有以下方法:

void seek(String topic, int partition, long offset);

void seekToBeginning(String topic, int partition);

void seekToBeginning(Collection=<TopicPartitions> partitions);

void seekToEnd(String topic, int partition);

void seekToEnd(Collection=<TopicPartitions> partitions);

void seekRelative(String topic, int partition, long offset, boolean toCurrent);

void seekToTimestamp(String topic, int partition, long timestamp);

void seekToTimestamp(Collection<TopicPartition> topicPartitions, long timestamp);

seekRelative用于执行相对查找。

  • offset 为负值和toCurrent false——相对于分区末尾的查找。
  • offset 为正值和toCurrent false——相对于分区的开始查找。
  • offset 为负值和toCurrent true——相对于当前位置搜索(倒回)。
  • offset 为正值和toCurrent true——相对于当前位置搜索(前进)。

当在onIdleContainer或onPartitionsAssigned方法中为多个分区seek到相同的时间戳时,首选第二种方法,因为在对consumer的offsetsForTimes方法的单个调用中查找时间戳的偏移量更有效。当从其他位置调用时,容器将收集所有时间戳seek请求,并对offsetsForTimes进行一次调用。
当检测到空闲容器时,还可以从onIdleContainer()执行seek操作。请参阅检测空闲和无响应消费者,了解如何启用空闲容器检测。
接受集合的seekToBeginning方法很有用,例如,当处理compacted topic时,你希望每次启动应用程序时都seek到开头:

public class MyListener implements ConsumerSeekAware {
...
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        callback.seekToBeginning(assignments.keySet());
    }
}

要在运行时任意seek,请使用相应线程的registerSeekCallback中的回调引用。
下面是一个简单的Spring Boot应用程序,它展示了如何使用回调;它向topic发送了10条记录;在控制台中点击<Enter>会导致所有分区都seek到开头。

@SpringBootApplication
public class SeekExampleApplication {

    public static void main(String[] args) {
        SpringApplication.run(SeekExampleApplication.class, args);
    }

    @Bean
    public ApplicationRunner runner(Listener listener, KafkaTemplate<String, String> template) {
        return args -> {
            IntStream.range(0, 10).forEach(i -> template.send(
                new ProducerRecord<>("seekExample", i % 3, "foo", "bar")));
            while (true) {
                System.in.read();
                listener.seekToStart();
            }
        };
    }

    @Bean
    public NewTopic topic() {
        return new NewTopic("seekExample", 3, (short) 1);
    }

}

@Component
class Listener implements ConsumerSeekAware {

    private static final Logger logger = LoggerFactory.getLogger(Listener.class);

    private final ThreadLocal<ConsumerSeekCallback> callbackForThread = new ThreadLocal<>();

    private final Map<TopicPartition, ConsumerSeekCallback> callbacks = new ConcurrentHashMap<>();

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        this.callbackForThread.set(callback);
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        assignments.keySet().forEach(tp -> this.callbacks.put(tp, this.callbackForThread.get()));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        partitions.forEach(tp -> this.callbacks.remove(tp));
        this.callbackForThread.remove();
    }

    @Override
    public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
    }

    @KafkaListener(id = "seekExample", topics = "seekExample", concurrency = "3")
    public void listen(ConsumerRecord<String, String> in) {
        logger.info(in.toString());
    }

    public void seekToStart() {
        this.callbacks.forEach((tp, callback) -> callback.seekToBeginning(tp.topic(), tp.partition()));
    }

}

另外,也可以使用AbstractConsumerSeekAware类,该类可以跟踪topic/partition要使用的回调。以下示例展示了每当容器空闲时,如何在每个分区中查找已处理的最后一条记录。它还有允许任意外部调用在分区上倒退一条记录的方法。

public class SeekToLastOnIdleListener extends s {

    @KafkaListener(id = "seekOnIdle", topics = "seekOnIdle")
    public void listen(String in) {
        ...
    }

    @Override
    public void onIdleContainer(Map<org.apache.kafka.common.TopicPartition, Long> assignments,
            ConsumerSeekCallback callback) {

            assignments.keySet().forEach(tp -> callback.seekRelative(tp.topic(), tp.partition(), -1, true));
    }

    /**
    * Rewind all partitions one record.
    */
    public void rewindAllOneRecord() {
        getSeekCallbacks()
            .forEach((tp, callback) ->
                callback.seekRelative(tp.topic(), tp.partition(), -1, true));
    }

    /**
    * Rewind one partition one record.
    */
    public void rewindOnePartitionOneRecord(String topic, int partition) {
        getSeekCallbackFor(new org.apache.kafka.common.TopicPartition(topic, partition))
            .seekRelative(topic, partition, -1, true);
    }

}

再介绍一下AbstractConsumerSeekAware比较方便的几个方法:

  • seekToBeginning()——定位到所有分配的分区的开始
  • seekToEnd()——定位到所有分配的分区的结束
  • seekToTimestamp(long time)——在所有分配的分区定位到该时间戳表示的偏移量
public class MyListener extends AbstractConsumerSeekAware {

    @KafkaListener(...)
    void listn(...) {
        ...
    }
}

public class SomeOtherBean {

    MyListener listener;

    ...

    void someMethod() {
        this.listener.seekToTimestamp(System.currentTimeMillis - 60_000);
    }

}
文章来源:https://blog.csdn.net/gabriel_wang_sh/article/details/135626988
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。