跳转到特定偏移量

为了进行跳转,您的监听器必须实现 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 在容器启动时以及每次分配分区时被调用。在初始化后任意时间进行跳转时,您应该使用此回调。您应该保存对回调的引用。如果您在多个容器(或在 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);

seek 方法的两种不同变体提供了一种跳转到任意偏移量的方法。以 Function 作为参数计算偏移量的方法是在框架的 3.2 版本中添加的。此函数提供了对当前偏移量(消费者返回的当前位置,即要获取的下一个偏移量)的访问。用户可以根据消费者中当前偏移量作为函数定义的一部分来决定跳转到的偏移量。

seekRelative 添加于 2.3 版本,用于执行相对跳转。

  • offset 为负且 toCurrentfalse - 相对于分区的末尾进行相对跳转。

  • offset 为正且 toCurrentfalse - 相对于分区的开头进行相对跳转。

  • offset 为负且 toCurrenttrue - 相对于当前位置进行相对跳转(倒带)。

  • offset 为正且 toCurrenttrue - 相对于当前位置进行相对跳转(快进)。

seekToTimestamp 方法也添加于 2.3 版本。

onIdleContaineronPartitionsAssigned 方法中为多个分区跳转到相同的时间戳时,第二种方法更可取,因为在对消费者的 offsetsForTimes 方法进行一次调用中查找时间戳的偏移量效率更高。当从其他位置调用时,容器将收集所有时间戳跳转请求并对 offsetsForTimes 进行一次调用。

当检测到空闲容器时,您还可以从 onIdleContainer() 执行跳转操作。请参阅检测空闲和无响应的消费者,了解如何启用空闲容器检测。

接受集合的 seekToBeginning 方法很有用,例如,当处理压缩主题并且您希望在每次应用程序启动时都跳转到开头时。
public class MyListener implements ConsumerSeekAware {

    ...

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

}

要在运行时任意跳转,请使用来自 registerSeekCallback 的相应线程的回调引用。

这是一个演示如何使用回调的简单 Spring Boot 应用程序;它向主题发送 10 条记录;在控制台中按 <Enter> 会导致所有分区跳转到开头。

@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类,该类跟踪哪个回调函数将用于某个主题/分区。以下示例展示了如何在每次容器空闲时,将每个分区的读取位置都设置为上次处理记录的位置。它还提供了一些方法,允许外部程序任意地将分区回退一个记录。

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 版本在抽象类中添加了便利方法

  • seekToBeginning() - 将所有分配的分区读取位置设置为开头。

  • seekToEnd() - 将所有分配的分区读取位置设置为结尾。

  • seekToTimestamp(long timestamp) - 将所有分配的分区读取位置设置为由该时间戳表示的偏移量。

示例

public class MyListener extends AbstractConsumerSeekAware {

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

public class SomeOtherBean {

    MyListener listener;

    ...

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

}