slick-3.0 - 是否可以使用 Akka Stream 从数据库表创建 "infinite"流

Question

我正在玩 Akka Streams 2.4.2，我想知道是否有可能设置一个使用数据库表作为源的流，并且只要有一条记录添加到表中，该记录就会具体化并推送到下游？

更新:2016 年 2 月 23 日

我已经实现了@PH88 的解决方案。这是我的表定义:

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

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

实现如下:

 implicit val system = ActorSystem("Publisher")
 implicit val materializer = ActorMaterializer()
 val db = Database.forConfig("pg-postgres")

 try{
  val newRecStream = Source.unfold((0, List[Record]())) { n =>
    try {
      val q = for (r <- TableQuery[Records].filter(row => row.id > n._1)) yield (r)
      val r = Source.fromPublisher(db.stream(q.result)).collect {
        case rec => println(s"${rec.id}, ${rec.value}"); rec
      }.runFold((n._1, List[Record]())) {
        case ((id, xs), current) => (current.id, current :: xs)
      }

      val answer: (Int, List[Record]) = Await.result(r, 5.seconds)
      Option(answer, None)
    }
    catch { case e:Exception => println(e); Option(n, e) }
  }


   Await.ready(newRecStream.throttle(1, 1.second, 1, ThrottleMode.shaping).runForeach(_ => ()), Duration.Inf)
 }
 finally {
   system.shutdown
   db.close
 }

但我的问题是，当我尝试调用 flatMapConcat 时，我得到的类型是 Serializable。

更新:2016 年 2 月 24 日

已更新以尝试来自@PH88 的db.run 建议:

implicit val system = ActorSystem("Publisher")
implicit val materializer = ActorMaterializer()
val db = Database.forConfig("pg-postgres")
val disableAutoCommit = SimpleDBIO(_.connection.setAutoCommit(false))
val queryLimit = 1

try {
 val newRecStream = Source.unfoldAsync(0) { n =>
     val q = TableQuery[Records].filter(row => row.id > n).take(queryLimit)
     db.run(q.result).map { recs =>
       Some(recs.last.id, recs)
     }
   }
   .throttle(1, 1.second, 1, ThrottleMode.shaping)
   .flatMapConcat { recs =>
      Source.fromIterator(() => recs.iterator)
   }
   .runForeach { rec =>
       println(s"${rec.id}, ${rec.value}")
   }

   Await.ready(newRecStream, Duration.Inf)
 }
 catch
 {
   case ex: Throwable => println(ex)
 }
 finally {
   system.shutdown
   db.close
 }

这行得通(我将查询限制更改为 1，因为目前我的数据库表中只有几个项目)- 除非它打印程序存在的表中的最后一行。这是我的日志输出:

17:09:27,982 |-INFO in ch.qos.logback.classic.LoggerContext[default] - Could NOT find resource [logback.groovy]
17:09:27,982 |-INFO in ch.qos.logback.classic.LoggerContext[default] - Could NOT find resource [logback-test.xml]
17:09:27,982 |-INFO in ch.qos.logback.classic.LoggerContext[default] - Found resource [logback.xml] at [file:/Users/xxxxxxx/dev/src/scratch/scala/fpp-in-scala/target/scala-2.11/classes/logback.xml]
17:09:28,062 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.ConsoleAppender]
17:09:28,064 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [STDOUT]
17:09:28,079 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property
17:09:28,102 |-INFO in ch.qos.logback.classic.joran.action.LoggerAction - Setting level of logger [application] to DEBUG
17:09:28,103 |-INFO in ch.qos.logback.classic.joran.action.RootLoggerAction - Setting level of ROOT logger to INFO
17:09:28,103 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [STDOUT] to Logger[ROOT]
17:09:28,103 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - End of configuration.
17:09:28,104 |-INFO in ch.qos.logback.classic.joran.JoranConfigurator@4278284b - Registering current configuration as safe fallback point
17:09:28.117 [main] INFO  com.zaxxer.hikari.HikariDataSource - pg-postgres - is starting.
1, WASSSAAAAAAAP!
2, WHAAAAT?!?
3, booyah!
4, what!
5, This rocks!
6, Again!
7, Again!2
8, I love this!
9, Akka Streams rock
10, Tuning jdbc
17:09:39.000 [main] INFO  com.zaxxer.hikari.pool.HikariPool - pg-postgres - is closing down.

Process finished with exit code 0

找到丢失的部分 - 需要更换它:

Some(recs.last.id, recs)

用这个:

 val lastId = if(recs.isEmpty) n else recs.last.id
 Some(lastId, recs)

当结果集为空时，对 recs.last.id 的调用抛出 java.lang.UnsupportedOperationException: empty.last。

Answer 1

一般来说，SQL 数据库是一种“被动”构造，不会像您描述的那样主动推送更改。您只能通过定期轮询来“模拟”“推送”，例如:

val newRecStream = Source

  // Query for table changes
  .unfold(initState) { lastState =>
    // query for new data since lastState and save the current state into newState...
    Some((newState, newRecords))
  }

  // Throttle to limit the poll frequency
  .throttle(...)  

  // breaks down into individual records...
  .flatMapConcat { newRecords =>
    Source.unfold(newRecords) { pendingRecords =>
      if (records is empty) {
        None
      } else {
        // take one record from pendingRecords and save to newRec.  Save the rest into remainingRecords.
        Some(remainingRecords, newRec)
      }
    }
  }

更新时间:2016 年 2 月 24 日

基于问题 2/23/2016 更新的伪代码示例:

implicit val system = ActorSystem("Publisher")
implicit val materializer = ActorMaterializer()
val db = Database.forConfig("pg-postgres")
val queryLimit = 10
try {
  val completion = Source
    .unfoldAsync(0) { lastRowId =>
      val q = TableQuery[Records].filter(row => row.id > lastRowId).take(queryLimit)
      db.run(q.result).map { recs =>
        Some(recs.last.id, recs)
      }
    }
    .throttle(1, 1.second, 1, ThrottleMode.shaping)
    .flatMapConcat { recs =>
      Source.fromIterator(() => recs.iterator)
    }
    .runForeach { rec =>
      println(s"${rec.id}, ${rec.value}")
    }

  // Block forever
  Await.ready(completion, Duration.Inf)

} catch {
  case ex: Throwable => println(ex)
} finally {
  system.shutdown
  db.close
}

它将在 unfoldAsync 中重复执行查询 DB，一次最多检索 10 (queryLimit) 条记录并将记录发送到下游 (-> throttle -> flatMapConcat -> runForeach)。最后的 Await 实际上会永远阻塞。

更新时间:2016 年 2 月 25 日

可执行的“概念验证”代码:

import akka.actor.ActorSystem
import akka.stream.{ThrottleMode, ActorMaterializer}
import akka.stream.scaladsl.Source
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

object Infinite extends App{
  implicit val system = ActorSystem("Publisher")
  implicit val ec = system.dispatcher
  implicit val materializer = ActorMaterializer()
  case class Record(id: Int, value: String)
  try {
    val completion = Source
      .unfoldAsync(0) { lastRowId =>
        Future {
          val recs = (lastRowId to lastRowId + 10).map(i => Record(i, s"rec#$i"))
          Some(recs.last.id, recs)
        }
      }
      .throttle(1, 1.second, 1, ThrottleMode.Shaping)
      .flatMapConcat { recs =>
        Source.fromIterator(() => recs.iterator)
      }
      .runForeach { rec =>
        println(rec)
      }

    Await.ready(completion, Duration.Inf)

  } catch {
    case ex: Throwable => println(ex)
  } finally {
    system.shutdown
  }
}