为 kafka 消息实现归档

Question

我最近开始使用 Kafka，并针对一些用例评估了 Kafka。

如果我们想提供根据消息内容为消费者（订阅者）过滤消息的功能，那么最好的方法是什么？

假设生产者公开了一个名为“交易”的主题，该主题具有不同的交易详细信息，例如市场名称、创建日期、价格等。

一些消费者对特定市场的交易感兴趣，而另一些消费者对特定日期之后的交易感兴趣等（基于内容的过滤）

由于无法在代理端进行过滤，因此实现以下情况的最佳方法是：

1. 如果过滤条件特定于消费者。我们应该使用 消费者拦截器（尽管建议使用拦截器进行日志记录 根据文档的用途）？
2. 如果过滤标准（基于内容的过滤）在消费者中很常见，应该采用什么方法？

监听主题并在本地过滤消息并写入新主题（使用拦截器或流）

Answer 1

如果我正确理解您的问题，您有一个主题和不同的消费者，他们对该主题的特定部分感兴趣。同时，您不拥有这些消费者，想要避免那些消费者只是阅读整个主题并自己进行过滤？

为此，构建一个新应用程序的唯一方法是读取整个主题，进行过滤（或实际拆分）并将数据写回两个（多个）不同的主题。外部消费者将从这些新主题中消费，并且只收到他们感兴趣的日期。

为此目的使用 Kafka Streams 将是一个非常好的方法。 DSL 应该提供您需要的一切。

作为替代方案，您可以使用KafkaConsumer 和KafkaProducer 编写自己的应用程序，在您的用户代码中手动进行过滤/拆分。这与使用 Kafka Streams 没有太大区别，因为 Kafka Streams 应用程序会在内部执行完全相同的操作。但是，使用 Streams，您完成它的工作量会少很多。

我不会为此使用拦截器。即使这可行，它似乎也不是您用例的好软件设计。

Answer 2

在为消费者设置“interceptor.classes”配置之前，创建您自己的实现 org.apache.kafka.clients.consumer.ConsumerInterceptor 的拦截器类并在方法“onConsume”中实现您的逻辑。