如何在akka流中使用mapAsync分组子流

Question

我的问题是，我有一个未知的组数，如果mapAsync的并行数少于我得到的组数，并且在最后一个接收器中出错

由于上游错误（Akka.Stream.Impl.StreamSubscriptionTimeoutSupport$$Anon$2),正在拆除SynchronousFileSink（/users/sam/dev/projects/akka-streams/target/log-error.txt）

我尝试按照akka流的模式指南http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-cookbook.html中的建议在中间放置一个缓冲区

groupBy {
  case LoglevelPattern(level) => level
  case other                  => "OTHER"
}.buffer(1000, OverflowStrategy.backpressure).
  // write lines of each group to a separate file
  mapAsync(parallelism = 2) {....

Answer 1

对Jrudolph完全正确的评论进行了扩展。

在此实例中不需要mapasync。作为一个基本示例，假设您有一个元组的源

import akka.stream.scaladsl.{Source, Sink}

def data() = List(("foo", 1),
                  ("foo", 2),
                  ("bar", 1),
                  ("foo", 3),
                  ("bar", 2))

val originalSource = Source(data)

然后可以执行groupBy来创建源的源

def getID(tuple : (String, Int)) = tuple._1

//a Source of (String, Source[(String, Int),_])
val groupedSource = originalSource groupBy getID

每一个分组的源都可以通过一个map并行处理，不需要任何花哨的东西。下面是一个示例,每个分组都在一个独立的流中求和：

import akka.actor.ActorSystem
import akka.stream.ACtorMaterializer

implicit val actorSystem = ActorSystem()
implicit val mat = ActorMaterializer()
import actorSystem.dispatcher

def getValues(tuple : (String, Int)) = tuple._2

//does not have to be a def, we can re-use the same sink over-and-over
val sumSink = Sink.fold[Int,Int](0)(_ + _)

//a Source of (String, Future[Int])
val sumSource  = 
  groupedSource map { case (id, src) => 
    id -> {src map getValues runWith sumSink} //calculate sum in independent stream
  }

现在，所有“foo”数字与所有“bar”数字并行求和。

MapAsync用于以下情况：您有一个封装函数返回future[T],而您试图发出T；你的问题不是这样的。此外，mapAsync还需要等待结果，这不是反应性的...