使用 slick 的 3.0.0 流媒体结果和 Postgresql 的正确方法是什么？答案

【问题标题】：What is the right way to work with slick's 3.0.0 streaming results and Postgresql?使用 slick 的 3.0.0 流媒体结果和 Postgresql 的正确方法是什么？
【发布时间】：2015-09-29 04:29:54
【问题描述】：

我正在尝试弄清楚如何使用流畅的流媒体。我使用 slick 3.0.0 和 postgres 驱动程序

情况如下：服务器必须将数据序列分配给客户端，并按大小（以字节为单位）限制成块。所以，我写了以下精巧的查询：

val sequences = TableQuery[Sequences]
def find(userId: Long, timestamp: Long) = sequences.filter(s ⇒ s.userId === userId && s.timestamp > timestamp).sortBy(_.timestamp.asc).result
val seq = db.stream(find(0L, 0L))

我将 seq 与 akka-streams Source 组合在一起，编写了自定义 PushPullStage，它限制了数据的大小（以字节为单位）并在达到大小限制时完成上游。它工作得很好。问题是 - 当我查看 postgres 日志时，我看到这样的查询 select * from sequences where user_id = 0 and timestamp > 0 order by timestamp;

因此，乍一看，似乎有很多（而且不必要的）数据库查询正在进行，只是在每个查询中使用了几个字节。使用 Slick 进行流式传输以最小化数据库查询并充分利用每个查询中传输的数据的正确方法是什么？

【问题讨论】：

标签： postgresql scala slick slick-3.0 akka-stream

【解决方案1】：

使用 Slick 和 Postgres 进行流式传输的“正确方法”包括三件事：

必须使用 db.stream()
必须在 JDBC 驱动程序中禁用 autoCommit。一种方法是通过后缀.transactionally 使查询在事务中运行。
必须将 fetchSize 设置为 0 以外的值，否则 postgres 会将整个结果集一次性推送到客户端。

例如：

DB.stream(
  find(0L, 0L)
    .transactionally
    .withStatementParameters(fetchSize = 1000)
).foreach(println)

有用的链接：

https://github.com/slick/slick/issues/1038

https://github.com/slick/slick/issues/809

【讨论】：

感谢您的回答，非常有帮助。我对“附录”感到困惑：处理 db 的执行上下文不是在 AsyncExecutor 中单独维护吗？
很高兴听到！关于附录：是的，你是对的。它应该是默认值。我正在删除附录，因为它混淆的程度超过了它的帮助，事后看来，我猜它确实与背压机制有关。我的消费者比网络快得多，因此在结果到达时分叉处理期货是一个更合适的解决方案。
MySQL 的情况是否有等价物？
为什么要禁用自动提交？

【解决方案2】：

在 Slick 中流式传输的正确方法是文档中提供的

val q = for (c <- coffees) yield c.image
val a = q.result
val p1: DatabasePublisher[Blob] = db.stream(a.withStatementParameters(
  rsType = ResultSetType.ForwardOnly, 
  rsConcurrency = ResultSetConcurrency.ReadOnly, 
  fetchSize = 1000 /*your fetching size*/
).transactionally)

【讨论】：