赞
踩
有几种方法可以设置分区的初始偏移量。
手动分配分区时,可以在配置的TopicPartitionOffset参数中设置初始偏移量(如果需要)(请参阅消息监听器容器)。你也可以在任何时候定位到特定的偏移量。
当你使用broker分配分区的组管理时:
为了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用于执行相对查找。
当在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比较方便的几个方法:
public class MyListener extends AbstractConsumerSeekAware { @KafkaListener(...) void listn(...) { ... } } public class SomeOtherBean { MyListener listener; ... void someMethod() { this.listener.seekToTimestamp(System.currentTimeMillis - 60_000); } }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。