메시지를 연기하는 가장 좋은 방법은 무엇입니까?

아래에서 요구 사항을 충족하도록 액터를 구성하는 한 가지 가능한 방법을 찾을 수 있습니다. 이 솔루션에서는 프로젝트별로 액터 인스턴스를 사용하여 해당 프로젝트에 특정한 항목을 캐시합니다. 그런 다음 프로젝트 항목을 가져오고 해당 프로젝트의 캐시를 처리하는 올바른 자식 액터에 위임하는 요청을 수신하는 라우팅 액터를 사용합니다. 실제 캐싱 행위자에서 캐시 할 항목이로드 될 때까지 요청을 연기하기 위해 stash / unstash를 사용했음을 알 수 있습니다 (코드에서 시뮬레이션 중입니다). 코드는 다음과 같습니다.

import akka.actor._
import scala.concurrent.Future
import akka.pattern._
import concurrent.duration._ 
import akka.util.Timeout

class ItemProviderRouter extends Actor{
  import ItemProvider._

  def receive = {
    case get @ GetItems(project) =>

      //Lookup the child for the supplied project.  If one does not
      //exist, create it
      val child = context.child(project).getOrElse(newChild(project))
      child.forward(get)
  }

  def newChild(project:String) = {
    println(s"creating a new child ItemProvider for project $project")
    context.actorOf(Props[ItemProvider], project)
  }

}

object ItemProvider{
  case class GetItems(project:String)
  case class Item(foo:String)
  case class LoadedItems(items:List[Item])
  case object ClearCachedItems
  case class ItemResults(items:List[Item])
}

class ItemProvider extends Actor with Stash{
  import ItemProvider._  

  //Scheduled job to drop the cached items and force a reload on subsequent request
  import context.dispatcher
  context.system.scheduler.schedule(5 minutes, 5 minutes, self, ClearCachedItems)

  def receive = noCachedItems

  def noCachedItems:Receive = {
    case GetItems(project) =>
      stash()      
      fetchItems(project)
      context.become(loadingItems)


    case ClearCachedItems =>
      //Noop
  }

  def loadingItems:Receive = {
    case get:GetItems => stash

    case LoadedItems(items) =>
      println(s"Actor ${self.path.name} got items to cache, changing state to cachedItems")
      context.become(cachedItems(items))
      unstashAll()    

    case ClearCachedItems => //Noop      
  }

  def cachedItems(items:List[Item]):Receive = {
    case GetItems(project) =>
      sender ! ItemResults(items)

    case ClearCachedItems =>
      println("Clearing out cached items")
      context.become(noCachedItems)       

    case other =>
      println(s"Received unexpected request $other when in state cachedItems")          
  }

  def fetchItems(project:String){
    println(s"Actor ${self.path.name} is fetching items to cache")

    //Simulating doing something that results in a Future
    //representing the items to cache    

    val fut = Future{
      Thread.sleep(5000)
      List(Item(s"hello $project"), Item(s"world $project"))
    }

    fut.map(LoadedItems(_)).pipeTo(self)
  }
}

그런 다음 테스트하려면 :

object ItemProviderTest extends App{
  import ItemProvider._
  val system = ActorSystem("test")
  import system.dispatcher
  val provider = system.actorOf(Props[ItemProviderRouter])

  implicit val timeout = Timeout(10 seconds)
  for(i <- 1 until 20){
    val afut = provider ? GetItems("a")
    val bfut = provider ? GetItems("b")

    afut onSuccess{
      case ItemResults(items) => println(s"got items list of $items for project a")
    }

    bfut onSuccess{
      case ItemResults(items) => println(s"got items list of $items for project b")
    }    
  }
} 

간단하게하기 위해 사용자 지정 라우터가 아닌 실제 행위자를 사용하여 라우팅을 수행하지만 성능 (즉, 사서함 히트)이 중요하다면 여기에서 사용자 지정 라우터를 구현할 수도 있습니다.