标签: cats-effect

我正在尝试使用cats-effect以纯粹的功能方式获取一些基本文件IO（写入/读取）。遵循本教程之后，下面就是我读取文件的内容：

private def readFile(): IO[String] = for {
  lines <-  bufferedReader(new File(filePath)).use(readAllLines)
} yield lines.mkString

def bufferedReader(f: File): Resource[IO, BufferedReader] =
  Resource.make {
    IO(new BufferedReader(new FileReader(f)))
  } { fileReader =>
    IO(fileReader.close()).handleErrorWith(_ => IO.unit)
  }

现在在handleErrorWith函数中，我可以记录发生的任何错误，但是如何为它添加适当的错误处理（例如，返回Resource[IO, Either[CouldNotReadFileError, BufferedReader]]）？

在PR审查期间,我被要求替换Sync[F].delay,Sync[F].catchNonFatal因为可能会抛出异常.

这确实有效:

scala> Sync[IO].delay(throw new Exception).recover{ case t: Throwable => 42 }.unsafeRunSync
res10: Int = 42

不确定这种行为是否具体IO,我也能找到相应的法律说它实际上是预期的,但我在主要的cat-effect文档中找不到关于API中异常的自动处理的提及.

有谁知道的理由和预期的行为,然后猫效应WRT异常抛出.delay或.map或.flatMap？

我看cats.effect.concurrent.Deferred,发现所有纯正它的同伴对象中的工厂方法返回的F[Deferred[F, A]],不只是Deferred[F, A]像

def apply[F[_], A](implicit F: Concurrent[F]): F[Deferred[F, A]] =
  F.delay(unsafe[F, A])

但

  /**
    * Like `apply` but returns the newly allocated promise directly instead of wrapping it in `F.delay`.
    * This method is considered unsafe because it is not referentially transparent -- it allocates
    * mutable state.
    */
  def unsafe[F[_]: Concurrent, A]: Deferred[F, A]

为什么？

在abstract class具有限定(文档略)两种方法:

abstract class Deferred[F[_], A] {
  def get: F[A]
  def complete(a: A): …

根据cats官方文档：https : //typelevel.org/cats-effect/typeclasses/liftio.html ，如果我们想把东西从IO提升到其他容器，你应该实现LiftIO trait，但示例明确运行unsafeRunXXX方法来获取出了效果，我想知道这是转型的唯一途径吗？

我经常做这样的事情：

import cats.effect.Sync
import cats.implicits._

case class User(name: String)
case object Error extends Exception

def validate[F[_]: Sync](name: String): F[Either[Error, User]] = Sync[F].pure(User(name).asRight)

def doSomething[F[_]: Sync]: F[User] = for {
   maybeUser <- validate("Name")
   user <- maybeUser.fold(Sync[F].raiseError[User](_), Sync[F].pure(_))
} yield user

简而言之，这意味着 if Eitheris leftthen use raiseError，如果它right只是返回值。

有没有更方便的“拆包” 方法Either？

我有这样的东西（这是来自https://github.com/typelevel/fs2的一个例子，有我的补充，我用评论标记）：

import cats.effect.{Blocker, ExitCode, IO, IOApp, Resource}
import fs2.{io, text, Stream}
import java.nio.file.Paths

object Converter extends IOApp {

