scala - 如何运行不阻塞主线程但在主线程失败时抛出异常的异步 IO 操作？

Question

我正在调查 Cats 以完成上述任务。我试着写下面的例子

1. 运行不阻塞主线程的 IO 操作
2. 如果失败则在主线程中抛出异常

import cats.effect.{IO, Async}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future

val apiCall = Future.successful("I come from the Future!")

val ioa: IO[String] =
  Async[IO].async { cb =>
    import scala.util.{Failure, Success}
    Thread.sleep(1000)
    throw new RuntimeException

    apiCall.onComplete {
      case Success(value) => cb(Right(value))
      case Failure(error) => cb(Left(error))
    }
 }

ioa.unsafeRunAsync(result => result match {
    case Left(result) => throw result
    case Right(_) =>
})

然而，它似乎在这两方面都失败了。它阻塞主线程，只在子线程中抛出异常。我尝试做的事情可行吗？

Answer 1

你可以这样做:

val longRunningOperation: IO[Unit] = ...
val app: IO[Unit] = ...

val program: IO[Unit] =
  IO.racePair(app, longRunningOperation).flatMap {
    case Left((_, fiberLongRunning)) =>  fiberLongRunning.cancel.void
    case Right((fiberApp, _)) => fiberApp.join.void
  }

override def run(args: List[String]): IO[ExitCode] =
  program.as(ExitCode.Success)

racePair 将同时运行两个 IO，如果其中一个失败，则它将取消另一个。
现在，如果 longRunningOperation 没有错误地完成，那么等待 app；但是，如果 app 是第一个完成的，那么我决定取消 longRunningOperation (你当然可以，改变那个)。

---

感谢 Adam Rosien 指出，如果您想在这两种情况下加入，那么最好是:

val program: IO[Unit] = (app, longRunningOperation).parTupled.void

---

您可以看到代码正在运行并使用它 here