如何将一些记录从TableA复制到TableB中，使用流畅的流和akka流

Question

有两个表TableA和TableB。

我需要将一些记录从TableA复制到TableB。我使用slick-3.0并使用以下方式：

import akka.stream._
import akka.stream.scaladsl._
...

//{{ READ DATA FROM TABLE A
val q = TableA.filter(somePredicate).result
val source = Source.fromPublisher {
      db.stream(q.result).mapResult { r => 
        val record: RecordA = someTransformation(r) 
        record
      }
    }.grouped(50) // grouping because I want to write records in batch mode
//}}

//{{ WRITE DATA TO TABLE B
val f:Future[Done] = source.runWith(Sink.foreach { batch: Seq[RecordA] =>
      //TODO how to write batch to TableB asynchronously?
      val insertAction = TableB ++= batch  // insert batch to table
      val fInsert: Future[_] = db.run(insertAction)
      Await.result(fInsert, ...)           // #1 this works only with blocking
})
//}}

但是我遇到了一个问题--如何异步地将批处理写入TableB (参见TODO)。现在，上面的代码只用于阻塞内部未来(参见#1注释)。是否有一种异步实现该任务的正确方法？

Answer 1

使用mapAsync，它期望返回一个未来，并在下一个阶段公开“未包装”结果。

source.mapAsync(4){batch: Seq[RecordA] =>
      val insertAction = TableB ++= batch  // insert batch to table
      db.run(insertAction)
}).to(Sink.ignore).run