如果你想在监听器本身中寻找消费者，只需在监听器方法中添加一个Consumer<?, ?> consumer参数即可。
请记住，容器可能已获取更多消息，因此您将在搜索生效之前获取它们。你可以设置max.poll.records=1 来避免这种情况。
你也可以在容器中添加自定义RemainingRecordsErrorHandler，在监听器中抛出异常，错误处理程序会获取剩余的记录而不是监听器。
另见Seeking to a Specific Offset。
为了寻找，你的监听器必须实现 ConsumerSeekAware，它有以下方法：
void registerSeekCallback(ConsumerSeekCallback callback);
void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);
void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);
第一个方法在容器启动时被调用。在初始化后的任意时间进行搜索时，您应该使用此回调。您应该保存对回调的引用。如果您在多个容器（或 ConcurrentMessageListenerContainer）中使用相同的侦听器，则应将回调存储在 ThreadLocal 或其他由侦听器 Thread 键入的结构中。
当使用组管理时，当分配改变时调用第二种方法。您可以使用此方法，例如，通过调用回调设置分区的初始偏移量。您必须使用回调参数，而不是传递给 registerSeekCallback 的参数。如果您自己显式分配分区，则永远不会调用此方法。在这种情况下使用 TopicPartitionInitialOffset。
回调有以下方法：
void seek(String topic, int partition, long offset);
void seekToBeginning(String topic, int partition);
void seekToEnd(String topic, int partition);
当检测到空闲容器时，您还可以从 onIdleContainer() 执行查找操作。有关如何启用空闲容器检测的信息，请参阅检测空闲和非响应消费者。
要在运行时任意搜索，请使用来自 registerSeekCallback 的回调引用以用于相应的线程。
这是一个例子；我们跟踪每个主题/分区的回调...
@SpringBootApplication
public class So56584233Application {
public static void main(String[] args) {
SpringApplication.run(So56584233Application.class, args);
}
@Bean
public ApplicationRunner runner(Listener listener, KafkaTemplate<String, String> template) {
return args -> {
IntStream.range(0, 10).forEach(i -> template.send(new ProducerRecord<>("so56584233", i % 3, "foo", "bar")));
while (true) {
System.in.read();
listener.seekToStart();
}
};
}
@Bean
public NewTopic topic() {
return new NewTopic("so56584233", 3, (short) 1);
}
}
@Component
class Listener implements ConsumerSeekAware {
private static final Logger logger = LoggerFactory.getLogger(Listener.class);
private final Map<TopicPartition, ConsumerSeekCallback> callbacks = new ConcurrentHashMap<>();
private static final ThreadLocal<ConsumerSeekCallback> callbackForThread = new ThreadLocal<>();
@Override
public void registerSeekCallback(ConsumerSeekCallback callback) {
callbackForThread.set(callback);
}
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
assignments.keySet().forEach(tp -> this.callbacks.put(tp, callbackForThread.get()));
}
@Override
public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
}
@KafkaListener(id = "so56584233", topics = "so56584233", 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()));
}
}