  val converter: Stream[IO, Unit] = Stream.resource(Blocker[IO]).flatMap  { blocker =>
    def fahrenheitToCelsius(f: Double): Double =
      (f - 32.0) * (5.0/9.0)

    io.file.readAll[IO](Paths.get("testdata/fahrenheit.txt"), blocker, 4096)
      .balanceAvailable // my addition
      .map ( worker => // my addition
        worker // my addition
          .through(text.utf8Decode)
          .through(text.lines)
          .filter(s => !s.trim.isEmpty && !s.startsWith("//"))
          .map(line => fahrenheitToCelsius(line.toDouble).toString)
          .intersperse("\n")
          .through(text.utf8Encode)
          .through(io.file.writeAll(Paths.get("testdata/celsius.txt"), blocker))
      ) // my addition
      .take(4).parJoinUnbounded // my addition
  } …

我已经围绕这个主题浏览了很多。

就来自 stackoverflow，我发现的最好的综合性帖子是

Try[Result]、IO[Result]、Either[Error,Result]，我到底应该使用哪个

如何向 cats-effect 的资源添加正确的错误处理

我想知道目前猫效果的技术水平如何？

https://typelevel.org/blog/2018/04/13/rethinking-monaderror.html中引入的 MonadBundler很有趣，但听起来没有维护，我想知道它是否已集成到 cats-effects 2 或 3 但已重命名？

另外，我想知道这个说法是否仍然正确：

IO 可能出现的另一个问题是，当您在 IO 上使用 EitherT monad 转换器时，最终会得到同一类型的两个不同的 MonadError 实例。即，您同时拥有 MonadError[EitherT[IO, E, ?], E] 和 MonadError[EitherT[IO, E, ?], Throwable]，这可能会导致一些微妙的错误。对于 EitherT[UIO, E, ?], E]，只有一种错误类型，因此只有一个 MonadError 实例。

由于我目前正在使用 cats-effects 而不是 Zio，我想知道使用 cats-effects IO 时处理错误的最佳实践是什么？

PS：在 stackoverflow 中，我发现的问题的最佳描述是在这篇博客文章中https://guillaumebogard.dev/posts/function-error-handling/

我有以下代码：

import cats.effect.IO
import cats.data.State
import cats.data.StateT
import cats.implicits._
import cats.effect.LiftIO

abstract class Example {
    object implicits {
        implicit def myEffectLiftIO: LiftIO[IOGameplay] =
            new LiftIO[IOGameplay] {
                override def liftIO[A](ioa: IO[A]): IOGameplay[A] = {
                    StateT.liftF(ioa)
                }
            }
    }

    type Gameplay[A] = State[GameState, A]
    type IOGameplay[A] = StateT[IO, GameState, A]
    type EitherDirection[A] = Either[Throwable, A]

    type Map = Array[Array[FieldType]]
    sealed trait FieldType
    case class GameState(map: Map, block: Block)
    case class Block(f1: Field, f2: Field) 
    case class Field()

    import implicits._
    val L = …

从本教程https://github.com/slouc/concurrency-in-scala-with-ce#threading \nasync 操作分为 3 组，需要显着不同的线程池来运行：

\n

非阻塞异步操作：

\n

有界池的线程数量非常少（甚至可能只有一个），但优先级非常高。这些线程基本上大部分时间都处于空闲状态，并不断轮询是否有新的异步 IO 通知。这些线程处理请求所花费的时间直接映射到应用程序延迟，因此除了接收通知并将其转发到应用程序的其余部分之外，在此池中不执行任何其他工作非常重要。\n有界池，具有非常大的延迟线程数量较少（甚至可能只有一个），但优先级非常高。这些线程基本上大部分时间都处于空闲状态，并不断轮询是否有新的异步 IO 通知。这些线程处理请求所花费的时间直接映射到应用程序延迟，因此除了接收通知并将其转发到应用程序的其余部分之外，不要在此池中完成其他工作，这一点非常重要。

\n

\n

阻塞异步操作：

\n

无界缓存池。无限制，因为阻塞操作可以（并且将会）阻塞线程一段时间，并且我们希望能够同时服务其他 I/O 请求。缓存是因为创建太多线程可能会耗尽内存，因此重用现有线程很重要。

\n

\n

CPU 密集型操作：

\n

固定池，其中线程数等于 CPU 核心数。这非常简单。回到过去，“黄金法则”是线程数 = CPU 核心数 + 1，但“+1”来自这样一个事实：总是为 I/O 保留一个额外的线程（如上所述，我们现在有单独的池）。

\n

在我的 Cats Effect 应用程序中，我使用基于 Scala Future 的 ReactiveMongo lib 来访问 MongoDB，它在与 MongoDB 通信时不会阻塞线程，例如执行非阻塞 IO。

它需要执行上下文。\n猫效果提供默认执行上下文IOApp.executionContext

我的问题是：我应该使用哪个执行上下文来进行非阻塞 io？

IOApp.executionContext？

但是，从IOApp.executionContext文档来看：

\n

为应用程序提供默认的 ExecutionContext。

\n

JVM 顶部的默认设置是根据可用 CPU 的数量延迟构建为固定线程池（请参阅 PoolUtils）。

\n

我正在学习 F[_] 作为其他类型的构造函数的概念，但是你如何向另一个人发音或在你的脑海中说出它（对于我们内部独白思考者）。

类似于“x到x => x + 1x 加一”的官方语言，我的内心独白是如何解析的？def stream[F[_]: Async]: Stream[F, Nothing] = ...