@KafkaListener 生命周期管理
为 @KafkaListener 注解创建的监听器容器不是应用程序上下文中的 bean。相反,它们注册到一个 KafkaListenerEndpointRegistry 类型的基础设施 bean 中。此 bean 由框架自动声明并管理容器的生命周期;它将自动启动任何将 autoStartup 设置为 true 的容器。所有容器工厂创建的所有容器必须处于相同的 phase。有关更多信息,请参阅 监听器容器自动启动。您可以通过使用注册表以编程方式管理生命周期。启动或停止注册表将启动或停止所有注册的容器。或者,您可以通过使用其 id 属性获取对单个容器的引用。您可以在注解上设置 autoStartup,它将覆盖容器工厂中配置的默认设置。您可以从应用程序上下文(例如自动装配)获取对 bean 的引用,以管理其注册的容器。以下示例展示了如何做到这一点
@KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
public void listen(...) { ... }
@Autowired
private KafkaListenerEndpointRegistry registry;
...
this.registry.getListenerContainer("myContainer").start();
...
注册表只维护它管理的容器的生命周期;声明为 bean 的容器不受注册表管理,可以从应用程序上下文获取。可以通过调用注册表的 getListenerContainers() 方法获取托管容器的集合。版本 2.2.5 添加了一个便捷方法 getAllListenerContainers(),它返回所有容器的集合,包括由注册表管理的容器和声明为 bean 的容器。返回的集合将包含任何已初始化的原型 bean,但不会初始化任何延迟 bean 声明。
应用程序上下文刷新后注册的端点将立即启动,无论其 autoStartup 属性如何,以遵守 SmartLifecycle 契约,其中 autoStartup 仅在应用程序上下文初始化期间考虑。后期注册的一个例子是具有 @KafkaListener 的原型范围的 bean,其中在上下文初始化后创建实例。从版本 2.8.7 开始,您可以将注册表的 alwaysStartAfterRefresh 属性设置为 false,然后容器的 autoStartup 属性将定义容器是否启动。 |
从 KafkaListenerEndpointRegistry 中检索 MessageListenerContainers
KafkaListenerEndpointRegistry 提供了用于检索 MessageListenerContainer 实例的方法,以适应各种管理场景
所有容器:对于涵盖所有监听器容器的操作,使用 getListenerContainers() 检索全面的集合。
Collection<MessageListenerContainer> allContainers = registry.getListenerContainers();
按 ID 指定容器:要管理单个容器,getListenerContainer(String id) 允许通过其 ID 进行检索。
MessageListenerContainer specificContainer = registry.getListenerContainer("myContainerId");
动态容器筛选:在版本 3.2 中引入的两个重载 getListenerContainersMatching 方法,允许对容器进行精细选择。一个方法将 Predicate<String> 作为参数用于基于 ID 的筛选,而另一个方法将 BiPredicate<String, MessageListenerContainer> 作为参数用于更高级别的筛选,其中可能包括容器属性或状态。
// Prefix matching (Predicate<String>)
Collection<MessageListenerContainer> filteredContainers =
registry.getListenerContainersMatching(id -> id.startsWith("productListener-retry-"));
// Regex matching (Predicate<String>)
Collection<MessageListenerContainer> regexFilteredContainers =
registry.getListenerContainersMatching(myPattern::matches);
// Pre-built Set of IDs (Predicate<String>)
Collection<MessageListenerContainer> setFilteredContainers =
registry.getListenerContainersMatching(myIdSet::contains);
// Advanced Filtering: ID prefix and running state (BiPredicate<String, MessageListenerContainer>)
Collection<MessageListenerContainer> advancedFilteredContainers =
registry.getListenerContainersMatching(
(id, container) -> id.startsWith("specificPrefix-") && container.isRunning()
);
利用这些方法可以高效地管理和查询应用程序中的 MessageListenerContainer 实例。