在 Doobie 中并行运行查询

Question

是否可以使用 Doobie 并行运行多个查询？

我有以下（伪）查询：

def prepareForQuery(input: String): ConnectionIO[Unit] = ???
val gettAllResults: ConnectionIO[List[(String, BigDecimal)]] = ???
def program(input : String) : ConnectionIO[List[(String, BigDecimal)]] = for{
    _ <- prepareForQuery(input)
    r <- gettAllResults
  } yield r

我尝试的是以下内容：

import doobie._
import doobie.implicits._
import cats.implicits._
val xa = Transactor.fromDataSource[IO](myDataSource)
val result = (program(i1),program(i2)).parMapN{case (a,b) => a ++ b}
val rs = result.transact(xa).unsafeRunSync

但是，没有找到 ConnectionIO 的 NonEmptyParallel 实例。

错误：(107, 54) 找不到参数 p 的隐式值： cat.NonEmptyParallel[doobie.ConnectionIO,F] val 结果 = (program(i1),program(i2)).parMapN{case (a ,b) => a ++ b}

我是否遗漏了一些显而易见的事情或尝试了一些无法完成的事情？ 谢谢

Answer 1

您不能在 ConnectionIO monad 中并行运行查询。但是只要你把它们变成你实际的运行时 monad（只要它有一个 Parallel 实例），你就可以。

例如，使用猫效应 IO 运行时 monad：

def prepareForQuery(input: String): ConnectionIO[Unit] = ???
val gettAllResults: ConnectionIO[List[(String, BigDecimal)]] = ???
def program(input : String) : ConnectionIO[List[(String, BigDecimal)]] = for{
    _ <- prepareForQuery(input)
    r <- gettAllResults
  } yield r

将您的ConnectionIO 变成IO

val program1IO: IO[List[(String, BigDecimal)]]] = program(i1).transact(xa)
val program2IO: IO[List[(String, BigDecimal)]]] = program(i2).transact(xa)

您现在有一个可以并行执行操作的 monad。

val result: IO[List[(String, BigDecimal)]]] = 
    (program1IO, program2IO).parMapN{case (a,b) => a ++ b}

---

要了解为什么ConnectionIO 不允许您并行执行操作，我将引用 tpolecat：

您不能并行运行 ConnectionIO。它是一种描述连接使用的语言，连​​接是线性的操作序列。

在 IO 中使用 parMapN，是的，您可以同时运行两个东西，因为它们运行在不同的连接上。

ConnectionIO 没有 parMapN，因为它没有（也不能）有 Parallel 实例。