按顺序启动 @KafkaListener

Doubao 中英对照 Starting @KafkaListeners in Sequence

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

最好通过一个示例来说明。

@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。现在这些 collection 已被弃用，取而代之的是类型为 ContainerGroup 的 bean，其 bean 名是组名加上 .group 后缀；在上面的示例中，会有 2 个 bean：g1.group 和 g2.group。Collection bean 将在未来的版本中被移除。