使用 Akka 控制外部 Futures

Question

我使用 Akka 已经有一段时间了，而且在更简单的项目中，我只使用了 Scala 2.10 期货。然而，前几天我不得不将返回 Futures 的库与 Akka Actors 混合在一起，我不知道如何集成 Actor 调度系统和失控的 futures。

在项目中使用 Akka Actor 的选择是能够微调 Actor 队列并控制代码的并行化，并且可能最终横向扩展（即使现在它不是优先级）。但是，如果我有这样的代码（故意简化）：

package externallib

object DoSomething {
    def foo(): Future[Something] = ....
}


package myapp

class MyActor extends Actor {
    def receive = {
        case "message" => externallib.DoSomething.foo() pipeTo sender
    }
}

.../
val actorRef = system.actorOf(Props[MyActor].withRouter(
                   RoundRobinRouter(nrOfInstances = 5)))
(0 to 20000) foreach {

    val futureSomething = actorRef ? "message"

}

那么 MyActor 实际上并没有做任何事情，除了在 externallib 中生成一个线程并直接返回。 externallib 未来最终会将结果通过管道传递给参与者的调用者。

但是，通过这种方式，管理 Actor 的受良好控制的路由器并没有真正控制生成的线程，因为它们是在 Actor 系统之外生成的，即使在同一个 ExecutionContext 中也是如此。在大循环的示例中，这意味着不是将消息排队发送给参与者，而是将快速消耗这些消息并在任何严格控制的队列之外生成 20000 个线程。

我在想我可以做这样的事情：

class MyActor extends Actor {
    def receive = {
        case "message" => 
            val res = Await.result(externallib.DoSomething.foo(), someDuration)
            sender ! res
    }
}

这将确保在 externallib 完成（或超时）之前不会向参与者发送新的 "message"。但是，这实际上可能需要两个线程（演员的一个和 DoSomething 中的一个）来等待单个计算。

有没有更好的方法来控制这些在参与者系统之外产生的期货？

Answer 1

MyActor 看起来很危险，在 Actor 内部阻塞并不是一个好主意，事实上，如果 ExecutionContext 中只有一个线程供 Actor 和未来使用，这将死锁。写这个的更好的方法是将onComplete放在future上，并让actor执行become的行为，使其不发送任何新的future（可能通过stashing所有传入的消息）。 future 上的onComplete 既可以向发送者发送消息，也可以向参与者发送消息以将其告诉become 原始行为。