Akka流:在图形阶段处理期货

Question

在akka流阶段FlowShape[A, B],我需要在A上进行的部分处理是使用使用A数据构建的查询来保存/查询数据存储.但是那个数据存储驱动程序查询给了我一个未来,我不知道如何最好地处理它(我的主要问题在这里).

case class Obj(a: String, b: Int, c: String)
case class Foo(myobject: Obj, name: String)
case class Bar(st: String)
//
class SaveAndGetId extends GraphStage[FlowShape[Foo, Bar]] {
 val dao = new DbDao // some dao with an async driver 

 override def createLogic(inheritedAttributes: Attributes) = new GraphStageLogic(shape) {
  setHandlers(in, out, new InHandler with Outhandler {
   override def onPush() = {
    val foo = grab(in)
    val add = foo.record.value()
    val result: Future[String] = dao.saveAndGetRecord(add.myobject)//saves and returns id as string

   //the naive approach
    val record = Await(result, Duration.inf)
    push(out, Bar(record))// ***tests pass every time

  //mapping the future approach
    result.map { x=>
     push(out, Bar(x))
    } //***tests fail every time

下一个阶段取决于查询返回的db记录的id,但我想避免Await.我不确定为什么映射方法失败:

"it should work" in {
  val source = Source.single(Foo(Obj("hello", 1, "world")))
  val probe = source
    .via(new SaveAndGetId))
    .runWith(TestSink.probe)
  probe
   .request(1)
   .expectBarwithId("one")//say we know this will be
   .expectComplete()
 }
 private implicit class RichTestProbe(probe: Probe[Bar]) {
  def expectBarwithId(expected: String): Probe[Bar] = 
   probe.expectNextChainingPF{
    case r @ Bar(str) if str == expected => r
  }
 }

当运行映射未来时,我会失败:

should work ***FAILED***
java.lang.AssertionError: assertion failed: expected: message matching partial function but got unexpected message OnComplete
at scala.Predef$.assert(Predef.scala:170)
at akka.testkit.TestKitBase$class.expectMsgPF(TestKit.scala:406)
at akka.testkit.TestKit.expectMsgPF(TestKit.scala:814)
at akka.stream.testkit.TestSubscriber$ManualProbe.expectEventPF(StreamTestKit.scala:570)

文档中的异步侧通道示例在舞台的构造函数中具有未来,而不是在舞台中构建未来,因此似乎不适用于我的情况.

Answer 1

我同意拉蒙的观点.FlowShape在这种情况下,构建一个新的并不是必需的,它太复杂了.mapAsync这里使用方法非常方便:

以下是要使用的代码段mapAsync:

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

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

object MapAsyncExample {

  val numOfParallelism: Int = 10

  def main(args: Array[String]): Unit = {
    Source.repeat(5)  
      .mapAsync(parallelism)(x => asyncSquare(x))           
      .runWith(Sink.foreach(println)) previous stage
  }

  //This method returns a Future
  //You can replace this part with your database operations
  def asyncSquare(value: Int): Future[Int] = Future {
    value * value
  }
}

在上面的代码片段中,Source.repeat(5)是一个5无限发射的虚拟源.有一个样本函数asyncSquare,它取a integer并计算其中的平方Future..mapAsync(parallelism)(x => asyncSquare(x))line使用该函数并将输出Future发送到下一阶段.在这个snipet中,下一个阶段是sink打印每个项目.

parallelism是asyncSquare可以并发运行的最大呼叫数.

Answer 2

我认为你的GraphStage过于复杂了。下面Flow执行相同的操作，无需编写自定义阶段：

val dao = new DbDao

val parallelism = 10 //number of parallel db queries

val SaveAndGetId : Flow[Foo, Bar, _] = 
  Flow[Foo]
    .map(foo => foo.record.value().myobject)
    .mapAsync(parallelism)(rec => dao.saveAndGetRecord(rec))
    .map(Bar.apply)

我通常尝试将GraphStage其视为最后的手段，几乎总是有一种惯用的方法可以通过使用 akka-stream 库提供的方法来获得相同的 Flow。