행위자 모델

1. 개요

• 새로운 행위자를 만든다
• 자신 혹은 다른 행위자에게 메시지를 발송한다
• 다음에 수신할 메시지에 대해 취할 행위를 정한다

2. 예제

2.1. Scala + Akka

#!syntax java
object Example {
  import akka.actor.{Actor, ActorRef, Props}

  object MapReduce {
    lazy val sys = akka.actor.ActorSystem()

    def apply[A, B, C](map: A => B, nrOfMapActors: Int,
                       reduce: (C, B) => C, state: C,
                       io: C => Unit) =
      sys.actorOf(Props(
        classOf[MapReduce[A, B, C]],
        map, nrOfMapActors,
        reduce, state,
        io))
  }

  class MapReduce[A, B, C](map: A => B, nrOfMapActors: Int,
                           reduce: (C, B) => C, state: C,
                           io: C => Unit) extends Actor {
    import context.actorOf

    lazy val ioActor = actorOf(Props(classOf[IO[C]], io))
    lazy val reduceActor =
      actorOf(Props(classOf[Reduce[B, C]], reduce, state, ioActor))
    lazy val mapActor =
      if (nrOfMapActors < 2)
        actorOf(Props(classOf[Map.Single[A, B]], map, reduceActor))
      else
        actorOf(Props(classOf[Map.Multiple[A, B]], map, nrOfMapActors, reduceActor))

    def receive = {
      case elems: Seq[A] =>
        for (elem <- elems) mapActor ! Message(elem)
    }
  }

  object Map {
    class Single[A, B](map: A => B, reduceActor: ActorRef) extends Actor {
      def receive = {
        case msg: Message[A] =>
          reduceActor ! Message(map(msg.contents))
      }
    }

    class Multiple[A, B](map: A => B,
                         nrOfActors: Int,
                         reduceActor: ActorRef) extends Actor {
      import akka.routing.{Router, RoundRobinRoutingLogic, ActorRefRoutee}
      import akka.actor.Terminated
      import context.{watch, actorOf}

      var router = Router(
        RoundRobinRoutingLogic(),
        for (_ <- 1 to nrOfActors)
          yield ActorRefRoutee(single)
      )

      def single =
        watch(actorOf(Props(classOf[Single[A, B]], map, reduceActor)))

      def receive = {
        case msg: Message[A] =>
          router route (msg, reduceActor)
        case Terminated(one) =>
          router = (router removeRoutee one) addRoutee single
      }
    }
  }

  class Reduce[A, B](reduce: (B, A) => B,
                     var state: B,
                     ioActor: ActorRef) extends Actor {
    def receive = {
      case msg: Message[A] =>
        state = reduce(state, msg.contents)
        ioActor ! Message(state)
    }
  }

  class IO[T](io: T => Unit) extends Actor {
    def receive = {
      case msg: Message[T] => io(msg.contents)
    }
  }

  case class Message[T](contents: T)
}

object Main extends App {
  type II = (Int, Int)

  val map: Seq[Int] => II = _.foldLeft (0, 0) {
    (acc, n) => (acc._1 + 1, acc._2 + (n & 1)) }
  val reduce: (II, II) => II = {
    case ((counted, odds), (c, o)) => (counted + c, odds + o) }
  val io: II => Unit = {
    case (counted, odds) => println(s"Counted: $counted  Odds: $odds") }

  val nums = {
    val rng = new java.security.SecureRandom
    Stream.fill(10000000){rng.nextInt()}.grouped(100).toSeq }
  val mapReduce = Example.MapReduce(map, 8, reduce, (0, 0), io)

  mapReduce ! nums
}

3. 순수 이론으로서의 양가성

4. 관련 개념

4.1. CSP