8장 비동기와 동시성 (AKKA)
to 김민석
시작하며...
우리는 대화를 할때 상대방의 생각을 읽을수 없다 하지만 나의 의사는 대화로 전달 가능하고 표현 가능하다 내가 타인에게 전달한 의사는 예상 할수도 있고 그렇지 못할수도 있다 이것은 전달 받은 사람의 행동과 책임에 달려 있다 그리고 개인은 어느 누구에게 강하게 종속되지 않는다 다만 어딘가에 속해 있을 뿐이다
문득 생각이 들었다 현실 세계를 프로그래밍 한다는건 너무 어렵다는 것이다
그러면 akka를 쓰면 모두 해결되는 건가?
아직 실무에 직접적으로 적용하거나 akka만의 사고방식으로 설계를 해보지 않았지만 이번 챕터를 공부하면서 내가 생각하던 것들에 좀더 접근할수 있었고 새로운 패러다임은 열린 생각을 가지게 해주는 좋은 경험이었 다고 생각한다 제일 중요한건 기존 사고 방식의 틀을 조금은 열어 놓고 보는게 중요한것 같다
참고자료
임백준의 akka 와 동영상 자료를 참고 하였고 scala코드는 자바개발자를 위한 스칼라 책을 참고 하였습니다
동시성 모델
아카는 차세대 동시성 모델로서 기존 객체는 액터와 같다 모든 액터는 메세지로만 통신 하며 각자의 책임의 행동을 정의한다
akka는 현대 소프트웨어가 갖춰야할 4가지 를 지향 한다
- 반응성 : 사용자에게 빠르게 응답해야 한다
- 탄력성 : 시스템은 때에 따라 스케일 업,아웃이 자유롭게 가능해야한다
- 유연성 : 시스템은 문제 발생시 유연하게 대처 가능해야 한다
- 메시지중심 : 비동기 메시지 전달 및 장소 투명성
간단한 예제
ActorRef 개념이 가장 중요하다 모든 액터는 ActorRef 타입에 의해서만 객체에 접근 가능하다
중요한 개념은 코드에 주석 포함하여 설명을 첨부 하였다
import akka.actor.{ ActorRef, ActorSystem, Props, Actor, Inbox }
import scala.concurrent.duration._
case object Greet
case class WhoToGreet(who: String)
case class Greeting(message: String)
class Greeter extends Actor {
var greeting = ""
def receive = {
case WhoToGreet(who) => greeting = s"hello, $who"
case Greet => sender ! Greeting(greeting) // Send the current greeting back to the sender
}
}
//액터는 시스템 내부에 스레드 풀을 보유하고 있다 액터의 개본적인 메모리 사용량은 300~400 바이트 의 가벼운 객체유
object HelloAkkaScala extends App {
// 액터를 생성 한다 (액터 시스템은 가벼운 구조물이 아니기 때문에 하나만 만드는것이 일반적이나 때에 따라 그 이상을 만들어 사용하기도 한다)
val system = ActorSystem("helloakka")
//Props : 액터를 만드는데 필요한 구성요소를 담는 구성 파일 클래스 불변 객체이다
//인수를 여러개 넘기거나 클래스 정보만 전달 할수 있다 system.actorOf(Props(classOf[Greeter],"객체1","객체2"), "greeter") (스칼라)
//액터의이름 : 반드시 고유한 문자열 이어야 한다 똑같은 이름의 액터를 한개이상 만들수 없다 동일한 액터를 2개 만들고 싶다면
//system.actorOf(Props[객체],”액터이름1”)
//system.actorOf(Props[객체],”액터이름2”)
//위처럼 다르게 만들어야 한다 액터는 트리구조 처럼 계층 구조를 형성할수 있다
val greeter = system.actorOf(Props[Greeter], "greeter")
// Create an "actor-in-a-box"
val inbox = Inbox.create(system)
/*
ActorRef 액터 시스템의 가장 중요한 개념이며 액터를 가르키는 참조 이기때문에 actorRef 타입에 의해서만 접근될수 있다
(액터객체를 감싸고 있는 객체)
이 객체는 물리적으로 어디에 존재하는지 알수 없으며 우리는 오직 ActorRef 만 가지고 수행을 하며 이는 동일한위치의 JVM 일수도 있고
다른 JVM 일수도 있다 이말은 우리가 객체를 컨트롤 할때 신경쓰지 않아도 된다는 말이며 클러스트링 할때 위력을 나타낸다 이는 장소투명성을 보장한다
*/
// ActorRef는 tell ! 과 ask ? 를 제공한다
greeter.tell(WhoToGreet("akka"), ActorRef.noSender)
/*
getSelf() 액터 자신의 ActorRef 객체를 리턴한다 (가장 일반적인 형태)
getSender() 메시지를 보내 온 발신인 주소를 다음 액터에게 포워딩 한다
ActorRef.noSender() 개념적 null 해당하는 액터 주소 리턴 발신인 주소가 아무 의미가 없는 경우에 사용한다
*/
inbox.send(greeter, Greet)
// Wait 5 seconds for the reply with the 'greeting' message
val Greeting(message1) = inbox.receive(5.seconds)
println(s"Greeting: $message1")
// Change the greeting and ask for it again
greeter.tell(WhoToGreet("typesafe"), ActorRef.noSender)
inbox.send(greeter, Greet)
val Greeting(message2) = inbox.receive(5.seconds)
println(s"Greeting: $message2")
val greetPrinter = system.actorOf(Props[GreetPrinter])
// after zero seconds, send a Greet message every second to the greeter with a sender of the greetPrinter
system.scheduler.schedule(0.seconds, 1.second, greeter, Greet)(system.dispatcher, greetPrinter)
}
// prints a greeting
class GreetPrinter extends Actor {
def receive = {
case Greeting(message) => println(message)
}
}
1. Actor
(그림 준비중..)
- 아카 디스패처가 쓰레드를 할당한다
- 쓰레드는 actor마다 1개만 생성된다
- 각 actor마다 매일박스를 가지며 요청을 순서대로 담는다
- 매일박스의 처리 요청을 onReive가 처리한다
- 처리가 끝나 시간이 지나면 생성된 쓰레드 풀로 회수되어 재사용 된다
장점: 멀티쓰레드 환경에서 안정성 보장 단점: onReive 스코프 에서 다시 동시성 영향을 받을수 있다 (새로운 쓰래드 생성 하여 다른 객체를 생성 및 변경 한다거나 등등..)
2. 메시지 순서
아카는 메시지로 명령 하고 행동한다 중요한 점은 순서를 생각하고 노드를 설계한다면 잘못된 예외를 만날수 있다
(그림 준비중..)
- A 가 B 에게 $기호를 보낸다
- A 가 B 에게 %기호를 보낸다
- A 가 C 에게 #기호를 보낸다
- B 가 C 에게 &기호를 보낸다
1번2번은 순서가 보장됨 2번3번은 순서 보장할수 없음
3. 고장을 허용 해라
상황 1
전통 적인 java 에서는 예외상황을 런타임과 체크되는 예외를 를 이용하여 처리한다 하지만 여기에도 문제가 있을수 있다
공유변수 i;
try{
i += 1; // <-- 변경
예외발생함수(); // <-- 예외발생
catch(E e){
//예외확인 및 처리
}
위 코드의 문제점은 변경한 상태를 복구 할수 없다는 점이다 물론 잘 만든다면 극복할수 있지만 개발자가 실수할 확율이 높다
상황 2
client <---> server --> [supervisor]
server 에서의 고장을 client 가 책임지는 상황은 좋지 못하다 server 내의 supervisor 가 적절히 처리하여 알려주는게 client 와 의존성을 줄이고 해당 역활에 충실할수 있다
akka 에서는 supervisor 라는 기능을 제공한다 보통 부모 노드가 supervisor 가 된다
기능
- resume: 에러무시 하고 처리
- restart: 상태를 초기화후 다시시작 (gc대상이된다)
- stop: 액터를 극단적으로 kill한다(gc대상이된다)
- esclat: 최상위 부모에 책임을 전가
책임
- oneforone : 한노드 책임
- oneforMany : 연대 책임
supervisor 의 기능은 자식의 잘못을 부모가 책임지게 만드는 것이다 해당 기능을 이용하여 고장에 대처 할수 있으며 기존 데이터를 유지하거나 초기화 전력을 취할수도 있다 이에 대한 전략은 한개의 노드나 연대적으로 물을수 있다 하지만 연대 책임의 경우에는 하위 모든 노드 에게 영향을 주기 때문에 신중하게 고려 해야한다
4. 상태기계
시스템을 만들때 우리는 상태와 상황에 따라 분기처리 를 하거나 시스템 목적 달성 여부를 판단한다
akka 는 비동기 처리 이기 때문에 상태에 따른 전략이 필요하다 서로의 상태를 유기적으로 알려주어 다음 처리를 효과적으로 하기 위함이다
이론적으로는 변환된 행동을 스택에 저장후 해당 상태에 맞는 행동을 실행 할수 있도록 도와준다
최초로 생성된 ActorSystem 는 context() 로 접근 가능하다 부모 노드는 각 자식 노드의 상태에 따라 행동이 분기 되어야 한다고 가정하면
context().become(다음행동할함수);
처리를 통해 다른 노드가 부모 노드 방문시 "다음행동할함수" 로 이동하게 된다
반대의 기능인 unbecome(); 도 존재한다
package test.actor01
import akka.actor.Actor
import akka.actor.ActorRef
import akka.event.LoggingReceive
object TravelAgent {
case class BookTrip(transport: ActorRef, accomodation: ActorRef, nbOfPersons: Int)
case object Done
case object Failed
}
class TravelAgent extends Actor {
import TravelAgent._
def receive = LoggingReceive {
case BookTrip(flightAgent, hotelAgent, persons) =>
hotelAgent ! Hotel.BookRoom(persons)
flightAgent ! Flight.BookSeat(persons)
//상태변경 다음 방문때 receive 가 아닌 awaitTransportOrAccomodation 함수에 처리됨
context.become(awaitTransportOrAccomodation(flightAgent, hotelAgent,sender))
}
//다시 한번 상태에 따라 처리 함수가 변경 됨을 볼수 있다
def awaitTransportOrAccomodation(transport: ActorRef, accomodation: ActorRef, customer: ActorRef): Receive = LoggingReceive {
case Flight.Done =>
println("1")
context.become(awaitAccomodation(customer))
case Hotel.Done =>
println("2")
context.become(awaitTransport(customer))
case Flight.Failed | Hotel.Failed =>
customer ! Failed
context.stop(self)
}
def awaitAccomodation(customer: ActorRef): Receive = LoggingReceive {
case Hotel.Done =>
customer ! Done
println("Hotel.Done")
context.stop(self)
case Hotel.Failed =>
customer ! Failed
context.stop(self)
}
def awaitTransport(customer: ActorRef): Receive = LoggingReceive {
case Flight.Done =>
println("Flight.Done")
customer ! Done
context.stop(self)
case Flight.Failed =>
customer ! Failed
context.stop(self)
}
}
5. 라우터
(그림 추가 예정 ..)
최상위 노드가 라우터 역활을 하는 노드에게 병렬 처리에 대한 요청을 할경우 사용할수 있다
라우터 actor 은 여러 개의 노드를 생성 하여 처리 가능하다
기능
RandomRouter: 무작위로 한개의 노드만 처리
SmallestMailboxRouter: 각각 actor 메일박스에서 처리 되지 않은 메시지 수가 가장 적은 것을 골라 실행 (데이터의 균형 이루어짐)
BroadCastRouter: 모두에게 전송
consistenHashingRouter: 해시를 이용한 분산 처리(일관적 해시 사용)
SatterGatterFirstComplet: 모두에게 전달하여 가장 빠른 응답을 하는 객체만 리턴 하고 나머지는 무시
주의: 라우터는 에러 전략을 에스컬레이터 전략을 기본으로 사용한다 책임과 관련한 개인및 연대 책임을 잘 판단해야 한다
6. Future
akka는 비동기 이기 때문에 함수에 리턴을 기대할때 ask 란 기능을 제공한다
보통은 tell 명령으로 보내고 잊기 전략을 취하지만
처리된 데이터를 받아 다른 행동을 기대 할수 있다
기능
- onComplete: 완료후 다른 행동을 처리
- onError: 에러에 대한 처리
- timeout: 요청 처리에 오래 걸린다면 설정 할수 있음
- pipeto: 원하는 액터에 메시지를 보내 처리하게 한다
단점 : 비동기 처리이기 때문에 디버깅 어려움, 의도와 다른 타임아웃 발생, code가 명확히지 않음
7. 클러스터
akka 에서 가장 중요하다고 강조 하고 있다 클러스터 부분은 조원과 함께 스터디 하며 실습 했으면 한다
개념은 akka로 구성된 각각의 jvm 기기 는 참여를 유도하는 클러스터 설정 파일이 존재한다
A, B, C 서버 가 존재 한다면
A 시스템이 B와 C를 참여 시킨다
참여가 된 akka 시스템은 서로 통신하며 해당 노드에 요청 할수 있다
8. 마무리...
문서 정리가 아직 부족하다 중간 에 그림도 표현해야 하고 예제 코드도 공유해야 한다 그리고 클러스터 부분은 실습을 통해 내용 보충이 필요하다 기타 오타 와 설명 추가는 시간 틈틈히 보충 하도록 하겠다 참고로 위 내용은 개념적 내용에 치중해 있다