如何运行不阻塞主线程但如果失败在主线程中抛出异常的异步 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，如果两个IO中的一个失败，那么另一个将运行cancel。
现在，如果longRunningOperation完成没有错误，然后等待app; 但是，如果app是一个洗完头，于是我决定cancel的longRunningOperation （当然，你可以的，更改）。

感谢 Adam Rosien 指出，如果您想join在两种情况下都这样做，那么最好是：

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

您可以在此处查看正在运行的代码并进行操作