akka 的 Actor 的接收方法与 Future（块）交互 - 新消息可以在 Future 完成之前传入吗？

Question

我编写了以下代码，以便在 Future 中调用一些异步工作时尝试确定参与者对传入消息的行为（实际上，异步工作是为了模拟 Slick 的异步数据库 API):

  val actor = system.actorOf(Props[WTFAsyncActor])

  for (i <- 1 until 100) {
    actor ! 'wtf + i.toString
    Thread.sleep(200l) // yes I know actors should not invoke this method
  }
  Thread.sleep(10000l) // same as above

Actor的代码如下：

class WTFAsyncActor extends Actor with ActorLogging{
  import scala.concurrent.ExecutionContext.Implicits.global
  override def receive: Receive = {

    case i@_ =>

      log.info(s"[$i] external block pre future")
      Thread.`yield`() // just trying to relinquish control to induce context switching

      Future {
        log.info(s"[$i] internal block pre sleep")
        Thread.sleep(1000l) // yeah bad - trying to emulate something meaningful like slick
        log.info(s"[$i] internal block post sleep")
      }

      log.info(s"[$i] external block post future")
  }
}

我得到以下日志（样本摘录运行）

10:17:58.408 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf1] external block pre future
10:17:58.420 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf1] external block post future
10:17:58.421 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf1] internal block pre sleep
10:17:58.599 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf2] external block pre future
10:17:58.600 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf2] external block post future
10:17:58.600 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf2] internal block pre sleep
10:17:58.800 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf3] external block pre future
10:17:58.801 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf3] external block post future
10:17:58.801 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf3] internal block pre sleep
10:17:59.001 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf4] external block pre future
10:17:59.002 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf4] external block post future
10:17:59.002 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf4] internal block pre sleep
10:17:59.202 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf5] external block pre future
10:17:59.206 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf5] external block post future
10:17:59.402 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf6] external block pre future
10:17:59.403 [wtf-async-test-akka.actor.default-dispatcher-2] INFO  org.wtf.test.WTFAsyncActor - ['wtf6] external block post future
10:17:59.421 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf1] internal block post sleep
10:17:59.421 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf6] internal block pre sleep
10:17:59.607 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf2] internal block post sleep
10:17:59.607 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf5] internal block pre sleep
10:17:59.608 [wtf-async-test-akka.actor.default-dispatcher-3] INFO  org.wtf.test.WTFAsyncActor - ['wtf7] external block pre future

我认为可以肯定地说，只要某个线程可用，新消息就会在新消息传入时被处理，而不管 Future 块尚未完成执行。我对吗？

我要揭示的是接收块是否必须等待 Future 块在处理新消息之前完成。 好像没有。（例如，在 wtf1 完成线程 3 上的计算之前，wtf2 进入调度程序线程 3）

对这种行为有什么注意事项吗？

请原谅这个看似愚蠢的问题。我没有深入了解 akka 代码库（目前我对 scala 和 akka 还太陌生）

Answer 1

在任何 Future 完成之前，receive 可以处理后续消息。这种行为实际上没有任何警告，但有一些事情需要注意。特别是，不要关闭 Future 中 actor 的任何可变状态。 Future 可能与 actor 正在处理的消息同时执行，打破了单线程访问 actors 状态的保证，并导致竞争条件，这可能会使您的 actor 状态处于无效位置。

如果您想从 Actor 中异步启动某些工作，常见的模式是启动子 Actor 来完成工作。子 actor 可以将任何结果的消息发送回其父 actor，并且您肯定知道子 actor 不能干扰父 actor 的状态。

Answer 2

没有针对此行为的警告。在 actor 中创建的 future 和在其他地方创建的 future 在行为上是相同的：它们创建一个异步计算，该计算将 return 在稍后的某个时间产生成功或失败的结果。重要的是，无法保证未来将执行哪个线程。执行 actor 的接收块的同一个线程很可能接下来会被分配来处理未来，然后再次分配给接收块，在这种情况下，一切似乎都是同步执行的。但除非您是运行单线程，否则这种情况不太可能经常发生。

akka 的 Actor 的接收方法与 Future（块）交互 - 新消息可以在 Future 完成之前传入吗？

akka's Actor's receive method interaction with a Future (block) - can new messages come in before Future completes?

scala

asynchronous

akka

future