Scala Akka Stream：如何通过 Seq

Question

我试图在 Future 中包装一些阻塞调用。return 类型是 Seq[User]，其中 User 是 case class。以下只是不会编译存在各种重载版本的投诉。有什么建议么？我试过几乎所有的变体都是 Source.apply 没有任何运气。

// All I want is Seq[User] => Future[Seq[User]]

def findByFirstName(firstName: String) = {
  val users: Seq[User] = userRepository.findByFirstName(firstName)

  val sink = Sink.fold[User, User](null)((_, elem) => elem)

  val src = Source(users) // doesn't compile

  src.runWith(sink)
}

Answer 1

如果您只需要 Future[Seq[Users]]，请不要使用 akka 流，但是 futures

import scala.concurrent._
import ExecutionContext.Implicits.global
val session = socialNetwork.createSessionFor("user", credentials)
val f: Future[List[Friend]] = Future {
  session.getFriends()
}

Answer 2

首先，我假设您使用的是 akka-http-experimental 的 1.0 版，因为 API 可能与之前的版本有所不同。

您的代码无法编译的原因是 akka.stream.scaladsl.Source$.apply() 需要 scala.collection.immutable.Seq 而不是 scala.collection.mutable.Seq.

因此您必须使用 to[T] 方法从可变序列转换为不可变序列。

文档：akka.stream.scaladsl.Source

此外，如您所见，Source$.apply() 接受 ()=>Iterator[T] 因此您也可以将 ()=>users.iterator 作为参数传递。

因为 Sink.fold(...) returns 最后一个求值表达式，你可以给一个空的 Seq() 作为第一个参数，迭代 users 并将元素附加到顺序，最后得到结果。

然而，可能有更好的解决方案可以创建一个 Sink 将每个计算的表达式放入 Seq，但我找不到它。

以下代码有效。

import akka.actor._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Source,Sink}
import scala.concurrent.ExecutionContext.Implicits.global

case class User(name:String)

object Main extends App{
    implicit val system = ActorSystem("MyActorSystem")
    implicit val materializer = ActorMaterializer()
    val users = Seq(User("alice"),User("bob"),User("charlie"))

    val sink = Sink.fold[Seq[User], User](Seq())(
      (seq, elem) => 
        {println(s"elem => ${elem} \t| seq => ${seq}");seq:+elem})

    val src = Source(users.to[scala.collection.immutable.Seq])
    // val src = Source(()=>users.iterator) // this also works

    val fut = src.runWith(sink) // Future[Seq[User]]
    fut.onSuccess({
      case x=>{
        println(s"result => ${x}")
      }
    })
}

上面代码的输出是

elem => User(alice)     | seq => List()
elem => User(bob)       | seq => List(User(alice))
elem => User(charlie)   | seq => List(User(alice), User(bob))
result => List(User(alice), User(bob), User(charlie))

Scala Akka Stream：如何通过 Seq

Scala Akka Stream: How to Pass Through a Seq

scala

akka

reactive-streams

akka-stream