签名
最好通过签名来突出区别: Flow.map
接受一个返回类型为T
的函数,而 Flow.mapAsync
接受一个返回类型为Future[T]
的函数。
实际示例
例如,假设我们有一个函数,该函数根据用户ID查询数据库以获取用户的全名:
type UserID = String
type FullName = String
val databaseLookup : UserID => FullName = ??? //implementation unimportant
假设有一个 akka 流的 Source
,其中包含 UserID
值,我们可以在流中使用 Flow.map
查询数据库并将全名打印到控制台:
val userIDSource : Source[UserID, _] = ???
val stream =
userIDSource.via(Flow[UserID].map(databaseLookup))
.to(Sink.foreach[FullName](println))
.run()
这种方法的一个限制是,这个流一次只能进行1个数据库查询。这种串行查询会成为瓶颈,并可能阻止流的最大吞吐量。
我们可以尝试通过使用 "Future" 来进行并发查询以提高性能:
def concurrentDBLookup(userID : UserID) : Future[FullName] =
Future { databaseLookup(userID) }
val concurrentStream =
userIDSource.via(Flow[UserID].map(concurrentDBLookup))
.to(Sink.foreach[Future[FullName]](_ foreach println))
.run()
这个简单的补充说明存在问题,因为我们有效地消除了背压。
Sink只是拉取Future并添加一个“foreach println”,与数据库查询相比相对较快。流将持续传播需求到源并在Flow.map内产生更多的Futures。因此,没有限制同时运行的databaseLookup数量。无限制的并行查询最终可能会超载数据库。
Flow.mapAsync来拯救,我们可以在保持同时访问db的情况下限制同时查找的数量。
val maxLookupCount = 10
val maxLookupConcurrentStream =
userIDSource.via(Flow[UserID].mapAsync(maxLookupCount)(concurrentDBLookup))
.to(Sink.foreach[FullName](println))
.run()
同时请注意,
Sink.foreach
变得更加简单了,它不再需要一个
Future[FullName]
,而是直接使用
FullName
了。
无序异步 Map
如果维护 UserID 到 FullName 的顺序不必要,那么可以使用
Flow.mapAsyncUnordered
。例如:你只需要将所有名称打印到控制台中,但并不关心它们打印的顺序。
mapAsync
和将异步边界应用于特定阶段是否类似?根据文档,标记异步边界将在一个 actor 中运行每个阶段,只是想知道它们是否相同。 - druuu"com.typesafe.akka" %% "akka-stream" % "2.6.3"
,尝试了这个例子,但出现了编译错误type mismatch; found : akka.stream.scaladsl.Flow[Boolean,Boolean,akka.NotUsed] required: akka.stream.Graph[akka.stream.FlowShape[String,?],?]
。如果有人能将此示例更新到最新版本的 Akka Streams,我将不胜感激。 - radumanolescumapAsync
不接受ExecutionContext
,我认为它使用的是与图形本身相同的调度程序。但是,我还没有阅读代码来确认这一点。 - Ramón J Romero y Vigil