在ZIO上将Hikari事务处理程序用于Doobie时遇到死锁

问题描述

我在ZIO应用程序中使用Doobie，有时会出现死锁（应用程序完全冻结）。如果我的应用程序仅在一个内核上运行，或者达到了数据库的最大并行连接数，则会发生这种情况。

我的代码如下：

def mkTransactor(cfg: DatabaseConfig): RManaged[Blocking,Transactor[Task]] =
    ZIO.runtime[Blocking].toManaged_.flatMap { implicit rt =>
      val connectEC = rt.platform.executor.asEC
      val transactEC = rt.environment.get.blockingExecutor.asEC

      HikariTransactor
        .fromHikariConfig[Task](
          hikari(cfg),connectEC,Blocker.liftExecutionContext(transactEC)
        )
        .toManaged
    }

  private def hikari(cfg: DatabaseConfig): HikariConfig = {
    val config = new com.zaxxer.hikari.HikariConfig

    config.setJdbcUrl(cfg.url)
    config.setSchema(cfg.schema)
    config.setUsername(cfg.user)
    config.setPassword(cfg.pass)

    config
  }

或者，我在Hikari（config.setLeakDetectionThreshold(10000L)）上设置了泄漏检测参数，但出现了泄漏错误，这不是处理数据库查询所花费的时间。

解决方法

Doobie文档中有关于执行上下文和对每个上下文的期望的很好的解释：https://tpolecat.github.io/doobie/docs/14-Managing-Connections.html#about-transactors

根据文档，“用于等待数据库连接的执行上下文”（问题中的connectEC）应为边界。

ZIO默认情况下只有两个线程池：

1. zio-default-async –有界，
2. zio-default-blocking –无界限

因此，很自然地相信我们应该使用zio-default-async，因为它是有界的。

不幸的是，zio-default-async假设其操作永远不会阻塞。这是非常重要的，因为它是ZIO解释程序（其运行时）用来运行的执行上下文。如果您对此进行阻止，则实际上可以阻止ZIO程序的评估进度。当只有一个内核可用时，这种情况会更经常发生。

问题在于，等待数据库连接的执行上下文会被阻塞，以等待Hikari连接池中的可用空间。因此，我们不应该在此执行上下文中使用zio-default-async。

下一个问题是：仅为connectEC创建一个新的线程池和相应的执行上下文是否有意义？没有什么可以禁止您这样做的，但是由于以下三个原因，可能没有必要：

• 您要避免创建线程池，特别是因为您可能已经从Web框架，DB连接池，调度程序等中创建了多个线程池。每个线程池都有其成本。一些例子是：

  • 要为jvm JVM管理的更多内容
  • 消耗更多的操作系统资源
  • 在线程之间切换，这在性能上是昂贵的
  • 使您的应用程序运行时变得更加复杂（例如复杂的线程转储等）

• ZIO线程池的人机工程学已经开始针对其使用情况进行了最佳优化

• 在一天结束时，您将不得不在某处管理超时，并且连接不是系统的一部分，该系统最有可能拥有足够的信息来知道应该等待多长时间：互动（例如，在应用的外部，更接近使用点）可能需要不同的超时/重试逻辑。

话虽如此，我们发现一种配置可以在生产环境中运行的应用程序中很好地工作：

// zio.interop.catz._ provides a `zioContextShift`

  val xa = (for {
    // our transaction EC: wait for aquire/release connections,must accept blocking operations
    te <- ZIO.access[Blocking](_.get.blockingExecutor.asEC)
  } yield {
    Transactor.fromDataSource[Task](datasource,te,Blocker.liftExecutionContext(te))
  }).provide(ZioRuntime.environment).runNow

  def transactTask[T](query: Transactor[Task] => Task[T]): Task[T] = {
    query(xa)
  }

我绘制了Doobie和ZIO执行上下文如何相互映射的图：https://docs.google.com/drawings/d/1aJAkH6VFjX3ENu7gYUDK-qqOf9-AQI971EQ4sqhi2IY