如何同时创建一个组件 - 一个 kafka 消费者和一个 kafka 生产者？
How do I create a component both - A kafka consumer and a kafka Producer?
我正在尝试创建一个组件，该组件使用来自一个主题的数据，对其进行处理并发送到另一个主题，即，我需要让我的组件既是消费者又是生产者。
如何在我的 Spring 启动应用程序中配置它？
听起来您正在寻找的是 Kafka Streams API。
它是一个开源 Java API，用于通过读取一个主题、运行 通过处理步骤和写入另一个主题来操纵飞行中的事件。
因此它既是生产者又是消费者。
有关示例，请参阅 Kafka Streams documentation。
为生产者和消费者设置不同的 Serde 配置：
KStream<String, String> wordCountInputStream = streamsBuilder.stream("word-count-input", Consumed.with(Serdes.String(), Serdes.String()));
KTable<String, Long> wordCounts = wordCountInputStream.mapValues(value -> value.toLowerCase()).....
wordCounts.toStream().to("word-count-output", Produced.with(Serdes.String(), Serdes.Long()));
我正在尝试创建一个组件，该组件使用来自一个主题的数据，对其进行处理并发送到另一个主题，即，我需要让我的组件既是消费者又是生产者。 如何在我的 Spring 启动应用程序中配置它？
听起来您正在寻找的是 Kafka Streams API。 它是一个开源 Java API，用于通过读取一个主题、运行 通过处理步骤和写入另一个主题来操纵飞行中的事件。 因此它既是生产者又是消费者。 有关示例，请参阅 Kafka Streams documentation。
为生产者和消费者设置不同的 Serde 配置：
KStream<String, String> wordCountInputStream = streamsBuilder.stream("word-count-input", Consumed.with(Serdes.String(), Serdes.String()));
KTable<String, Long> wordCounts = wordCountInputStream.mapValues(value -> value.toLowerCase()).....
wordCounts.toStream().to("word-count-output", Produced.with(Serdes.String(), Serdes.Long()));