小编Art*_*hka的帖子

有一个 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）？