如何将消息从单个生产者发布到两个主题并在 Kafka 的单个侦听器中消费答案

【问题标题】：How to publish the messages to two topics from single producer and consuming in single listener in Kafka如何将消息从单个生产者发布到两个主题并在 Kafka 的单个侦听器中消费
【发布时间】：2021-08-18 16:17:03
【问题描述】：

我正在尝试从单个生产者发布两个不同主题的消息。

我在这里创建了两个主题：

   @Bean
    public NewTopic multi1() {
        return TopicBuilder.name("multi1").partitions(1).build();
    }
    @Bean
    public NewTopic multi2() {
        return TopicBuilder.name("multi2").partitions(1).build();
    }

这就是我将消息发送到两个主题的方式：

public void sendingtomultitopic()
{
    IntStream.range(0, 100).forEach(i->this.template.send("multi1", "mutli1 data value->"+i));      
    IntStream.range(0, 100).forEach(i->this.template.send("multi2", "multi2 data value->"+i));
    logger.info("sending finished");
}

上述方法是发送到多个主题的正确方法吗？

下面我正在尝试使用消息

@KafkaListener(topics = {"multi1,multi2"}, groupId = "diffgroupid3")
    public void consumingfrommultitopics(String data) {
        logger.info(String.format("consumingfromtwotopics -> %s", data));
    }

我收到异常：

org.apache.kafka.common.errors.InvalidTopicException: Invalid topics: [multi1,multi2]

我能够在这两个主题中发布数据。

但是这个消费者没有检索到任何消息，请在这里帮助我吗？

【问题讨论】：

尝试定义多个主题名称，例如topics = {"multi1","multi2"}
是的，我错过了那个拼写错误
这是将消息发布到多个主题的正确方法
是的，我想是的
是的；这是正确的。

标签： spring spring-boot apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

您在“主题”表单中有错字，应该是

@KafkaListener(topics = {"multi1","multi2"}...

（查看Can a single Spring's KafkaConsumer listener listens to multiple topic?）

您还可以有两个单独的@KafkaListener，每个主题对应一个。什么最适合您的用例。

【讨论】：

【解决方案2】：

您可以从配置（application.yml）中使用

test:
  kafka:
    sender:
      serverAddress: xxxxx:port
    topic:
      consumer: topic1,topic2

@KafkaListener(topics = "#{'${test.kafka.topic}'.split(',')}"
  public void consumingfrommultitopics(String data) {
        logger.info(String.format("consumingfromtwotopics -> %s", data));
    }

【讨论】：