按顺序启动 @KafkaListener

一个常见的用例是在一个监听器消费完一个主题中的所有记录后,再启动另一个监听器。例如,在处理来自其他主题的记录之前,您可能希望将一个或多个压缩主题的内容加载到内存中。从版本 2.7.3 开始,引入了一个新的组件 ContainerGroupSequencer。它使用 @KafkaListenercontainerGroup 属性将容器分组,并在当前组中的所有容器都空闲时启动下一个组中的容器。

通过一个例子最能说明这一点。

@KafkaListener(id = "listen1", topics = "topic1", containerGroup = "g1", concurrency = "2")
public void listen1(String in) {
}

@KafkaListener(id = "listen2", topics = "topic2", containerGroup = "g1", concurrency = "2")
public void listen2(String in) {
}

@KafkaListener(id = "listen3", topics = "topic3", containerGroup = "g2", concurrency = "2")
public void listen3(String in) {
}

@KafkaListener(id = "listen4", topics = "topic4", containerGroup = "g2", concurrency = "2")
public void listen4(String in) {
}

@Bean
ContainerGroupSequencer sequencer(KafkaListenerEndpointRegistry registry) {
    return new ContainerGroupSequencer(registry, 5000, "g1", "g2");
}

这里,我们在 g1 和 g2 两个组中有 4 个监听器。

在应用上下文初始化期间,sequencer 将所提供组中所有容器的 autoStartup 属性设置为 false。它还将任何容器(尚未设置该属性的)的 idleEventInterval 设置为提供的值(在本例中为 5000ms)。然后,当 sequencer 由应用上下文启动时,将启动第一个组中的容器。当收到 ListenerContainerIdleEvent 时,每个容器中的每个单独子容器都会停止。当 ConcurrentMessageListenerContainer 中的所有子容器都停止时,父容器会停止。当一个组中的所有容器都停止后,将启动下一个组中的容器。组的数量或组中的容器数量没有限制。

默认情况下,最后一个组(即上面的 g2)中的容器在空闲时不停止。要修改此行为,请在 sequencer 上将 stopLastGroupWhenIdle 设置为 true

顺带一提,之前每个组中的容器都被添加到一个类型为 Collection<MessageListenerContainer> 的 bean 中,其 bean 名称就是 containerGroup。这些集合现在已被弃用,取而代之的是类型为 ContainerGroup 的 bean,其 bean 名称是组名后加上 .group 后缀;在上面的例子中,将会有两个 bean,g1.groupg2.groupCollection bean 将在未来的版本中移除。