kafka 流 - 在源不是 kafka 主题的情况下，我可以使用 kafka 流处理吗？答案

【问题标题】：kafka streams - can I use kafka streams processing in cases where the source is not a kafka topic?kafka 流 - 在源不是 kafka 主题的情况下，我可以使用 kafka 流处理吗？
【发布时间】：2022-06-16 20:40:03
【问题描述】：

我有一个应用程序（称为 smscb-router），如图所示。

它从旧系统 (sms) 读取数据。
根据内容（回调类型），我必须放入相应的传出主题（例如billing-n-cdr，dr-cdr，...）

我认为流 API 更适合这种情况，因为它具有执行内容映射检查的地图功能。我不确定的是，我可以从非 kafka-topic 源中读取源数据吗？

我在互联网博客上看到的所有示例，都以从源主题读取并放入其他目标主题的上下文来解释蒸汽应用程序。

那么，这是否可以从非主题源读取，例如 redis 存储或消息队列，例如 RabbitMQ？

【问题讨论】：

标签： apache-kafka-streams

【解决方案1】：

我们最近有一个实现，我们必须从网络连接的驱动器轮询 .xml 文件并将其转换为 KAFKA 事件，即将每条记录发布到输出主题中。在这种情况下，我们甚至不会称它为我们使用 Streams API 开发的东西，但它只是一个 KAFKA 生产者组件。

Java File Poller Module (Quartz time based) -> XML Schema Management -> KAFKA Producer Component -> Output Topic (KAFKA Broker)。

您将获得 KAKFA Producer API 在重试方面的所有原生功能，并且您可以使用 producer.send (Sync) 或 producer.send.get(Asyn) 进行回调。

希望这会有所帮助。 Streams API 适用于需要通过使用有状态操作进行规范化的大型且非常复杂的事情。

谢谢，克里斯托弗

【讨论】：

这有帮助。但我正在研究聚合和过滤/映射等，我将不得不使用代码，而不是声明方式。
那么更好的选择是将您的遗留数据转换为 KAFKA 事件，与我提到的选项对齐并将其写入中间主题（如果您有创建一个控件）并使用它中间主题作为您使用 Stream API 进行进一步有状态操作的源主题。所以你的流程是 Legacy System data -> Producer API -> Intermittent Topic -> Stream API.. 你也会得到更好的控制。
是的，我认为这将是一个选项。

【解决方案2】：

Kafka Streams 只是关于主题到主题的数据流

所有外部系统都应该通过另一种方法集成：

理想的 Kafka Connect ：例如这个： https://docs.confluent.io/kafka-connect-rabbitmq-source/current/overview.html

您也可以在第一步使用手动消费者，但最好重用 Kafka Connect 中内置的所有可用性机制。（没有代码，只是一些 Json 配置）。

【讨论】：