Seek 到特定偏移量

Doubao 中英对照 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 为键的结构中。

使用消费者组管理时，onPartitionsAssigned 在分配分区时被调用。例如，你可以使用此方法通过调用回调来为分区设置初始偏移量。你也可以使用此方法将该线程的回调与已分配分区关联（见下例）。你必须使用回调参数，而不是传入 registerSeekCallback 的那个。从 2.5.5 版本开始，即使在使用 主题分区分配 时也会调用此方法。

onPartitionsRevoked 在容器停止或 Kafka 撤销分配时被调用。你应丢弃该线程的回调，并移除与被撤销分区的任何关联。

该回调具有以下方法：

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

void seek(String topic, int partition, Function<Long, Long> offsetComputeFunction);

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

String getGroupId();

seek 方法的两种不同变体提供了 seek 到任意偏移量的方式。以 Function 作为参数来计算偏移量的方法是在框架 3.2 版本中新增的。该函数提供对当前偏移量（消费者返回的当前位置，即下一次要获取的偏移量）的访问。用户可以在函数定义中根据消费者中的当前偏移量决定要 seek 到哪个偏移量。

seekRelative 在 2.3 版本中新增，用于执行相对 seek。

• offset 为负且 toCurrent 为 false——相对于分区末尾进行 seek。
• offset 为正且 toCurrent 为 false——相对于分区开头进行 seek。
• offset 为负且 toCurrent 为 true——相对于当前位置进行 seek（回退）。
• offset 为正且 toCurrent 为 true——相对于当前位置进行 seek（快进）。

seekToTimestamp 方法也是在 2.3 版本中新增的。

备注

当在 onIdleContainer 或 onPartitionsAssigned 方法中为多个分区 seek 到同一时间戳时，更推荐使用第二种方法，因为在单次调用消费者的 offsetsForTimes 方法中查找这些时间戳的偏移量效率更高。从其他位置调用时，容器会收集所有时间戳 seek 请求，并对 offsetsForTimes 进行一次调用。

你也可以在检测到空闲容器时，从 onIdleContainer() 执行 seek 操作。有关如何启用空闲容器检测，请参阅 检测空闲和无响应的消费者。

备注

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

public class MyListener implements ConsumerSeekAware {

    ...

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

}

要在运行时任意 seek，请对相应线程使用来自 registerSeekCallback 的回调引用。

以下是一个简单的 Spring Boot 应用，演示了如何使用该回调；它向主题发送 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()));
    }

}

为简化操作，2.3 版本新增了 AbstractConsumerSeekAware 类，它会跟踪某个主题/分区应使用哪个回调。以下示例展示了如何在每次容器空闲时，将每个分区 seek 到已处理的最后一条记录。它还有一些方法，允许任意外部调用将分区回退一条记录。

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() {
        getTopicsAndCallbacks()
            .forEach((tp, callbacks) ->
                    callbacks.forEach(callback -> callback.seekRelative(tp.topic(), tp.partition(), -1, true))
            );
    }

    /**
    * Rewind one partition one record.
    */
    public void rewindOnePartitionOneRecord(String topic, int partition) {
        getSeekCallbacksFor(new TopicPartition(topic, partition))
            .forEach(callback -> callback.seekRelative(topic, partition, -1, true));
    }

}

2.6 版本向该抽象类新增了一些便捷方法：

• seekToBeginning()——将所有已分配分区 seek 到开头。
• seekToEnd()——将所有已分配分区 seek 到末尾。
• seekToTimestamp(long timestamp)——将所有已分配分区 seek 到该时间戳所表示的偏移量。

示例：

public class MyListener extends AbstractConsumerSeekAware {

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

public class SomeOtherBean {

    MyListener listener;

    ...

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

}

从 3.3 版本开始，ConsumerSeekAware.ConsumerSeekCallback 接口中引入了一个新方法 getGroupId()。当你需要识别与某个特定 seek 回调关联的消费者组时，此方法特别有用。

备注

当使用继承 AbstractConsumerSeekAware 的类时，在一个监听器中执行的 seek 操作可能会影响同一类中的所有监听器。这未必总是期望的行为。为解决这个问题，你可以使用回调提供的 getGroupId() 方法。这使你能够有选择地执行 seek 操作，只针对感兴趣的消费者组。