Scala Slick：永无止境的流答案

【问题标题】：Scala Slick: Never ending streamScala Slick：永无止境的流
【发布时间】：2016-03-20 07:53:10
【问题描述】：

使用 Slick，您可以执行以下操作以从表中生成结果流：

val q = for (e <- events) yield e.name
val p: DatabasePublisher[String] = db.stream(q.result)

p.foreach { s => println(s"Event: $s") }

这将打印events 表中的所有事件并在最后一行之后终止。

假设当新行进入events 表时，您可以通过某种方式得到通知，是否可以编写一个流，在插入事件时连续输出事件？一种tail -f 用于数据库表。

我认为 Slick 本身不会支持这一点，但我认为应该可以使用 Akka 流来提供帮助。因此，如果您可以从 Slick Source 获取某些内容直到它为空，然后等待事件指示表中的更多数据，然后流式传输新数据。可能通过使用ActorPublisher 来绑定此逻辑？

只是想知道是否有人在这方面有任何经验或任何建议？

【问题讨论】：

这感觉有点像 Spark 的 Dstream：spark.apache.org/docs/latest/api/scala/…

标签： scala slick akka-stream reactive-streams

【解决方案1】：

您对ActorPublisher 的看法是正确的 :) 这是一个使用 PostgreSQL、async DB driver 和 LISTEN/NOTIFY mechanism 的简单示例：

演员：

class PostgresListener extends ActorPublisher[String] {

  override def receive = {
    case _ ⇒
      val configuration = URLParser.parse(s"jdbc://postgresql://$host:$port/$db?user=$user&password=$password")
      val connection = new PostgreSQLConnection(configuration)
      Await.result(connection.connect, 5.seconds)

      connection.sendQuery(s"LISTEN $channel")
      connection.registerNotifyListener { message ⇒ onNext(message.payload) }
  }
}

服务：

def stream: Source[ServerSentEvent, Unit] = {
  val dataPublisherRef = Props[PostgresListener]
  val dataPublisher = ActorPublisher[String](dataPublisherRef)

  dataPublisherRef ! "go"

  Source(dataPublisher)
    .map(ServerSentEvent(_))
    .via(WithHeartbeats(10.second))
}

build.sbt in libraryDependencies:

"com.github.mauricio"  %% "postgresql-async"         % "0.2.18"

Postgres 触发器应该调用select pg_notify('foo', 'payload')

据我所知，Slick 不支持LISTEN。

【讨论】：

我现在有了一些工作，与您的回答略有不同，但使用 ActorPublisher 非常感谢您的确认。另外我认为您的答案是使用来自 akka-sse 项目（ServerSentEvent/WithHeartbeats）的代码？这对于这个问题不是必需的，但恰好是我正在寻找的东西。所以也感谢你的指点！
是的，你是对的，那是 Akka，我通过 SSE 将数据从 DB 流式传输到 UI。复制粘贴代码时忘记了这一点：）不客气！

【解决方案2】：

ActorPublisher 自 Akka 2.5.0 以来已被弃用。这是使用postgresql-async 库并在actor 内部创建SourceQueue 的替代方法：

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

import com.github.mauricio.async.db.postgresql.PostgreSQLConnection
import com.github.mauricio.async.db.postgresql.util.URLParser

import scala.concurrent.duration._
import scala.concurrent.Await

class DbActor(implicit materializer: ActorMaterializer) extends Actor with ActorLogging {
  private implicit val ec = context.system.dispatcher

  val queue =  
    Source.queue[String](Int.MaxValue, OverflowStrategy.backpressure)
      .to(Sink.foreach(println))
      .run()

  val configuration = URLParser.parse("jdbc:postgresql://localhost:5233/my_db?user=dbuser&password=pwd")
  val connection = new PostgreSQLConnection(configuration)
  Await.result(connection.connect, 5 seconds)

  connection.sendQuery("LISTEN my_channel")
  connection.registerNotifyListener { message =>
    val msg = message.payload
    log.debug("Sending the payload: {}", msg)
    self ! msg
  }

  def receive = {
    case payload: String =>
      queue.offer(payload).pipeTo(self)
    case QueueOfferResult.Dropped =>
      log.warning("Dropped a message.")
    case QueueOfferResult.Enqueued =>
      log.debug("Enqueued a message.")
    case QueueOfferResult.Failure(t) =>
      log.error("Stream failed: {}", t.getMessage)
    case QueueOfferResult.QueueClosed =>
      log.debug("Stream closed.")
  }
}

上面的代码只是打印来自 PostgreSQL 的通知，因为它们发生了；您可以将Sink.foreach(println) 替换为另一个Sink。运行它：

import akka.actor._
import akka.stream.ActorMaterializer

object Example extends App {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()
  system.actorOf(Props(classOf[DbActor], materializer))
}

【讨论】：