FSM-s를 사용하여 Akka 액터가 장기 실행 작업을 수행하는 방법을 나타냅니다.

Question

FSM과 함께 설명 된 내 액터가 (유휴 상태에서) 트리거를 기다리고 있습니다. 도착하면 일부 데이터를 처리하기 시작하고 (실행 상태가 됨) 완료되면 다시 유휴 상태가됩니다.

FSM 모델을 올바르게 이해 한 경우,이 관점에서 가공 시작 (유휴 -> 실행 중) 및 처리 완료 (실행 -> 유휴)라는 두 가지 이벤트가있었습니다.

배우의 관점에서 보면 단 하나의 메시지 만있었습니다.

하나의 가능성은 처리 자체를 다른 액터로 위임하는 것입니다. 그래서 트리거링 이벤트를 전달하고 Running 상태로 전환 한 다음 결과가 나오면 Idle로 이동합니다. FSM 자체가 요청에 신속하게 대응할 수있는 이점이 있지만 (예 : 현재 상태를 묻는 등) 설계가 더욱 복잡해집니다.

액터가 처리를 완료했을 때 완성 된 메시지를 자기에게 보내면 실행 -> 유휴 상태 전환이 발생하지만 좀 이상하게 보입니다.

다른 옵션에는 어떤 것이 있습니까?

참고 : FSM 모델을 고수하기 위해 다양한 전환이있는 여러 다른 상태가 있습니다. 당신이 처리를 수행하고 FSM으로 전환 할 필요가있는 배우를 가지고있는 것처럼, 나는 당신이이 지침을 따르는 것이 좋습니다 보인다 때문에

Answer 1

(일부 기본 코드 윤곽이 목록은 다음과) :

• 배우 A는 수신 일부 처리를 트리거하는 해당 메시지
• 액터 A는 별도의 FSM 액터, 예를 들어 F (akka.actor.FSM)를 적절한 상태로 전환합니다. 시작시 액터 A는 해당 컨텍스트 (예 : 모든 트랜잭션 또는 트랜잭션 당 상태 또는 기타 컨텍스트)의 상태를 추적하기 위해 특정 FSM을 생성합니다. 아래의 코드 개요는 모든 처리 또는 완료된 트랜잭션을 예제의 컨텍스트로 사용하지만 변경해야 할 수도 있습니다.
• 액터 A는 메시지에 대해 트리거되어야하는 처리를 트리거합니다. 배우가 일반적으로 차단해서는 안되지만, 여기 when an Akka actor may block에 대한 더 많은 가이드 라인을 제공하는 답변이 있습니다.
• 대체 : 블로킹없이 장시간 실행되는 처리를 트리거하고 상대방이 처리 단계를 거쳐 필요한 이벤트를 받도록하려면 프런트 액터 A를 제거하고 FSM 액터 F 만 가질 수 있습니다. onTransition 이 경우에. 그래서 내 코드 개요 제안은 내가 질문을 이해 무엇을 기반으로

은 다음과 같습니다

/* Events */ 
sealed trait MyEvents 
case class ProcessingStarted(txnId: Long) extends MyEvents 
case class ProcessingFinished(txnId: Long, result: Result) extends MyEvents 

/* Valid states for your FSM */ 
sealed trait MyStates 
case object Idle extends MyStates 
/* Constructor arguments could be anything, I randomly chose a Long for a transaction ID which may be specific to a job */ 
case object Processing extends MyStates 
/* define more case classes or objects depending on the rest of the states */ 

/* Valid internal state data types for FSM */ 
sealed trait MyDataTypes 
case object Uninitialized extends MyDataTypes 
case class StateData(processingIds: Seq[Long], resultMap: Map[Long, Result]) extends MyDataTypes 

import akka.actor.{ Actor, ActorRef, FSM } 
import scala.concurrent.duration._ 

class ActorF extends FSM[MyStates, MyDataTypes] { 
    startWith(Idle, Uninitialized) 

    when(Idle) { 
    case Event(ProcessingStarted(txnId), Uninitialized) => 
     goto(Processing) using StateData(Seq(txnId), Map.empty[Long, Result]) 
    case Event(ProcessingStarted(txnId), StateData(emptyIds, resultMap)) => 
     goto(Processing) using StateData(Seq(txnId), resultMap) 
    } 

    when(Processing) { 
    case Event(ProcessingFinished(txnId, result), StateData(processingIds, resultMap)) => { 
     val remainingIds = processingIds diff Seq(txnId) 
     val newResultMap = resultMap + (txnId -> result) 
     if (remainingIds.isEmpty) { 
     goto(Idle) using StateData(remainingIds, newResultMap) 
     } else { 
     stay using StateData(remainingIds, newResultMap) 
     } 
    } 
    } 

    initialize() 
} 

// inside Actor A do something like this for creating the FSM actor (just like other actors) 
val f = system.actorOf(Props(classOf[ActorF], this)) 

// send an event message to it just like other types of actors 
f ! ProcessingStarted(txnId) 

당신이 onTransition를 사용할 수있는 코드의 다른 부분에 대한 처리 요청을 비 차단 해제 트리거하도록 선택하는 경우 필요에 따라 트리거 코드를 추가하십시오. MyEvents 케이스 클래스를 다른 시제로 업데이트 할 수도 있습니다. 위에서 사용 된 이벤트 이름 지정은 다른 이벤트가 그 이벤트를 트리거하는 데 책임이 있음을 보여주는 것입니다 (예 : 행위자 A가 일부 작업을 수행하기 위해 초기 메시지를 받음).

Supervision capabilities of Akka 또한 여기에서 배우의 감독에 사용할 수 있습니다.

자세한 내용은 FSM을 작성하고 테스트하고 비 차단 방법을 사용하여 외부에서 장시간 실행되는 처리를 트리거하는 데 도움이되는 다음을 읽어보십시오. 이는 모든 사용자의 요구에 유용 할 수 있습니다 : 상세한 답변

• Scala docs for Akka 2.2.x: FSMs
• Scala docs for Akka 2.2.x: Testing
• Scala docs for Akka 2.2.x: Akka IO for non-blocking
• Scala docs for Akka 2.2.x: ZeroMQ