scalikejdbc 使用 oracle 进行流式传输答案

【问题标题】：scalikejdbc streaming with oraclescalikejdbc 使用 oracle 进行流式传输
【发布时间】：2017-08-02 13:17:23
【问题描述】：

我正在使用 ScalikeJDBC 实现流式源，并且需要它来运行多种数据库类型，包括。 Oracle、Sybase 等

文档有点混乱，不确定这是否是一个选项：

目前，scalikejdbc-streams 原生支持 MySQL 和 PostgreSQL。正常使用 SQL#iterator 工厂方法时，ScalikeJDBC 会自动启用所需的设置以使用游标功能。如果您不喜欢这种行为，您可以自定义调整 DBSession 属性

是否可以通过 MySQL 和 PostgreSQL 以外的其他 DB 处理流式读取？

【问题讨论】：

标签： scala scalikejdbc

【解决方案1】：

（由于您的问题是关于创建流媒体源，因此此答案仅针对流媒体支持的发布者端，而忽略订阅者端。）

支持流式传输要求数据库一次返回几行查询结果，通常基于游标，而不是一次全部返回。不同的数据库有不同的方法来实现这一点。 ScalikeJDBC 原生支持对 MySQL 和 PostgreSQL 驱动程序使用流式iterator 方法。也就是说，使用 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，因为 this:

/**
 * 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 默认情况下不支持带有除 MySQL 和 PostgreSQL 驱动程序之外的驱动程序的流 iterator。

这并不意味着不能使用其他驱动程序进行流式传输。您引用的文档部分具有以下代码示例：

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）取决于驱动程序（假设驱动程序支持一次检索少量行的查询结果）。

【讨论】：