使用 slick 的 3.0.0 流结果和 Postgresql 的正确方法是什么? [英] What is the right way to work with slick's 3.0.0 streaming results and Postgresql?

问题描述

我想弄清楚如何使用流畅的流媒体.我使用带有 postgres 驱动程序的 slick 3.0.0

I am trying to figure out how to work with slick streaming. I use slick 3.0.0 with postgres driver

情况如下:服务器必须将客户端数据序列拆分为受大小(以字节为单位)限制的块.所以，我写了以下巧妙的查询:

The situation is following: server have to give client sequences of data split into chunks limited by size(in bytes). So, I wrote following slick query:

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 日志时，我看到这样的查询 从序列中选择 *，其中 user_id = 0 和时间戳 >0 按时间戳排序；

I combined seq with akka-streams Source, wrote custom PushPullStage, that limits size of data(in bytes) and finishes upstream when it reaches size limit. It works just fine. The problem is - when I look into postgres logs, I see query like that select * from sequences where user_id = 0 and timestamp > 0 order by timestamp;

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

So, at first glance it appears to be much (and unnecessary) database querying going on, only to use a few bytes in each query. What is the right way to do streaming with Slick so as to minimize database querying and to make best use of the data transferred in each query?

推荐答案

使用 Slick 和 Postgres 进行流式处理的正确方法"包括三件事:

The "right way" to do streaming with Slick and Postgres includes three things:

1. 必须使用 db.stream()

1. Must use db.stream()

必须在 JDBC 驱动程序中禁用 autoCommit.一种方法是通过添加后缀 .transactionally.

Must disable autoCommit in JDBC-driver. One way is to make the query run in a transaction by suffixing .transactionally.

必须将 fetchSize 设置为 0 以外的值，否则 postgres 会将整个 resultSet 一次性推送到客户端.

Must set fetchSize to be something else than 0 or else postgres will push the whole resultSet to the client in one go.

例如:

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

这篇关于使用 slick 的 3.0.0 流结果和 Postgresql 的正确方法是什么?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！