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

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

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

custom-dispatcher {
  type = Dispatcher
  throughput = 100
}

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

custom-dedicated-dispatcher {
    type = PinnedDispatcher
}

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

지난 몇년간 멀티스레딩 프로그래밍을 통해 개발을 했고 문제를 풀려고 하였습니다.

하지만,5년간 운영하면서 어떠한 한계에 부딪쳤습니다. 스레드문제처리에대해 개발내에 숙련이 유지되지 않는다는점이며

단일기기에서 충분한 사용자처리 성능을 이끌수있으나, 그속에서도 대부분의 운영 로그가 스레드에관한것이고 이것을 해결하는데 골칫거리였습니다.

실제 서비스에관련된 로그가 20%라고 하면 TCP처리및 스레드에관련된 로그가 80%이상이였습니다.

로그의 복잡성은, 개발코드의 복잡성을 반영하며 문제파악및 확장이 어려워집니다.

더욱이 현재 느끼는 심각한문제는, 그렇게 쌓인 팀라이브러리가 AKKA에서 고민하는 추상화된 스레드모델에(디스페쳐) 한참 못미치고

거의 쓰일데가 없다란 것입니다.

오히려 어떠한 동시성을 위한 추상적 레벨의 라이브러리를통해, 제공하는 옵션을 어떻게 이해하고 사용해야하는가?

멀티스레드및 비동기프로그래밍을 어떻게 이해해야하고, 다시 학습하게 되는 계기가 되었습니다.

말이 길었지만 , AKKA에서는 아래와 같은 스레드 옵션을 지원합니다.

• thread-count - dispatcher 가 사용할 스레드수를 지정합니다.
• deadlock-timeout - 기본적으로 지정없으면 무한적으로 실행될수 있습니다. 너무 낮은값을 지정하면 작업 손실이 되니 신중하게 값을 설정합니다.
• threadtype -background,foreground. 두 옵션이 있으며 OS또는 플래폼 특성적이니  how .NET handles 을 참고합니다.

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");
}

    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);
            });            
        }
    }

            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");

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

스레드 3개가 번갈아가며 일을합니다.

우리는 스레드 코드를 생성한적이 없으나, 우리의 의도를 전달할수가 있습니다.

    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)      //In:샘플은 String이지만, 어떠한 Request를 의미하는게 좋습니다
        {
            Task.Run(async () =>
            {
                await Task.Delay(1000); //어떠한 값을 기다림
                DelayReply reply = new DelayReply();
                reply.message = str;
                return reply;
            }).PipeTo(Self);  //처리완료 결과가 파이프를 통해 DelayReply 처리기로 넘깁니다.
        }

        public void Handle(DelayReply data) //Out:어떠한 Result/Response임을 명시하는게 좋습니다.
        {
            string logtrace = string.Format("I'am {0} RE:{1}", Self.Path, data.message);
            log.Info(data.message);
            Sender.Tell(data);
        }
                
    }

var tasks = new List<Task>();
tasks.Add(actorA.Ask("request", TimeSpan.FromSeconds(1)));
tasks.Add(actorB.Ask("another request", TimeSpan.FromSeconds(5)));

Task.WhenAll(tasks).PipeTo(actorC, Self);