4

鉴于：

构建.sbt

scalaVersion := "2.13.2"
libraryDependencies += "org.typelevel" %% "cats-effect" % "2.1.3"

src/main/scala/net/Main.scala

package net

import cats.effect._
import cats.implicits._
import java.util.concurrent.TimeUnit
import scala.concurrent.duration._

object App extends IOApp { self: IOApp =>
  override def run(args: List[String]): IO[ExitCode] =
    for {
      _ <- uncancellable
      _ <- notUncancellable
    } yield ExitCode.Success

  private def uncancellable: IO[Unit] = {
    val tick: IO[Unit] = Concurrent[IO].uncancelable(self.timer.sleep(10.seconds))

    for {
      _ <- IO(println("uncancellable"))
      fiber <- Concurrent[IO].start(tick)
      _ <- IO(println("seconds begin: " + FiniteDuration.apply(System.nanoTime(), TimeUnit.NANOSECONDS).toSeconds))
      _ <- fiber.cancel
      _ <- fiber.join
      _ <- IO(println("seconds done : " + FiniteDuration.apply(System.nanoTime(), TimeUnit.NANOSECONDS).toSeconds))
    } yield ()
  }

  private def notUncancellable: IO[Unit] = {
    val tick: IO[Unit] = self.timer.sleep(10.seconds)

    for {
      _ <- IO(println("notUncancellable"))
      fiber <- Concurrent[IO].start(tick)
      _ <- IO(println("seconds begin: " + FiniteDuration.apply(System.nanoTime(), TimeUnit.NANOSECONDS).toSeconds))
      _ <- fiber.cancel
      _ <- fiber.join
      _ <- IO(println("seconds done : " + FiniteDuration.apply(System.nanoTime(), TimeUnit.NANOSECONDS).toSeconds))
    } yield ()
  }
}

运行它会显示以下输出：

sbt:cats-effect-cancellation-question> run
[info] Compiling 1 Scala source to /Users/kevinmeredith/Workspace/cats-effect-cancellation-questions/target/scala-2.13/classes ...
[info] Done compiling.
[info] Packaging /Users/kevinmeredith/Workspace/cats-effect-cancellation-questions/target/scala-2.13/cats-effect-cancellation-question_2.13-0.1.jar ...
[info] Done packaging.
[info] Running net.App 
uncancellable
seconds begin: 303045
seconds done : 303055
notUncancellable
seconds begin: 303055
^C$

请注意，大约 30 秒后，我取消了它。

为什么没有"seconds done :打印出来：

notUncancellable
seconds begin: 303055
^C$

?

4

1 回答 1

1

uncancellable我相信是不言自明的。

在不可取消的情况下，您会遇到类似于此 GitHub 问题的情况。

正如亚历山德鲁·内德尔库所说：

fiber.cancel在fiber.joinIO 的情况下是非终止的。因此fiber.join永远不会完成，并且该保证永远不会有机会被评估。

如果您也取消它，您可以强制进行评估，如果您关心结果，则在实际应用程序中您需要这样做fiber.join。

据我所知，这是对合同的一种可能解释

  /**
   * Returns a new task that will await for the completion of the
   * underlying fiber, (asynchronously) blocking the current run-loop
   * until that result is available.
   */
  def join: F[A]

取消的光纤不能返回成功的值——这很明显。但是如果它返回另一个失败的异常......它也会返回一个值，该值可以被认为是由 Fiber 计算的值 - 它不应该返回任何值，因为它被取消了！

出于这个原因，在这种情况下，您的整个线程都在等待一个从未到达的值。

为了避免此类陷阱，您可以使用不那么“低级”的东西racePair或类似的东西，这样可以避免自己处理此类问题。您可以阅读 Oleg Pyzhcov 关于光纤安全的短文。

于 2020-05-24T19:47:23.010 回答