(由于您的问题是关于创建流媒体源,因此此答案仅针对流媒体支持的发布者端,而忽略了订阅者端。)
对流的支持要求数据库一次返回几行查询结果,通常基于游标,而不是一次全部返回。不同的数据库有不同的方法来实现这一点。ScalikeJDBC 原生支持对iterator
MySQL 和 PostgreSQL 驱动程序使用流式方法。也就是说,使用 MySQL 和 PostgreSQL 驱动程序,以下工作:
import scalikejdbc._
import scalikejdbc.streams._
// set up a connection pool
import scala.concurrent.ExecutionContext.Implicits.global
val publisher: DatabasePublisher[Int] = DB.readOnlyStream {
sql"select id from users order by id".map(r => r.get[Int]("id")).iterator
}
由于这个原因,以上内容适用于 MySQL 和 PostgreSQL :
/**
* Forcibly changes the database session to be cursor query ready.
*/
val defaultDBSessionForceAdjuster: DBSessionForceAdjuster = (session) => {
// setup required settings to enable cursor operations
session.connectionAttributes.driverName match {
case Some(driver) if driver == "com.mysql.jdbc.Driver" && session.fetchSize.exists(_ > 0) =>
/*
* MySQL - https://dev.mysql.com/doc/connector-j/5.1/en/connector-j-reference-implementation-notes.html
*
* StreamAction.StreamingInvoker prepares the following required settings in advance:
*
* - java.sql.ResultSet.TYPE_FORWARD_ONLY
* - java.sql.ResultSet.CONCUR_READ_ONLY
*
* If the fetchSize is set as 0 or less, we need to forcibly change the value with the Int min value.
*/
session.fetchSize(Int.MinValue)
case Some(driver) if driver == "org.postgresql.Driver" =>
/*
* PostgreSQL - https://jdbc.postgresql.org/documentation/94/query.html
*
* - java.sql.Connection#autocommit false
* - java.sql.ResultSet.TYPE_FORWARD_ONLY
*/
session.conn.setAutoCommit(false)
case _ =>
}
}
请注意,最后一个case
子句意味着 ScalikeJDBC默认情况下不支持使用iterator
MySQL 和 PostgreSQL 以外的驱动程序进行流式传输。
这并不意味着不能使用其他驱动程序进行流式传输。您引用的文档部分具有以下代码示例:
val publisher: DatabasePublisher[Int] = DB readOnlyStream {
sql"select id from users".map(r => r.int("id"))
.iterator
.withDBSessionForceAdjuster(session => {
session.conn.setAutoCommit(true)
})
}
文档所说的是,要为 MySQL 和 PostgreSQL 以外的数据库启用流式传输,您需要DBSession
像上面的示例一样自定义属性,以便启用游标支持。这种定制究竟需要什么(例如,调整fetchSize
或禁用autoCommit
连接)取决于驱动程序(假设驱动程序支持一次检索少量行的查询结果)。