scala - 如何在 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)

我可以在流中使用 future 值进行操作，还是有更好的方法来处理异步任务(例如 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 主题，然后您将其带入您的 Kafka Streams 应用程序。对于异步调用的并行吞吐量比端到端延迟(例如 Web 服务调用或对 NoSQL 数据库的查询)快得多的情况，我们就是这样做的。

将您的所有扩充数据发布到其自己的 KTable 并将其加入到 Kafka Streams 应用程序中。事实上，Kafka Streams 擅长通过 KTables 将流数据与丰富数据连接起来。如果丰富数据可以表示为表格，我们就使用它。如果必须动态计算富集数据，则它不起作用。