运行 在 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 实例。

Error:(107, 54) could not find implicit value for parameter p: cats.NonEmptyParallel[doobie.ConnectionIO,F] val result = (program(i1),program(i2)).parMapN{case (a ,b) => a ++ b}

我是不是遗漏了一些明显的东西或者尝试了一些无法完成的事情？ 谢谢

Answer 1

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

例如，使用 cats-effect IO 运行time 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：

You can't run ConnectionIO in parallel. It's a language describing the use of a connection which is a linear sequence of operations.

Using parMapN in IO, yes, you can run two things at the same time because they're running on different connections.

There is no parMapN with ConnectionIO because it does not (and cannot) have a Parallel instance.