问Kafka Streams DSL over Kafka Consumer API

EN

通常情况下，这取决于用例，什么时候使用KafkaStreams，什么时候使用普通的KafkaStreams/消费者。一般而言，我不敢选择其中一个。

首先，KafkaStreams是建立在KafkaProducers/Consumer之上的，所以使用KafkaStreams可能的任何事情对于普通的Consumer/Producers也是可能的。

我想说的是，与普通的消费者/生产者相比，KafkaStreams应用程序接口没有那么复杂，但也没有那么灵活。现在我们可以开始长时间的讨论，讨论什么是“更少”。

在开发Kafka Streams时，您可以直接应用filter、map、join或aggregate等方法，因为所有的消费和生产部分都是在幕后抽象出来的。

当您使用普通的消费者/生产者开发应用程序时，您需要考虑如何在subscribe、poll、send、flush等级别构建客户端。

如果你想拥有更低的复杂度(但灵活性也更低) Kafka，ksqldb是另一个选择，你可以选择构建你的应用程序。

以下是您可能更喜欢Kafka Streams而不是核心生产者/消费者API的一些场景：

1. 它允许您轻松构建复杂的处理流水线。所以。让我们假设(一个人为设计的示例)您有一个包含客户订单的主题，并且您希望根据交付城市过滤订单，并将其保存到DB表中以便于持久化，并将其保存到Elasticsearch索引中以获得快速搜索体验。在这样的场景中，您将消费来自源主题的消息，使用Streams DSL filter函数根据城市过滤掉不必要的订单，并将过滤数据存储到单独的Kafka主题(使用KStream.to()或KTable.to())，最后使用Kafka Connect，消息将存储到数据库表和Elasticsearch中。你也可以使用核心的生产者/消费者应用程序接口来做同样的事情，但这需要更多的代码。

1. 在数据处理管道中，您可以在同一事务中执行消费-处理-生产。因此，在上面的例子中，Kafka将确保从源主题到DB和Elasticsearch的只有一次的语义和事务。不会因为网络故障和重试而引入任何重复的消息。当您进行聚合时，此功能尤其有用，例如在单个产品级别进行订单计数。在这种情况下，重复总是会给出错误的结果。

1. 您还可以通过更低的延迟来丰富传入数据。让我们假设在上面的示例中，您希望使用存储的客户数据中的客户电子邮件地址来丰富订单数据。如果没有Kafka Streams，你会怎么做？您可能会为网络上的每个传入订单调用REST API，这肯定是一个影响吞吐量的代价高昂的操作。在这种情况下，您可能希望将所需的客户数据存储在压缩的Kafka主题中，并使用KTable或GlobalKTable将其加载到流应用程序中。现在，您只需在KTable中对客户电子邮件地址执行简单的本地查找。请注意，这里的KTable数据将存储在Kafka Streams附带的嵌入式RocksDB中，而且由于KTable由Kafka主题支持，因此流应用程序中的数据将持续实时更新。换句话说，不会有陈旧的数据。这本质上是物化视图模式的一个示例。

1. 假设你想连接两个不同的数据流。因此，在上面的示例中，您希望只处理成功付款的订单，并且付款数据来自另一个Kafka主题。现在，可能会出现付款延迟或付款事件先于订单事件的情况。在这种情况下，您可能需要执行一小时窗口连接。因此，如果订单和相应的支付事件在一个小时的窗口内，订单将被允许沿着管道继续进行进一步处理。如您所见，您需要存储一个小时窗口的状态，该状态将存储在Kafka Streams的Rocks DB中。