寻找特定偏移量 (Seeking to a Specific Offset) | Spring for Apache Kafka3.1.10-SNAPSHOT中文文档|Spring官方文档|SpringBoot 教程|Spring中文网

此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring for Apache Kafka 3.2.4！spring-doc.cn

为了进行查找，您的侦听器必须实现，它具有以下方法：ConsumerSeekAwarespring-doc.cn

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);

当容器启动时以及分配分区时，将调用 The 。在初始化后的任意时间进行 seek 时，您应该使用此回调。您应该保存对回调的引用。如果您在多个容器（或）中使用相同的侦听器，则应将回调存储在 a 或由侦听器键控的其他结构中。registerSeekCallbackConcurrentMessageListenerContainerThreadLocalThreadspring-doc.cn

使用组管理时，在分配分区时调用。例如，您可以使用此方法通过调用 callback 来设置 partition 的初始偏移量。您还可以使用此方法将此线程的回调与分配的分区相关联（请参阅下面的示例）。您必须使用 callback 参数，而不是传递给 . 从版本 2.5.5 开始，即使使用手动分区分配，也会调用此方法。onPartitionsAssignedregisterSeekCallbackspring-doc.cn

onPartitionsRevoked在容器停止或 Kafka 撤销分配时调用。您应该丢弃此线程的回调并删除与已撤销分区的任何关联。spring-doc.cn

该回调有以下方法：spring-doc.cn

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在版本 2.3 中添加，以执行相对查找。spring-doc.cn

offsetnegative 和 - 相对于分区末尾查找。toCurrentfalsespring-doc.cn
offsetpositive 和 - seek 相对于分区的开头。toCurrentfalsespring-doc.cn
offsetnegative 和 - 相对于当前位置查找（rewind）。toCurrenttruespring-doc.cn
offsetpositive 和 - 相对于当前位置进行搜索（快进）。toCurrenttruespring-doc.cn

这些方法也在 2.3 版中添加。seekToTimestampspring-doc.cn

在 or 方法中为多个分区查找相同的时间戳时，首选第二种方法，因为在对使用者方法的单次调用中查找时间戳的偏移量效率更高。从其他位置调用时，容器将收集所有时间戳查找请求，并对 .onIdleContaineronPartitionsAssignedoffsetsForTimesoffsetsForTimes

您还可以在检测到空闲容器时执行查找操作。有关如何启用空闲容器检测的信息，请参阅检测空闲和无响应的使用者。onIdleContainer()spring-doc.cn

接受集合的方法很有用，例如，在处理压缩的主题时，并且您希望在每次启动应用程序时都寻找开头：seekToBeginning

public class MyListener implements ConsumerSeekAware {

    ...

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

}

要在运行时任意查找，请使用 for 相应线程的回调引用。registerSeekCallbackspring-doc.cn

这是一个简单的 Spring Boot 应用程序，它演示了如何使用回调;它向主题发送 10 条记录;点击控制台会导致所有分区查找到开头。<Enter>spring-doc.cn

@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()));
    }

}

为了简单起见，版本 2.3 添加了类，该类跟踪要用于主题/分区的回调。以下示例演示如何在每次容器空闲时在每个分区中查找处理的最后一条记录。它还具有允许任意外部调用按一条记录倒带分区的方法。AbstractConsumerSeekAwarespring-doc.cn

public class SeekToLastOnIdleListener extends AbstractConsumerSeekAware {

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

    @Override
    public void onIdleContainer(Map<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 TopicPartition(topic, partition))
            .seekRelative(topic, partition, -1, true);
    }

}

版本 2.6 向 abstract 类添加了便捷方法：spring-doc.cn

seekToBeginning()- 查找所有已分配的分区到开头。spring-doc.cn
seekToEnd()- 查找所有分配的分区到末尾。spring-doc.cn
seekToTimestamp(long timestamp)- 将所有分配的分区查找到该时间戳表示的偏移量。spring-doc.cn

例：spring-doc.cn

public class MyListener extends AbstractConsumerSeekAware {

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

public class SomeOtherBean {

    MyListener listener;

    ...

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

}