상태기계 : 한 번에 한 개의 상태를 반영하는 기계 ex) 커피 자동 판매기
커피 한 잔이 1,000 원 이라고 할 때 동전을 집어넣지 않은 자판기는 초기상태를 반영하고 있다.(initial)
500원짜리 동전을 하나 집어넣으면 자판기는 500원이 입력된 상태를 반영한다.
그 상태에서 500원을 더 넣으면 자판기는 이제 커피를 내보낼 준비가 되었다.(ready)
혹은 초기상태에서 1,000원을 넣으면 중간단계를 생략하고 곧바로 준비상태가 된다.
이러한 과정을 간단히 그림으로 나타내면 아래와 같다.
타원은 각각의 상태를 나타내고, 직선은 사건을 의미한다. 이것이 상태기계다.
메서드를 동기적인 방식으로 직접 호출하는 대신
메시지를 전달하는 비동기적인 방식으로만 통제할 수 있는 액터 세계에서는 상태기계를 이용해서 해결할 수 있는 문제가 많다.
그래서 아카는 상태기계를 구현할 수 있는 편리한 API를 제공한다.
앞에서 보았던 PingActor의 상태는 대략 네 개로 구분할 수 있다.
-
아직 아무 메시지도 전달되지 않은 초기 상태(initial)
-
"work" 메시지가 전달되어 초기화가 시작된 상태(zeroDone)
-
"done" 메시지가 한 번 전달된 상태(oneDone)
-
"done" 메시지가 두 번 전달되어 작업이 완료된 상태(allDone)
각각의 상태들에 이름을 붙여 그림과 같은 상태기계로 표현할 수 있다.
개념적으로 초기상태(initial) 과 최종상태(allDone) 은 동일한 상태로 보아도 무방하다.
아래의 상황을 가정해보자.
-
PingActor가 Main으로부터 “work” 메시지를 수신했다.
-
PingActor가 Ping1Actor에게 “work”를 전송했다.
-
PingActor가 “done” 메시지를 수신했다.
-
PingActor가 Main으로부터 “work” 메시지를 수신했다.
-
PingActor가 “done” 메시지를 수신했다.
각 단계별로 상태가 어떻게 변하는지 설명하면 다음과 같다.
-
아직 변화 없음.
-
initial --> zeroDone
-
zeroDone --> oneDone
-
??
-
oneDone --> allDone
문제는 현재 상태가 "oneDone" 일 때 기다리고 있는 "done"이 아니라 "work"를 수신한다면 어떠한 동작을 취해야 하는가이다.
정답은 PingActor가 그 메시지를 메일박스 어딘가에 저장 해두고 아무런 일도 하지 않으면 된다.
커피 판매기에서 500원을 넣은 후 커피 추출 버튼을 눌렀을 때 아무런 일이 일어나지 않는 것과 동일한 이치이다.
500원이 입력된 판매기는 커피를 내보내라는 버튼이 아니라 500원 추가라는 메시지를 기다리기 때문에 다른 메시지에 대해서는 반응하지 않는다.
"done" 메시지가 한 번 도착한 PingActor는 oneDone인 상태에서 추가적인 "done" 메시지를 기다리기 때문에 "work"라는 메시지에 대해서는 반응을 보이지 않는다.
다만 그 메시지는 상태가 바뀌었을 때 처리하기 위해 메일박스에 잘 보관해둔다.
이번 예제 소스는 아카가 제공하는 become, unbecome API를 이용해서 이와 같은 동작을 수행하는 상태기계를 구현한다.
// main 클래스
import akka.actor.{ActorSystem, Props}
import org.study.actor.PingActor
object Main {
def main(args: Array[String]) = {
val actorSystem = ActorSystem.create("TestSystem")
val ping = actorSystem.actorOf(Props.create(classOf[PingActor]))
ping ! "work"
ping ! "reset"
}
}
// become / unbecome API를 이용하여 상태기계를 구현하는 PingActor 클래스
import akka.actor.{Actor, Props, UntypedActorWithStash}
import akka.event.Logging
/**
* 아카의 상태기계를 구현한 액터
*/
class PingActor extends UntypedActorWithStash {
val log = Logging(context.system, this)
val child = context.actorOf(Props.create(classOf[Ping1Actor]), "ping1Actor")
context.become(initial)
def onReceive(message: Any) = { }
/** 맨 처음 상태에서 "work" 메시지를 받으면 zeroDone 상태가 된다. */
def initial : Actor.Receive = {
case msg@"work" =>
child ! msg
context.become(zeroDone)
case _ => stash()
}
/** zeroDone 상태에서 "done" 메시지를 받으면 oneDone 상태가 된다. */
def zeroDone : Actor.Receive = {
case msg@"done" =>
log.info("Received the first done")
context.become(oneDone)
case _ => stash()
}
/** oneDone 상태에서 "done" 메시지를 받으면 allDone 상태가 된다. */
def oneDone : Actor.Receive = {
case msg@"done" =>
log.info("Received the second done")
unstashAll()
context.become(allDone)
case _ => stash()
}
/** allDone 상태에서 "reset" 메시지를 받으면 다시 initial 상태가 된다. */
def allDone : Actor.Receive = {
case msg@"reset" =>
log.info("Received a reset")
context.become(initial)
case _ => stash()
}
}
// "work"라는 메시지를 받으면 Ping2Actor, Ping3Actor에게 전송해주는 역할을 수행하는 Ping1Actor 클래스
import akka.actor.{Actor, Props}
import akka.event.Logging
class Ping1Actor extends Actor {
val log = Logging(context.system, this)
val child1 = context.actorOf(Props.create(classOf[Ping2Actor]), "ping2Actor")
val child2 = context.actorOf(Props.create(classOf[Ping3Actor]), "ping3Actor")
override def receive = {
case msg : String =>
log.info(s"Ping1Actor received $msg")
child1.tell(msg, sender)
child2.tell(msg, sender)
}
}
// 비즈니스 업무를 시뮬레이션 하는 Ping2,Ping3Actor 클래스
import akka.actor.Actor
import akka.event.Logging
class Ping2Actor extends Actor {
val log = Logging(context.system, this)
override def receive = {
case msg : String =>
log.info(s"Ping2Actor received $msg")
work
sender ! "done"
}
private def work = {
Thread.sleep(1000) // 실전에서는 절대 금물!!!
log.info("Ping2Actor working...")
}
}
import akka.actor.Actor
import akka.event.Logging
class Ping3Actor extends Actor {
val log = Logging(context.system, this)
override def receive = {
case msg : String =>
log.info(s"Ping3Actor received $msg")
work
sender ! "done"
}
private def work = {
Thread.sleep(1000) // 실전에서는 절대 금물!!!
log.info("Ping3Actor working...")
}
}액터는 메시지를 담는 메일박스와 별도로 상태에 따른 동작을 담아두기 위한 스택도 포함하고 있다.
자바에서는 procedure 객체를 사용하지만 스칼라에서는 함수를 사용하였다.
PingActor의 소스코드를 보면 context() 를 이용하여 become() 메서드를 호출하면 된다.
액터의 어느 장소에서 context.become(initial) 과 같이 호출하면 인수로 전달되는 initial 이라는 객체가 스택에 저장된다.
만약 unbecome 이라는 메서드를 호출하면 스택의 꼭대기에 있던 객체가 꺼내어지고 현재 상태가 되기 직전의 상태로 돌아간다.
become 메서드가 호출되지 않았거나 become 후에 unbecome이 호출되어서 동작 스택이 비어 있으면 액터는 receive 함수에 정의된 동작을 수행한다.
PingActor의 코드를 보면 initial, zeroDone, oneDone, allDone이 라는 네 개의 객체가 정의되고 있다.
PingActor는 생성자에서 become (initial)을 호출하면서 초기상태가 된다.
이로써 PingActor에서 정의 된 onReceive 메서드는 의미를 상실했다.
initial 상태는 오직 “work” 메시지만 기다린다. 나머지 메시지는 모두 stash () 에 의해서 저장된다.
stash () 메서드는 자기가 기다리고 있는 메시지가 아닌 다른 모든 메시지를 메일박스로 되돌려 보내는 동작을 수행한다. 기다리던 메시지가 전달되면 필요한 동작을 수행하고 그 다음 단계인 zeroDone 상태로 넘어간다.
이제 스택의 맨 윗자리는 initial 상태가 아니라 zeroDone 상태가 차지한다.
zeroDone 상태는 더 간단하다. 오직 “done” 메시지를 기다리고, 만약 기다리 던 메시지가 전달되면 oneDone 상태로 넘어간다.
여기에서도 마찬가지로 기다리는 메시지가 아닌 다른 메시지는 모두 stash ()에 의해서 메일박스에 저장된다.
oneDone 상태도 마찬가지다. 오직 “done” 메시지를 기다린다. 기다리던 메 시지가 전달되면 allDone 상태로 넘어간다. 기다린 메시지가 아닌 다른 메시지는 모두 stash ()에 의해서 메일박스에 저장된다.
여기에서 한 가지 주목할 부분이 있다. oneDone 상태는 그 다음 단계인 allDone으로 넘어가기 전에 unstashAll ()을 호출한다. 지금까지 메일박스에 저장해둔 메시지를 하나씩 다시 처리하라는 명령이다.
/** 맨 처음 상태에서 "work" 메시지를 받으면 zeroDone 상태가 된다. */
def initial : Actor.Receive = {
case msg@"work" =>
child ! msg
context.become(zeroDone)
case _ => stash()
}
/** zeroDone 상태에서 "done" 메시지를 받으면 oneDone 상태가 된다. */
def zeroDone : Actor.Receive = {
case msg@"done" =>
log.info("Received the first done")
context.become(oneDone)
case _ => stash()
}
/** oneDone 상태에서 "done" 메시지를 받으면 allDone 상태가 된다. */
def oneDone : Actor.Receive = {
case msg@"done" =>
log.info("Received the second done")
unstashAll()
context.become(allDone)
case _ => stash()
}
/** allDone 상태에서 "reset" 메시지를 받으면 다시 initial 상태가 된다. */
def allDone : Actor.Receive = {
case msg@"reset" =>
log.info("Received a reset")
context.become(initial)
case _ => stash()
}