|
此版本仍在开发中，尚不被认为是稳定的。对于最新的稳定版本，请使用 Spring for Apache Kafka 3.3.9！ |
暂停和恢复侦听器容器
添加了 2.1.3 版本pause()和resume()方法到监听器容器。
以前，您可以在ConsumerAwareMessageListener并通过监听ListenerContainerIdleEvent，它提供了对Consumer对象。
虽然可以使用事件侦听器暂停空闲容器中的使用者，但在某些情况下，这不是线程安全的，因为不能保证在使用者线程上调用事件侦听器。
要安全地暂停和恢复消费者，您应该使用pause和resume侦听器容器上的方法。
一个pause()在下一个之前生效poll();一个resume()在当前之后生效poll()返回。
当容器暂停时，它将继续poll()使用者，如果正在使用组管理，则避免重新平衡，但它不会检索任何记录。
有关更多信息，请参阅 Kafka 文档。
从 2.1.5 版开始，您可以调用isPauseRequested()看看是否pause()被叫了。
然而，消费者可能还没有真正停下来。isConsumerPaused()如果全部Consumer实例实际上已经暂停了。
此外（也是从 2.1.5 开始），ConsumerPausedEvent和ConsumerResumedEvent实例与容器一起发布为source属性和TopicPartition涉及的实例partitions财产。
从 2.9 版开始，新的容器属性pauseImmediate，当设置为 true 时，会导致暂停在处理当前记录后生效。
默认情况下，暂停在处理完上一次轮询中的所有记录后生效。
请参阅 pauseImmediate。
以下简单的 Spring Boot 应用程序演示了使用容器注册表获取对@KafkaListener方法的容器并暂停或恢复其消费者以及接收相应的事件：
@SpringBootApplication
public class Application implements ApplicationListener<KafkaEvent> {
public static void main(String[] args) {
SpringApplication.run(Application.class, args).close();
}
@Override
public void onApplicationEvent(KafkaEvent event) {
System.out.println(event);
}
@Bean
public ApplicationRunner runner(KafkaListenerEndpointRegistry registry,
KafkaTemplate<String, String> template) {
return args -> {
template.send("pause.resume.topic", "thing1");
Thread.sleep(10_000);
System.out.println("pausing");
registry.getListenerContainer("pause.resume").pause();
Thread.sleep(10_000);
template.send("pause.resume.topic", "thing2");
Thread.sleep(10_000);
System.out.println("resuming");
registry.getListenerContainer("pause.resume").resume();
Thread.sleep(10_000);
};
}
@KafkaListener(id = "pause.resume", topics = "pause.resume.topic")
public void listen(String in) {
System.out.println(in);
}
@Bean
public NewTopic topic() {
return TopicBuilder.name("pause.resume.topic")
.partitions(2)
.replicas(1)
.build();
}
}以下列表显示了上述示例的结果：
partitions assigned: [pause.resume.topic-1, pause.resume.topic-0]
thing1
pausing
ConsumerPausedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
resuming
ConsumerResumedEvent [partitions=[pause.resume.topic-1, pause.resume.topic-0]]
thing2