Как использовать сгруппированные подпотоки с помощью mapAsync в потоках akka

Расширение комментария Джрудольфа, который полностью правильный ...

В этом случае вам не нужен 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 включает ожидание результатов, что не является реактивный ...