Dispatcher는 ActorSystem 내에서 실행되는 모든 코드를 스케줄링 합니다.

각 Actor의 처리량과 시간 점유율을 조정하여 각자에게 공정한 리소스를 제공합니다.

구성 변경을 하지 않는한 일반적으로 시나리오에 맞게 최적화 된 .Net ThreadPOOL을 사용합니다.

Dispatcher 설정기능을 제공함으로 , 액터그룹별로 다른 성능목표를 가지고 , 성능전략을 세울수가 있습니다.

AKKA를 사용하기 위해서는 스레드를 직접 생성하고 복잡한 스레드모델을 사용할 필요가 없다라고 앞에서

설명을 하였지만 메시지 전송부분에서만 그렇고 이번장에 있어서는 Dispatcher를 잘 활용하고 이해하기위해서는

멀티스레드 개념을 익혀둘필요가 있습니다.

C#이 지원하는 비동기 동시처리 프로그래밍에서 ASYNC ,AWAIT,TASK등의 지원으로 ( .net 45+)

닷넷프레임워크 내에서도 직접 스레드 생성하는것을 지양합니다.

C#이 지원하는 순수 동시처리 는 파이프 를 통해 ACTOR와 휼륭하게 연동이 될수있습니다.

`Dispatcher`

custom-dispatcher {
  type = Dispatcher
  throughput = 100
}

.NET ThreadPool 을 통해 작동이되며 대부분의 경우 이것만으로 충분합니다.

몇개의 스레드를 둘것인가? 보다 몇개의 메시지를 동시에 처리하게 할것인가? 에대한 설정입니다.

TaskDispatcher

custom-task-dispatcher {
    type = TaskDispatcher
    throughput = 100
}

TPL 인프라를 사용합니다. ThreadPool과 유사하지만 병렬처리 Thread를 사용해야하는

특수한 케이스일때 사용합니다. 대부분 신경쓸필요없이 프레임워크가 다중코어처리를 잘 수행하나

병렬처리가 꼭 유용한 경우, 액터설정을 할수가 있습니다.

`PinnedDispatcher`

custom-dedicated-dispatcher {
    type = PinnedDispatcher
}

AKKA가 준비한 ,단일 전용 스레드만 사용합니다. 어떠한 목적을 두고 준비된 Dispatcher는 아니며

서비스 메시지 처리와, 시스템 성능 메시지처리를 분리한다고 했을시

성능확인을 위해 그리많은 메시지처리가 되지 않고, 서비스 메시지처리와 분리하여 작동시키고자 할때 활용될수 있습니다.

ForkJoinDispatcher

            my-dispatcher {
              type = ForkJoinDispatcher 
              throughput = 100
              throughput-deadline-time = 0ms
              dedicated-thread-pool {
                thread-count = 3
                deadlock-timeout = 3s
                threadtype = background
              }
            }

대부분 Dispatcher는 목적에맞게 최적화된 각자의 스레드풀 메카니즘이 있고 신경쓸필요없으나

ForJoinDispatcher의 경우 전용 스레드풀을 생성하여, 스레드수를 직접 할수있는 유일한 Dispatcher입니다.

이 스케줄러를 이용할시, 세밀한 튜닝 전략으로 일부 액터를 분리할수가 있습니다.

`SynchronizedDispatcher`

synchronized-dispatcher {
    type = "SynchronizedDispatcher"
    throughput = 10
}


private void Form1_Load(object sender, System.EventArgs e)
{
    system.ActorOf(Props.Create<UIWorker>().WithDispatcher("synchronized-dispatcher"), "ui-worker");
}

 SynchronizedDispatcher는 SynchronizationContext 를 사용하며 Actor가 UI를 업데이트를 한다고하면

UI를 업데이트할수있는 전용 Dispatcher 입니다. Actor는 UI Thread내에서 작동되는 녀석이 아니며

Actor와 별개로 대부분 타 스레드가 UI 객체를 변경하고자 할때 문제가 생깁니다. ( WPF,WINFORM 모두 동일)

이러한 문제에대해 단순한 처리는 UI 스레드내에서 어떠한 작업을 진행하는것인데, 그러면 UI가 멈춰서 좋은 UX를 제공하기 불가합니다.

관리툴 목적으로, 네이티브한 UI 를 만드는 경우 활용할수 있습니다.

TEST1-액터내에서 블락킹되는경우

지연액터

    public class ReActor : ReceiveActor
    {
        private ILoggingAdapter log = Context.GetLogger();

        public ReActor()
        {
            Receive<string>(message => {

                if(message == "slow")  //slow메시지를 받으면, 지연시킵니다.(테스트 지연코드)
                {
                    Task.Delay(500).Wait();
                }

                string reply = string.Format("I'am {0} RE:{1}", Self.Path, message);
                log.Info(reply);
                Sender.Tell(reply);
            });            
        }
    }

액터3개에게 동시에 100개의 메시지를 보냄

            var actor1 = actorSystem.ActorOf(Props.Create<ReActor>().WithDispatcher("my-dispatcher"), "my-actor1");
            var actor2 = actorSystem.ActorOf(Props.Create<ReActor>().WithDispatcher("my-dispatcher"), "my-actor2");
            var actor3 = actorSystem.ActorOf(Props.Create<ReActor>().WithDispatcher("my-dispatcher"), "my-actor3");

            for (int i=0; i < 100; i++)
            {
                actor1.Tell("slow");
                actor2.Tell("slow");
                actor3.Tell("slow");
            }

            Console.WriteLine("completed - send all");

Akka의 로그는 기본적으로 고유 스레드 아이디를 표시해줍니다.

TestResult1 : type = ForkJoinDispatcher thread-count =1

스레드 1개만 사용이 됩니다.

TestResult1 : type = ForkJoinDispatcher thread-count =3

스레드 3개를 번갈아가며 사용합니다.

대부분 사용할일이 없지만, 어떠한 액터가 메시지를 받으면 블락킹이되고 오래걸리는경우 (일반적으로 외부 RestAPI)

이와같은 전력을 통해 스레드수분리가 가능합니다. ( 기본 메시지처리기에 부하를 분리)

적용2-ACTOR내에서 논블럭킹처리

RestAPI를 대신하여 호출해주는 Actor 설계시 다음과 같은 조건을 가집니다.

결과에따른 호출 순서 상관없음
Actor Task에서 분리되어 , RestAPI 결과값을 반환해야함
성능을 위해 동시에 실행

C#의 비동기 프로그래밍의 힘을 빌려 Task에 완료된 비동기처리를 Pipe로 연결하여

결과가 완료되면, 완료 처리를 하게 합니다.

    public class DelayReply
    {
        public string message;
    }

    public class ReActor : ReceiveActor
    {
        private ILoggingAdapter log = Context.GetLogger();

        public ReActor()
        {
            string myPath = Self.Path.ToString();

            Receive<string>(message => {
                Handle(message);                
            });

            Receive<DelayReply>(message => {
                Handle(message);
            });
        }

        public void Handle(string str)      //InMessage
        {
            Task.Run(async () =>
            {
                await Task.Delay(1000); //어떠한 값을 기다림
                DelayReply reply = new DelayReply();
                reply.message = str;
                return reply;
            }).PipeTo(Self);
        }

        public void Handle(DelayReply data) //Out
        {
            string logtrace = string.Format("I'am {0} RE:{1}", Self.Path, data.message);
            log.Info(data.message);
            Sender.Tell(data);
        }
                
    }

Content

Space Tools