ConnectionIO事务中的并行操作

Question

因此，我有一个程序，在这个程序中，我从数据库中获取文件路径列表，删除文件系统上的文件，最后从数据库中删除文件路径。我将所有操作放入事务中，以确保从数据库中删除路径，当文件系统中的所有文件都被删除时。

就像这样

val result = for {
deletePath <- (fr""" select path from files""").query[String].stream //Stream[doobie.ConnectionIO,String]
_ <- Stream.eval(AsyncConnectionIO.liftIO(File(deletePath).delete()) //Stream[doobie.ConnectionIO,Unit]
_ <- Stream.eval(sql"delete from files where path = ${deletePath}".withUniqueGeneratedKeys) 
}

result.compile.drain.transact(transactor)

不幸的是，文件系统是分布式的，这意味着单独的操作是缓慢的，但它允许同时进行多个操作。

因此，我的问题是，如何在这里并行文件系统删除操作？

Answer 1

是的，你可以的。只需使用适当的组合器，而不是for语法。

val result =
  (fr""" select path from files""")
    .query[String]
    .stream
    .parEvalMapUnordered(maxConcurrent = 64) { deletePath =>
      AsyncConnectionIO.liftIO(File(deletePath).delete()) >>
      sql"delete from files where path = ${deletePath}".withUniqueGeneratedKeys
    }

result.compile.drain.transact(transactor)

记住将maxConcurrent参数更改为对您的用例有意义的东西。

(我无法测试代码，所以它可能有一些排字)