scala - 使用 Akka Streams 2.4.2 和 Slick 3.0 从 postgres 读取

Question

尝试新创建的 Akka Streams。除了一件小事之外，它似乎正在工作 - 没有输出。

我有下表定义:

case class my_stream(id: Int, value: String)

class Streams(tag: Tag) extends Table[my_stream](tag, "my_stream") {
  def id = column[Int]("id")
  def value = column[String]("value")
  def * = (id, value) <> (my_stream.tupled, my_stream.unapply)
}

我试图像这样将表的内容输出到标准输出:

def main(args: Array[String]) : Unit = {
  implicit val system = ActorSystem("Subscriber")
  implicit val materializer = ActorMaterializer()

  val strm = TableQuery[Streams]
  val db = Database.forConfig("pg-postgres")

  try{
    var src = Source.fromPublisher(db.stream(strm.result))
    src.runForeach(r => println(s"${r.id},${r.value}"))(materializer)
  } finally {
    system.shutdown
    db.close
  }
}

我已经通过配置调试日志来验证查询正在运行。然而，我得到的只是这个:

08:59:24.099 [main] INFO  com.zaxxer.hikari.HikariDataSource - pg-postgres - is starting.
08:59:24.428 [main] INFO  com.zaxxer.hikari.pool.HikariPool - pg-postgres - is closing down.

Answer 1

原因是 Akka Streams 是异步的和 runForeach返回将在流完成后完成的 Future，但未处理该 Future，因此 system.shutdown和 db.close立即执行而不是在流完成后执行。