KafkaListener 监听多个主题时，能否优先考虑单个主题？答案

【问题标题】：Can you give priority to a single topic when KafkaListener listens to multiple topics?KafkaListener 监听多个主题时，能否优先考虑单个主题？
【发布时间】：2021-05-27 16:08:42
【问题描述】：

当我的@KafkaListener 监听多个主题时，有人知道如何优先考虑单个 kafka 主题吗？

下面是我的代码示例：

@KafkaListener(id = "priority", topics = { "${prio-topic}" }, concurrency = "1", autoStartup = "true")
    @KafkaListener(id = "nonPriority", topics = { "${not-prio-topic-1}",
            "${not-prio-topic-2}", "${not-prio-topic-3}",
            "${not-prio-topic-4}", concurrency = "1", autoStartup = "true")
    public synchronized void  listenManyEntryTopic(String message) {}

我的问题是我想在其他 nonPrio 主题之前阅读主题 prio-topic。只有当我的prio-topic 为空时，我才应该开始使用其他主题而无需任何特定顺序。任何提示/建议表示赞赏。感谢您的帮助！

【问题讨论】：

标签： java spring apache-kafka listener spring-kafka

【解决方案1】：

由于您有两个侦听器容器，您可以将@Header(KafkaHeaders.RECEIVED_TOPIC) String topic 作为参数添加到您的方法中。

然后，当您收到来自优先主题的消息时，您可以 stop() 另一个侦听器容器（使用 KafkaListenerEndpointRegistry bean），如果它正在运行。

在主主题容器中配置idleEventInterval 并为ListenerContainerIdleEvents（或ApplicationListener bean）添加@EventListner 方法。

然后，当您检测到空闲的主容器时，您可以重新启动非主容器。

编辑

@SpringBootApplication
public class So66366140Application {

    private static final Logger LOG = LoggerFactory.getLogger(So66366140Application.class);

    public static void main(String[] args) {
        SpringApplication.run(So66366140Application.class, args);
    }

    @Bean
    public NewTopic topic1() {
        return TopicBuilder.name("so66366140-1").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic topic2() {
        return TopicBuilder.name("so66366140-2").partitions(1).replicas(1).build();
    }

    @Autowired
    KafkaListenerEndpointRegistry registry;

    @KafkaListener(id = "so66366140-1", topics = "so66366140-1")
    @KafkaListener(id = "so66366140-2", topics = "so66366140-2", autoStartup = "false")
    public void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        LOG.info(in);
        if (topic.equals("so66366140-1")
                && this.registry.getListenerContainer("so66366140-2").isRunning()) {
            LOG.info("Stopping non-pri container");
            this.registry.getListenerContainer("so66366140-2").stop();
        }
    }

    @EventListener
    void events(ListenerContainerIdleEvent event) {
        LOG.info(event.toString());
        if (event.getListenerId().startsWith("so66366140-1")
                && !this.registry.getListenerContainer("so66366140-2").isRunning()) {
            LOG.info("Starting non-pri container");
            this.registry.getListenerContainer("so66366140-2").start();
        }
    }

    @Bean
    public ApplicationRunner runner(KafkaTemplate<String, String> template) {
        return args -> {
            IntStream.range(0,  10).forEach(i -> {
                template.send("so66366140-1", "foo");
                template.send("so66366140-2", "bar");
                try {
                    Thread.sleep(6_000);
                }
                catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        };
    }

}

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.listener.idle-event-interval=5s

您也可以使用暂停/恢复来代替停止/开始。

【讨论】：

我添加了一个例子。
请注意，这将为非优先主题中出现的第一条消息添加等于 idleEventInterval 的延迟。如果应用程序希望尽快读取所有内容但仍保持优先级，则两个单独的消费者可能更可取。
对；但他已经有两个消费者（通过 2 个@KafkaListener 注释），所以这是内置的。

【解决方案2】：

kafka 中没有区分优先级和非优先级主题消息的功能。为了解决您的问题，其中一种解决方案是将优先处理主题与非优先主题分开，即专用 app1 仅消费优先主题消息并处理它们，同时 app2 消费非优先消息并处理它们。

【讨论】：