java - 推迟消息的最佳方式是什么？

Question

我有一个角色“ItemProvider”，它可以接收“getItems”消息。ItemProvider 管理项目的 Items。因此，我可以有多个请求项目 A 项目的“getItems”消息和请求项目 B 项目的其他“getItems”消息。

第一次“itemProvider”收到这样的消息时，它需要调用服务来实际获取项目(这可能需要一分钟时间，服务会返回一个 future，这样它就不会阻止参与者)。在此等待期间，其他“getItems”消息可能会到达。

项目“ItemProvider”缓存它从服务接收的“Items”。因此，在 1 分钟加载时间后，它可以立即提供元素。

我非常确定“ItemProvider”应该使用 Akka 的成为功能。但它应该如何处理无法立即服务的客户呢？

我可以想到以下选项:

1. ItemProvider 持有一个待处理消息列表。它无法提供的消息将添加到此列表中。当 ItemProvider“准备好”时，它将处理待处理的客户端

2. ItemProvider 将消息发送回其父级。并且父级将重新发出消息

3. ItemProvider 使用调度程序。并在将来再次收到消息。

4. 也许不使用become而是使用AbstractFSM类？

   有人知道实现 ItemProvider 的最佳 Akka 方法吗？

Answer 1

看看 Akka 的 Stash feature (usage example)。下面是(未经测试的)代码，用于在从服务器请求实际项目时存储 getItems 消息，然后在服务器请求完成后处理所有 getItems 消息

import akka.actor.{Actor, Stash}

class ItemProviderActor extends Actor with Stash {
  private[this] itemsOpt : Option[Items] = None

  def receive = processing

  def processing: Receive = {
    case m:GetItems => {
      if(itemsOpt.nonEmpty) {
        // respond immediately
        itemsOpt.foreach(sender() ! _)
      }
      else {
        // Stash current request and initiate cache update
        context.become(retrivingData)
        stash()

        // Will send future results of item retrieval as a message to self
        retrieveItems().pipeTo(self)
      }
    }
  }

  def retrivingData: Receive = {
    case m: Items => 

      // items are retrieved, update cache
      itemsOpt = Option(m)

      // resume normal processing
      context.become(processing)

      // put all pending item requests back to actor's message queue
      unstashAll()


    case m:GetItems => 
      // busy retrieving items, store request to serve later
      stash()
  }

  def retrieveItems() : Future[Items] = {
    ???
  }

}