使用猫效应的移位函数

Question

我正在尝试使用 cats effect shift 函数来运行代码异步。

函数实现：

  def asyncSendMsg(producer: KkProducer)(record: KkRecord) : IO[Either[String, RecordMetadata]] =
    for {
      res <- trySendMsg(producer)(record).shift(BlockingFileIO).shift(Main)
    } yield(res)

  def trySendMsg(producer: KkProducer)(record: KkRecord): IO[Either[String, RecordMetadata]] =
    IO {
      try {
        Right(producer.send(record).get())
      } catch {
        case e: Exception => Left(e.getMessage())
      }
    }

尝试编译，我收到：

[error] /home/developer/Desktop/scala/PureProducer/src/main/scala/TheProducer.scala:51:43: value shift is not a member of cats.effect.IO[Either[String,org.apache.kafka.clients.producer.RecordMetadata]]
[error]       res <- trySendMsg(producer)(record).shift(BlockingFileIO).shift(Main)
[error]                                           ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed
[error] Total time: 2 s, completed Nov 13, 2017 2:18:06 PM

我必须导入哪个库才能使用移位功能？

Answer 1

IO.shift 是在 IO 的伴生对象上定义的（API 在不同版本的猫之间发生了变化），您可以在调用之前在 for comprehension 中使用它IO调用：

val nonBlockingExecContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))

val res = for {
  _ <- IO { println(Thread.currentThread().getName) }
  _ <- IO.shift(nonBlockingExecContext)
  _ <- IO { println(Thread.currentThread().getName) }
} yield ()

res.unsafeRunSync()

产量：

main
pool-1-thread-1

使用猫效应的移位函数

Using shift function from cats effect

scala

scala-cats