scala - 使用多个 ConnectionPool 的正确方法

Question

在我的应用程序中，我必须与多个 MySQL 交互(只读) DB 一个一个。对于每个数据库，我需要一定数量的连接。与数据库的交互不会一次性发生:我查询数据库，花一些时间处理结果，再次查询数据库，再次处理结果等等。

这些交互中的每一个都需要多个连接 [ 我同时触发多个查询 ]，因此我需要一个 ConnectionPool它在我开始与数据库交互时产生并一直存在，直到我完成对该数据库的所有查询(包括我不查询时的中间时间间隔，只处理结果)。

我能够成功创建一个 ConnectionPool具有所需的连接数并获得 implicit session如下所示

def createConnectionPool(poolSize: Int): DBSession = {
 implicit val session: AutoSession.type = AutoSession

 ConnectionPool.singleton(
   url = "myUrl",
   user = "myUser",
   password = "***",
   settings = ConnectionPoolSettings(initialSize = poolSize)
 )

 session
}

然后我通过这个 implicit session在我需要与数据库交互的整个方法中。这样，我就可以开火 poolSize没有同时使用这个的查询 session .很公平。

def methodThatCallsAnotherMethod(implicit session: DBSession): Unit = {
  ...
  methodThatInteractsWithDb
  ...
}

def methodThatInteractsWithDb(implicit session: DBSession): Unit = {
  ...
  getResultsParallely(poolSize = 32, fetchSize = 2000000)
  ...
}

def getResultsParallely(poolSize: Int, fetchSize: Int)(implicit session: DBSession): Seq[ResultClass] = {
  import java.util.concurrent.Executors
  import scala.concurrent.ExecutionContext
  import scala.concurrent.duration._

  implicit val ec: ExecutionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(poolSize))

  val resultsSequenceFuture: Seq[Future[ResultClass]] = {
    (0 until poolSize).map { i =>
      val limit: Long = fetchSize
      val offset: Long = i * fetchSize

      Future(methodThatMakesSingleQuery(limit, offset))
    }
  }
  val resultsFutureSequence: Future[Seq[ResultClass]] = Future.sequence(resultsSequenceFuture)

  Await.result(resultsFuture, 2.minutes)
}

这种技术有两个问题:

我的应用程序很大，有很多嵌套的方法调用，所以通过 implicit session通过这样的所有方法(见下文)是不可行的。

除了上述与不同 DB 的交互之外，我还需要在整个应用程序的整个生命周期中与另一个(固定)DB 建立单一连接。此连接将用于每隔几分钟进行一次小型写入操作(记录我与其他 DB 的交互进度)。因此，我需要多个 ConnectionPool s，每个 DB 一个

据我所知 ScalikeJdbc的 docs ，我想出了以下不需要我通过 implicit session 的方法。到处。

def createConnectionPool(poolName: String, poolSize: Int): Unit = {
  ConnectionPool.add(
    name = poolName,
    url = "myUrl",
    user = "myUser",
    password = "***",
    settings = ConnectionPoolSettings(initialSize = poolSize)
  )
}

def methodThatInteractsWithDb(poolName: String): Unit = {
  ...
  (DB(ConnectionPool.get(poolName).borrow())).readOnly { implicit session: DBSession =>
    // interact with DB
    ...
  }
  ...
}

虽然这有效，但我不再能够并行化数据库交互。这种行为很明显，因为我使用的是 borrow()方法，从池中获取单个连接。这反过来又让我想知道为什么 AutoSession更早的事情:为什么我能够使用单个 implicit session 同时触发多个查询?如果那东西奏效了，那为什么它不起作用呢？但我没有找到如何获得 DBSession 的例子。来自 ConnectionPool支持多个连接。

总而言之，我有 2 个问题和 2 个解决方案:每个问题一个。但我需要一个解决这两个问题的单一(通用)解决方案。
ScalikeJdbc的有限文档没有提供很多帮助和博客/文章 ScalikeJdbc几乎不存在。
请提出正确的方法/一些解决方法。

框架版本

Scala 2.11.11

"org.scalikejdbc" %% "scalikejdbc" % "3.2.0"

Answer 1

感谢 @Dennis Hunziker ，我能够弄清楚the correct way释放从 ScalikeJdbc 借来的连接的 ConnectionPool .可以按如下方式完成:

import scalikejdbc.{ConnectionPool, using}
import java.sql.Connection

using(ConnectionPool.get("poolName").borrow()) { (connection: Connection) =>
    // use connection (only once) here
}
// connection automatically returned to pool

有了这个，现在我可以并行化与池的交互。

解决我管理多个 ConnectionPool的问题s 并使用跨多个连接 class es，我最终写了一个 ConnectionPoolManager , 完整代码可在 here 中找到.通过卸载任务

创建池

从池中借用连接

移除池

到 singleton我可以在我的项目中的任何地方使用的对象，我能够清除很多困惑并消除了需要通过 implicit session跨方法链。

编辑-1

虽然我已经 linked ConnectionPoolManager 的完整代码, 这里有一个关于如何去做的快​​速提示

以下方法 ConnectionPoolManager允许您从 ConnectionPool 借用连接秒

def getDB(dbName: String, poolNameOpt: Option[String] = None): DB = {
  // create a pool for db (only) if it doesn't exist
  addPool(dbName, poolNameOpt)

  val poolName: String = poolNameOpt.getOrElse(dbName)
  DB(ConnectionPool.get(poolName).borrow())
}

此后，在整个代码中，您可以使用上述方法从池中借用连接并进行查询

def makeQuery(dbName: String, poolNameOpt: Option[String]) = {
  ConnectionPoolManager.getDB(dbName, poolNameOpt).localTx { implicit session: DBSession =>
    // perform ScalikeJdbc SQL query here
  }
}