apache-nifi - 在 NiFi 上如何区分来自两个不同 Kafka 的消息？

Question

我有两个不同的应用程序，名为 A1 和 A2。每个应用程序都有自己的 Kakfa 服务器。来自这两个 KAKFA 服务器(代理)的消息将发送到 NiFi。

每个 Kakfa 都有不同的主题名称，基于此我可以区分来自 Kakfa 的消息。但是除了 Kafka 的主题名称之外，NiFi 中还有其他方法可以区分来自两个不同 Kakfa 的消息吗？他们是否有任何 NiFi 处理器检查主题名称，然后决定下一条路线？

非常感谢。问候，是的

Answer 1

如果您使用 NiFi 的 Kafka 处理器 (ConsumeKafka/ConsumeKafkaRecord) 从 Kafka 接收消息，它们将以 FlowFiles 的形式输出消息。它们带有一个名为 kafka.topic 的属性，该属性将包含消息来自的主题的名称。

要根据主题名称路由消息，您可以使用 RouteOnAttribute 处理器。例如，您有两个主题 topicA 和 topicB。然后你必须像这样配置 RouteOnAttribute 处理器:

然后根据您的要求连接关系 topic-a 和 topic-b 以分离流。如果您要添加更多 Kafka 源，您所要做的就是使用一种动态关系更新 RouteOnAttribute。例如:topic-c : ${kafka.topic:equals('topicC')}