如何在 Kafka Streams 中使用 Futures

Question

有一个 kafka 集群，我从中消费两个主题并加入它。使用 join 的结果，我对数据库进行了一些操作。对 DB 的所有操作都是异步的，因此它们返回给我一个 Future（scala.concurrent.Future，但无论如何它与 java.util.concurrent.CompletableFuture 相同）。所以结果我得到了这样的代码：

val firstSource: KTable[String, Obj]
val secondSource: KTable[String, Obj2]

def enrich(data: ObjAndObj2): Future[EnrichedObj]
def saveResultToStorage(enrichedData: Future[EnrichedObj]): Future[Unit]

firstSource.leftJoin(secondSource, joinFunc)
           .mapValues(enrich)
           .foreach(saveResultToStorage)

我可以在流中使用未来值进行操作，还是有更好的方法来处理异步任务（例如 Akka 流中的 .mapAsync）？

Answer 1

我有同样的问题。据我所知，Kafka Streams 的设计目的不是像 Akka Streams 那样处理多速率流。Kafka Streams 没有 Akka 具有的多速率原语等价物，如 mapAsync、throttle、conflate、buffer、batch 等。Kafka Streams 擅长处理主题和有状态数据聚合之间的连接。Akka Streams 擅长多速率和异步处理。

您有几种选择来处理这个问题：

• 在 Kafka Streams 应用程序中进行阻塞调用。这是最简单的，如果您的 Future 调用的吞吐量不大于它们的延迟，则很好。Kafka Streams 为每个分区使用单独的线程，因此您可以使用正在处理的 Kafka 主题的分区来驱动并行性。
• 使用Reactive Kafka库处理 Akka Streams 中的扩充，将扩充的结果发布到另一个 Kafka Topic，然后将其引入您的 Kafka Streams 应用程序。对于异步调用的并行吞吐量比端到端延迟（例如 Web 服务调用或对 NoSQL 数据库的查询）快得多的情况，我们就是这样做的。
• 将您的所有扩充数据发布到其自己的 KTable 并将其加入到 Kafka Streams 应用程序中。事实上，Kafka Streams 擅长通过 KTables 将流数据与丰富数据连接起来。如果丰富数据可以表示为表格，我们就使用它。如果必须动态计算富集数据，则它不起作用。