是否可以使用 Akka Stream 从数据库表中创建“无限”流

Question

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

更新：2/23/16

我已经实施了@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。

更新：2/24/16

更新以尝试来自@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
  }
}

Answer 2

这是数据库无限流工作代码。这已经过测试，在流式应用程序运行时将数百万条记录插入到 postgresql 数据库中 -

package infinite.streams.db

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.alpakka.slick.scaladsl.SlickSession
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.stream.{ActorMaterializer, ThrottleMode}
import org.slf4j.LoggerFactory
import slick.basic.DatabaseConfig
import slick.jdbc.JdbcProfile
import scala.concurrent.duration._
import scala.concurrent.{Await, ExecutionContextExecutor}

case class Record(id: Int, value: String) {
  val content = s"<ROW><ID>$id</ID><VALUE>$value</VALUE></ROW>"
}

object InfiniteStreamingApp extends App {

  println("Starting app...")

  implicit val system: ActorSystem = ActorSystem("Publisher")
  implicit val ec: ExecutionContextExecutor = system.dispatcher
  implicit val materializer: ActorMaterializer = ActorMaterializer()

  println("Initializing database configuration...")
  val databaseConfig: DatabaseConfig[JdbcProfile] = DatabaseConfig.forConfig[JdbcProfile]("postgres3")
  implicit val session: SlickSession = SlickSession.forConfig(databaseConfig)

  import databaseConfig.profile.api._

  class Records(tag: Tag) extends Table[Record](tag, "test2") {
    def id = column[Int]("c1")

    def value = column[String]("c2")

    def * = (id, value) <> (Record.tupled, Record.unapply)
  }

  val db = databaseConfig.db

  println("Prime for streaming...")

  val logic: Flow[(Int, String), (Int, String), NotUsed] = Flow[(Int, String)].map {
    case (id, value) => (id, value.toUpperCase)
  }

  val fetchSize = 5
  try {
    val done = Source
      .unfoldAsync(0) {
      lastId =>
        println(s"Fetching next: $fetchSize records with id > $lastId")
        val query = TableQuery[Records].filter(_.id > lastId).take(fetchSize)
        db.run(query.result.withPinnedSession)
          .map {
            recs => Some(recs.last.id, recs)
          }
    }
      .throttle(5, 1.second, 1, ThrottleMode.shaping)
      .flatMapConcat {
        recs => Source.fromIterator(() => recs.iterator)
      }
      .map(x => (x.id, x.content))
      .via(logic)
      .log("*******Post Transformation******")
//      .runWith(Sink.foreach(r => println("SINK: " + r._2)))
// Use runForeach or runWith(Sink)
      .runForeach(rec => println("REC: " + rec))

    println("Waiting for result....")
    Await.ready(done, Duration.Inf)

  } catch {
    case ex: Throwable => println(ex.getMessage)
  } finally {
    println("Streaming end successfully")
    db.close()
    system.terminate()
  }

}



application.conf

akka {
  loggers = ["akka.event.slf4j.Slf4jLogger"]
  loglevel = "INFO"
}
# Load using SlickSession.forConfig("slick-postgres")
postgres3 {
  profile = "slick.jdbc.PostgresProfile$"
  db {
    dataSourceClass = "slick.jdbc.DriverDataSource"
    properties = {
      driver = "org.postgresql.Driver"
      url = "jdbc:postgresql://localhost/testdb"
      user = "postgres"
      password = "postgres"
    }
    numThreads = 2
  }
}