我有一个演员，当他收到一条消息时，他开始循环计算，并且他做了一段时间（就像他做了 100 次一样）。现在我需要他对可能尽快收到的其他消息做出反应。最好的方法是在他的循环中添加一些指令，例如“如果队列中有消息做出反应然后返回这里”，但我还没有看到这样的功能。
我认为演员可以向自己发送消息而不是循环，然后这些消息将在最后排队，他会在其间对其他消息做出反应，但我听说沟通很糟糕（更多时间消耗而不是计算）并且不知道与自我的交流是否如此。
我的问题是您如何看待这种解决方案，您还有其他想法如何处理计算之间的通信吗？
【问题讨论】：
不应在主接收方法中进行耗时的计算，因为它会降低系统的响应能力。将计算置于阻塞中
Future 或Task 或其他异步对象，并在计算完成时向actor发送消息。当计算在不同的线程上继续时，actor 可以继续尽快处理消息。
如果参与者需要在运行时（响应消息）修改计算，这会变得更加复杂，但解决方案取决于计算是什么以及需要什么样的修改，因此实际上不可能给出一个一般性的答案。
【讨论】：
通常在 Akka 中，您希望限制“每个单元”完成的工作量，在这种情况下，一个单元：
Future/Task 或相同的回调中完成的工作过长的工作单元很容易通过消耗一个线程来限制整个系统的响应能力。对于不消耗 CPU 但被阻塞等待 I/O 的任务，可以在不同的线程池中执行，但对于一些消耗 CPU 的工作，这并没有真正的帮助。
因此，如果您正在执行循环，那么广义的方法是将循环的状态暂停到一条消息中并将其发送给您自己。它引入了一个小的性能损失（构建消息的延迟，将其发送给自己（保证为本地发送），并且在系统空闲时解构它可能会在微秒级），但可以改善整体系统延迟。
例如，假设我们有一个演员将计算nth 斐波那契数。我正在使用 Akka Typed 来实现这一点，但广泛的原则适用于 Classic：
object Fibonacci {
sealed trait Command
case class SumOfFirstN(n: Int, replyTo: ActorRef[Option[Long]]) extends Command
private object Internal {
case class Iterate(i: Int, a: Int, b: Int) extends Command
val initialIterate = Iterate(1, 0, 1)
}
case class State(waiting: SortedMap[Int, Set[ActorRef[Option[Long]]]]) {
def behavior: Behavior[Command] =
Behaviors.receive { (context, msg) =>
msg match {
case SumOfFirstN(n, replyTo) =>
if (n < 1) {
replyTo ! None
Behaviors.same
} else {
if (waiting.isEmpty) {
context.self ! Internal.initialIterate
}
val nextWaiting =
waiting.updated(n, waiting.get(n).fold(Set(replyTo))(_.incl(replyTo))
copy(waiting = nextWaiting).behavior
}
case Internal.Iterate(i, a, b) =>
// the ith fibonacci number is b, the (i-1)th is a
if (waiting.rangeFrom(i).isEmpty) {
// Nobody waiting for this run to complete
if (waiting.nonEmpty) {
context.self ! Internal.initialIterate
}
Behaviors.same
} else {
var nextWaiting = waiting
var nextA = a
var nextB = b
(1 to 10).foreach { x =>
val next = nextA + nextB
nextWaiting.get(x + i).foreach { waiters =>
waiters.foreach(_ ! Some(next))
}
nextWaiting = nextWaiting.removed(x + i)
nextA = nextB
nextB = next
}
context.self ! Internal.Iterate(i + 10, nextA, nextB)
copy(waiting = nextWaiting)
}
}
}
}
}
请注意，针对同一数字的多个请求（如果时间足够接近）只会计算一次，而中间结果的临时关闭请求不会导致额外计算。
【讨论】：
一个选项是委派任务，使用例如：Future，并使用一个单独的 ExecutionContext，其 fixed-pool-size（可在 application.conf 中配置）等于 CPU（或内核）的数量，以便计算使用可用的内核有效地完成。正如@Tim 所说，一旦计算完成，您可以通知主要参与者。
另一种选择是让路由器后面的另一个参与者进行计算，同时将路由的数量限制为 CPU 的数量。
一个简单的示例：
object DelegatingSystem extends App {
val sys = ActorSystem("DelegatingSystem")
case class TimeConsuming(i: Int)
case object Other
class Worker extends Actor with ActorLogging {
override def receive: Receive = {
case message =>
Thread.sleep(1000)
log.info(s"$self computed long $message")
}
}
class Delegator extends Actor with ActorLogging {
//Set the number of routees to be equal to #of cpus
val router: ActorRef = context.actorOf(RoundRobinPool(2).props(Props[Worker]))
override def receive: Receive = {
case message:TimeConsuming => router ! message
case _ =>
log.info("process other messages")
}
}
val delegator = sys.actorOf(Props[Delegator])
delegator ! TimeConsuming(1)
delegator ! Other
delegator ! TimeConsuming(2)
delegator ! Other
delegator ! TimeConsuming(3)
delegator ! Other
delegator ! TimeConsuming(4)
}
【讨论】：