akka - 在 Akka 中处理 IO/阻塞操作的最正确方法

Question

我有一个关于处理 IO 阻塞操作的正确方法的问题，我读到正确的模型是将阻塞操作包装到 Future 中并使用这个对象而不是调用 actor 内部的阻塞操作。

但我认为这至少有两个解决方案

case class AccountBalance(id: Int, userId: Int, total: Int)

object AccountBalance {
  def getUserAccountBalance(userId: Int)(
      implicit ec: ExecutionContext): Future[AccountBalance] = {
    Future {
      AccountBalance(1, userId, 1)
    }
  }

  def updateAccountBalance(id: Int, total: Int)(implicit ec:   ExecutionContext): Future[AccountBalance] = {
    Future {
      AccountBalance(id, 1, total)
    }
  }

  // Main logic 
  def getAndInc(userId: Int)(implicit ec: ExecutionContext) = {
    getUserAccountBalance(userId).flatMap { balance =>
    updateAccountBalance(balance.id, balance.total + 1)
  }
 }
}

在第一种方法中，我在 actor 中使用 AccountBalanece.getAndInc 方法:

class Approach1 extends Actor {
  implicit val executionContext = context.dispatcher

  def receive = {
    case Calculate(userId) =>
      AccountBalance.getAndInc(userId) pipeTo sender
  }
}

另一种解决方案(对我来说更舒服)

class Approach2 extends Actor {
  implicit val executionContext = context.dispatcher

  var firstSender: ActorRef = null
  var userId: Int = -1

  def receive = {
    case Calculate(givenUserId) =>
      userId = givenUserId
      firstSender = sender

      AccountBalance.getUserAccountBalance(userId) pipeTo self

    case AccountBalance(id, _, total) =>
      AccountBalance.updateAccountBalance(id, total + 1) pipeTo    firstSender

  }
} 

哪种解决方案更好(或既危险又不可用)？

关于 ExecutionContext 的第二个问题，例如我在实际应用中使用 context.dispatcher 我使用下一个:

class A extends Actor {

  implicit val executionContext = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(10)) 

  override def postStop = {
     executionContext.shutdown()
  }

  def receive = { case _ => }

}

如果我在 actor 停止后使用 executionContext.shutdown，这是关闭线程池并释放所有资源的正确方法吗？

Answer 1

正如响应式编程所定义的那样，需要确保执行线程不被阻塞。因此，任何阻塞操作都应由不同的执行上下文处理，以免阻塞分派(dispatch)参与者命令/消息的线程。我使用第二种模式，其中我的 Actor 使用“成为”并转到不同的行为等待来自阻塞操作的响应。这是为了不处理任何新请求并丢失原始发送者 actorRef。如果您仍希望 Actor 接收其他消息并发送这些消息，您需要在原始发件人和收到的响应之间建立链接。

是的，以这种方式停止它会起作用，但我建议在创建线程池时要非常小心，并在阻塞参与者之间共享它们。请注意，如果关闭抛出异常，actor A 不会被主管重新启动，并且您只想关闭 actor A，而不是所有 akka 系统。