如何同时创建一个组件 - 一个 kafka 消费者和一个 kafka 生产者？答案

【问题标题】：How do I create a component both - A kafka consumer and a kafka Producer?如何同时创建一个组件 - 一个 kafka 消费者和一个 kafka 生产者？
【发布时间】：2020-08-26 05:35:44
【问题描述】：

我正在尝试创建一个组件，该组件使用来自一个主题的数据，对其进行处理并将其发送到另一个主题，即，我需要使我的组件同时成为消费者和生产者。如何在我的 Spring Boot 应用程序中配置它？

【问题讨论】：

网上找教程就行了……

标签： apache-kafka kafka-consumer-api apache-kafka-streams spring-kafka kafka-producer-api

【解决方案1】：

听起来您正在寻找的是 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()));

【讨论】：

谢谢约瑟夫。是的，我正在使用 Kafka 流。我只是无法理解如何提供生产者/消费者配置。如，我们如何使用相同的属性对象为两者设置配置？
如果您需要指定不同的 Serde 作为消费者和生产者，您可以在构建 KStream（针对消费者）和发布到输出主题（针对生产者）时指定。让我更新我的答案来举个例子