apache-kafka - 只读取来自 kafka 主题的特定消息

Question

场景:

我在阅读时将数据 JSON 对象数据写入 kafka 主题我想根据消息中存在的值读取一组特定的消息。我正在使用 kafka-python 库。

示例消息:

{flow_status: "completed", value: 1, active: yes}
{flow_status:"failure",value 2, active:yes}

在这里，我只想读取 flow_Status 为已完成的消息。

Answer 1

在 Kafka 中，不可能做那样的事情。消费者一个接一个地消费消息，一个接一个地从最新提交的偏移量开始(或者从头开始，或者寻找特定的偏移量)。取决于您的用例，也许您的场景中可能会有不同的流程:执行流程的消息进入主题，然后处理操作的应用程序，然后将结果(完成或失败)写入两个不同的主题: 这样你就完成了所有的失败。另一种方法是使用 Kafka Streams 应用程序进行过滤，但考虑到它只是一个糖，实际上，流应用程序将始终读取所有消息，但允许您轻松过滤消息。