如何使用猫效应IO在Play应用程序中实现多线程池

Question

在我的Play应用程序中，我使用cats-effect的IO来服务我的请求，而不是控制器中的Future，如下所示(超级简化)：

def handleServiceResult(serviceResult: ServiceResult): Result = ...

def serviceMyRequest(request: Request): IO[ServiceResult] = ...

def myAction = Action { request =>
  handleServiceResult(
    serviceMyRequest(request).unsafeRunSync()
  )
}

然后在Play的默认线程池上(异步地)处理请求。现在，我想实现多个线程池来处理不同类型的请求。如果我使用的是Futures，我可以这样做：

val myCustomExecutionContext: ExecutionContext = ...

def serviceMyRequest(request: Request): Future[ServiceResult] = ...

def myAction = Action.async { request =>
  Future(serviceMyRequest(request))(myCustomExecutionContext)
    .map(handleServiceResult)(defaultExecutionContext)
}

但我使用的不是Futures，而是IO，我不确定实现它的正确方式。这看起来很有希望，但看起来有点笨拙：

def serviceMyRequest(request: Request): IO[ServiceResult] = ...

def myAction = Action { request =>
  val ioServiceResult = for {
    _ <- IO.shift(myCustomExecutionContext)
    serviceResult <- serviceMyRequest(request)
    _ <- IO.shift(defaultExecutionContext)
  } yield {
    serviceResult
  }
  handleServiceResult(ioServiceResult.unsafeRunSync())
}

这是实现它的正确方式吗？这里有没有最佳实践？我是不是搞砸了？谢谢。

Answer 1

好的，既然这看起来还不是很成熟，这就是我最终要实现的：

trait PlayIO { self: BaseControllerHelpers =>

  implicit class IOActionBuilder[A](actionBuilder: ActionBuilder[Request, A]) {

    def io(block: Request[A] => IO[Result]): Action[A] = {
      actionBuilder.apply(block.andThen(_.unsafeRunSync()))
    }

    def io(executionContext: ExecutionContext)(block: Request[A] => IO[Result]): Action[A] = {
      val shiftedBlock = block.andThen(IO.shift(executionContext) *> _ <* IO.shift(defaultExecutionContext))
      actionBuilder.apply(shiftedBlock.andThen(_.unsafeRunSync()))
    }

  }

}

然后(使用问题中的框架)如果我将PlayIO混合到控制器中，我可以这样做，

val myCustomExecutionContext: ExecutionContext = ...

def handleServiceResult(serviceResult: ServiceResult): Result = ...

def serviceMyRequest(request: Request): IO[ServiceResult] = ...

def myAction = Action.io(myCustomExecutionContext) { request =>
  serviceMyRequest(request).map(handleServiceResult)
}

这样我就可以在myCustomExecutionContext上执行动作的代码块，然后，一旦完成，线程就会切换回Play的默认执行上下文。

更新：

这会更灵活一些：

trait PlayIO { self: BaseControllerHelpers =>

  implicit class IOActionBuilder[R[_], A](actionBuilder: ActionBuilder[R, A]) {

    def io(block: R[A] => IO[Result]): Action[A] = {
      actionBuilder.apply(block.andThen(_.unsafeRunSync()))
    }

    def io(executionContext: ExecutionContext)(block: R[A] => IO[Result]): Action[A] = {
      if (executionContext == defaultExecutionContext) io(block) else {
        val shiftedBlock = block.andThen(IO.shift(executionContext) *> _ <* IO.shift(defaultExecutionContext))
        io(shiftedBlock)
      }
    }

  }

}

Update2:

根据上面的评论，这将确保我们始终切换回默认线程池：

trait PlayIO { self: BaseControllerHelpers =>

  implicit class IOActionBuilder[R[_], A](actionBuilder: ActionBuilder[R, A]) {

    def io(block: R[A] => IO[Result]): Action[A] = {
      actionBuilder.apply(block.andThen(_.unsafeRunSync()))
    }

    def io(executionContext: ExecutionContext)(block: R[A] => IO[Result]): Action[A] = {
      if (executionContext == defaultExecutionContext) io(block) else {
        val shiftedBlock = block.andThen { ioResult =>
          IO.shift(executionContext).bracket(_ => ioResult)(_ => IO.shift(defaultExecutionContext))
        }
        io(shiftedBlock)
      }
    }

  }

}