Versions Compared

Old Version 6

changes.mady.by.user PSMON

Saved on

compared with

New Version Current

changes.mady.by.user PSMON

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Info

오늘날 우리가 인터넷에서 서비스를 사용하는 방식은 데이터 전송을 비롯한 많은 스트리밍 데이터 인스턴스를 포함하고 있습니다.

그것이 다운로드 서비스이든,영화스트리밍 서비스이든 대용량의 데이터를 우리의 어플리케이션이 가진 메모리와 네트워크대역으로는

순간적으로 처리할수가 없습니다. 스트리밍은 항상 생산자와 소비자로 구분되어 있으며 어떻게 생산자와 소비자의 다양한 속도차이를

고려하여 잘처리하느냐에 대한 고민이 필요하여, Akka Stream은 이러한 스트림 처리를 위한 여러가지 장치를 제공합니다.

그중 유압조절을 할수있는 Throttle 장치에대해 간략하게 알아보고 활용해보겠습니다.

스트림처리

문제

  • 강우량은 측정은 할수 있지만, 정확하게 예측은 할수 없습니다.  이러한 비가 계속 내리며 저장탱크(버퍼)는 오로지 내리는 비만 저장을하며 용량의 한계가 있습니다.
  • 비가 내리는중 저장된 탱크의 물을 정제하여, 마실수 있을때가 될때 소비자에게 쉬지않고 계속 지속적으로 소비하여, 효율적으로 고객에게 물을 공급해야합니다.

스트림을 활용하여 생산된(비)를 지속적으로 소비를 하여 저장탱크가 터지지 않고 소비자에게 물이 공급 하는 것이넘치지 않고, 또는 지속적으로 빗물을 소비하여

소비자에게 끊임없이 물을 공급해봅시다. 물론 비는 일정하지는 않지만 계속 내리고 있습니다.이 프로젝트에서 설명하고자하는 주요내용입니다. 

배치처리는 스트림과 반대되는 개념으로 구현난이도는 쉬우나 , 오늘날의 트래픽을 처리하기에는 한계가 분명 존재합니다.

해결 시나리오

  • 비를 받는 저장탱크와, 정수에 사용되는 저장탱크 두개를 준비합니다.
  • A-throttleActor) : A작업자는, 쌓인 빗물을 작업자 탱크에게 지속적(1초) 으로 스트림을 흘려보냅니다. 
  • B-throttleWork) : B작업자는 초당 10ml 마실수 있는 물을 만들어 낼수있으며, 고객에게 서빙을 제공 합니다.
  • C : 소비자는 작업자가 무엇을하는지 관심이 없고 물이 생기면 마십니다.

...

Akka Stream Flow 가 제공하는 기능들이며, 이것은 작업자가 직접 설계를 해야합니다.실시간 데이터가 버퍼에 넘치지 않도록 지속적으로 활동할수 있으며

저장탱크가 넘치지 않도록 개발자가 직접 설계에 반영해야합니다.  ( AKKA가 모든것을 책임지지는 않습니다.)

코드 흐름 - 사용

유닛테스트를 활용하여 위와같은 흐름을 어떻게 제어를 하는지 사용부를 먼저 알아 보겠습니다.

복잡한 기능을 단순화하고,유닛테스트를 통해 검증하는것이 웹노리 Akak.net이 추구하는 방향입니다.(웹노리 깨알 광고)

Code Block
themeEmacs
titleunittest
collapsetrue
주요코드설명
// 빗물을 처리하는 액터(지정된 초마다,빗물을 모아서 전달)
var throttleActor = Sys.ActorOf(Props.Create(() => new ThrottleActor(timeSec)));
// 물을 생산하는 액터(물생산은 전달받는대로, 초당 5개씩만 처리,느리다 싶으면 Work를 더 늘려준다.)
var throttleWork = Sys.ActorOf(Props.Create(() => new ThrottleWork(elemntPerSec, timeSec)));

// 생산자와 소비자를 연결합니다
// DI가 사용이되지 않았으며,참조객체 디펀던시가 없습니다. 이것은 네트워크로 작업자가 분리될수 있음을 의미합니다.
throttleActor.Tell(new SetTarget(throttleWork));
throttleWork.Tell(new SetTarget(probe));

// 빗물은 불규칙적으로 쌓일수 있으나, 작업자에게는 어느정도 모인 물이 방류됩니다.
throttleActor.Tell(new Queue(/*Any "비" Object*/));

//고객은 작업자가 무엇을했는지 관심없습니다. 그냥 플리즈 워터(ExpectMsg)했을때, 지정된 시간에 물을 받으면 됩니다.
probe.ExpectMsg<DelayMsg>();
Code Block
themeEmacs
titleUnitTest
collapsetrue
using System;
using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.NUnit3;
using AkkaNetCore.Actors.Utils;
using AkkaNetCore.Models.Message;
using NUnit.Framework;

namespace AkkaNetCoreTest.Actors
{
    class ThrottleActorTest : TestKit
    {using System;
using Akka.Actor;
using Akka.TestKit;
using Akka.TestKit.NUnit3;
using AkkaNetCore.Actors.Utils;
using AkkaNetCore.Models.Message;
using NUnit.Framework;

namespace AkkaNetCoreTest.Actors
{
    class ThrottleActorTest : TestKit
    {
        protected TestProbe probe;

        [SetUp]
        public void Setup()
        {
            //스트림을 제공받는 최종 소비자 ( 물을 제공 받는 고객 )
            probe = this.CreateTestProbe();
        }

        [TestCase(15)]
        publicprotected void ThrottleActorAreOK(int cutoffSec)TestProbe probe;

        {[SetUp]
        public void   // 초당 5개 처리한정 ( 더 처리하고 싶으면 이값을 늘린다.)
Setup()
        {
            //스트림을 제공받는 최종 소비자 ( 물을 int제공 timeSec받는 =고객 1;)
            int elemntPerSecprobe = 5this.CreateTestProbe();
        }

    
    [TestCase(15)]
        varpublic throttleActorvoid = Sys.ActorOf(Props.Create(() => new ThrottleActor(timeSec)));
ThrottleActorAreOK(int cutoffSec)
        {
            // var초당 throttleWork5개 =처리한정 Sys.ActorOf(Props.Create(() => new ThrottleWork(elemntPerSec, timeSec)));
( 더 처리하고 싶으면 이값을 늘린다.)
            //int 밸브에게timeSec 작업자= 지정1;
 ( 밸브는 초당 스트림을 모아서 방출한다 )
     int elemntPerSec = 5;    // 작업자는 방류된 스트림을 기본적으로 쌓아두고, 초당 지정된 개수만
 처리한다.
           var throttleActor.Tell(new SetTarget(throttleWork = Sys.ActorOf(Props.Create(() => new ThrottleActor(timeSec)));
            

var throttleWork           // 소비자지정 : 소비자는 몇개가 초당 처리되던 상관없이, 완료된 작업만 제공받는다.
            throttleWork.Tell(new SetTarget(probe));

= Sys.ActorOf(Props.Create(() => new ThrottleWork(elemntPerSec, timeSec)));

            // 밸브에게 작업자 지정 ( 밸브는 초당 스트림을 모아서 방출한다 )
            Within(TimeSpan.FromSeconds(cutoffSec), () =>
       // 작업자는 방류된 스트림을 기본적으로 쌓아두고, 초당 지정된 개수만 처리한다.
     {
       throttleActor.Tell(new SetTarget(throttleWork));        // 50개 처리완료는 10초이내에 끝나야함...

            // 소비자지정 : 소비자는 for(int i=0; i<50; i++)
   몇개가 초당 처리되던 상관없이, 완료된 작업만 제공받는다.
             {throttleWork.Tell(new SetTarget(probe));

            Within(TimeSpan.FromSeconds(cutoffSec), () =>
      string  seq = (i + 1).ToString(); {
                // 50개 처리완료는 10초이내에 throttleActor.Tell(new Queue(new DelayMsg()끝나야함...
                for(int i=0;   {i<50; i++)
                {
        Delay = 0,
          string seq = (i + 1).ToString();
         Seq     = seq,
     throttleActor.Tell(new Queue(new DelayMsg()
                 Message = $"초당:{elemntPerSec} 테스트-{seq}",
                        StateDelay = DelayMsgState.Reserved0,
                    }));
    Seq = seq,
          }

              Message  DelayMsg lastMessage = null;  //물이라고 가정하자..(빗물->정제->포장 과정 생략)
= $"초당:{elemntPerSec} 테스트-{seq}",
                        forState (int i = 0; i < 50; i++)
= DelayMsgState.Reserved
                    }));
   {
             }

       lastMessage =probe.ExpectMsg<DelayMsg>();
        DelayMsg lastMessage = null;  //물이라고 가정하자..(빗물->정제->포장 과정 }생략)
                //마지막 메시지의 Seq는 50이여야함
    for (int i = 0; i < 50; i++)
            Assert.AreEqual("50", lastMessage.Seq);

    {
             });
        }
lastMessage =probe.ExpectMsg<DelayMsg>();
         /* 위 테스트의 결과로그,블락킹없이 실시간으로 초당 5개씩 처리함}
            [49:09] -  초당:5 테스트-1
    //마지막 메시지의 Seq는 50이여야함
         [49:09] -  초당:5 테스트-2    Assert.AreEqual("50", lastMessage.Seq);

            [49:09] -  초당:5 테스트-3
  });
        }
        /* 위 [49:09] -  초당:5 테스트-4테스트의 결과로그,블락킹없이 실시간으로 초당 5개씩 처리함
            [49:1009] -  초당:5 테스트-51
            [49:1009] -  초당:5 테스트-62
            [49:1009] -  초당:5 테스트-73
            [49:1009] -  초당:5 테스트-84
            [49:10] -  초당:5 테스트-95
            [49:1110] -  초당:5 테스트-106
         
   [49:10] -    */초당:5 테스트-7
    }
}
Code Block
themeEmacs
title주요코드설명
// 빗물을 처리하는 액터(지정된 초마다,빗물을 모아서 전달)
var throttleActor = Sys.ActorOf(Props.Create(() => new ThrottleActor(timeSec)));
// 물을 생산하는 액터(물생산은 전달받는대로, 초당 5개씩만 처리,느리다 싶으면 Work를 더 늘려준다.)
var throttleWork = Sys.ActorOf(Props.Create(() => new ThrottleWork(elemntPerSec, timeSec)));

// 생산자와 소비자를 연결합니다
// DI가 사용이되지 않았으며,참조객체 디펀던시가 없습니다. 이것은 네트워크로 작업자가 분리될수 있음을 의미합니다.
throttleActor.Tell(new SetTarget(throttleWork));
throttleWork.Tell(new SetTarget(probe));

// 빗물을, 빗물 처리자에게 불규칙적으로 계속 보낼수 있습니다.
throttleActor.Tell(new Queue("빗물"));

//생산자는 그냥 물을 받게됩니다.
probe.ExpectMsg<DelayMsg>(); [49:10] -  초당:5 테스트-8
            [49:10] -  초당:5 테스트-9
            [49:11] -  초당:5 테스트-10         
        */
    }
}



액터구현

ThrottleActor 는 이 장의 설명을 위해,  작성하였으나 데이터 처리 대상의 제약을 두지 않았기때문에 다양하게 활용가능합니다.

Code Block
themeEmacs
titleThrottleActor
linenumberstrue
collapsetrue
주요 코드설명
AkkaStream을 Actor내에 사용하기 위해 준비되는코드입니다.var materializer = Context.Materializer();

// FSM이 사용되었으며, 특정초마다 수집한 데이터를 처리합니다. (스트림 출발점)
When(State.Active, state =>
{
.......	
}, TimeSpan.FromSeconds(CollectSec));

// 빗물담당자는 빗물을 한꺼번에 흘려보냈지만
// 작업자는 안정적으로 초당 5개 처리 제약을두며 흘러들어온 버퍼를 꾸준하게 작업한만큼 비웁니다.
// 처리대상 소스 :한꺼번에 몰려욤
Source<object, NotUsed> source = Source.From(batchMessage.Obj);  

// 조절기를 달아서 천천히 처리,가속 기울기를 선택할수 있는것은 보너스입니다.
source.Throttle(element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping)
Code Block
themeEmacs
titleThrottleActor
linenumberstrue
collapsetrue
using System;
using System.Collections.Immutable;
using Akka;
using Akka.Actor;
using Akka.Event;
using Akka.Monitoring;
using Akka.Streams;
using Akka.Streams.Dsl;
using AkkaNetCore.Models.Message;

namespace AkkaNetCore.Actors.Utils
{
    // 일괄 처리(데이터 인입)    
    public class ThrottleWork : ReceiveActor
using System;
using System.Collections.Immutable;
using Akka;
using Akka.Actor;
using Akka.Event;
using Akka.Monitoring;
using Akka.Streams;
using Akka.Streams.Dsl;
using AkkaNetCore.Models.Message;

namespace AkkaNetCore.Actors.Utils
{
    // 일괄 처리(데이터 인입)    
    public class ThrottleWork : ReceiveActor
    {
        private readonly ILoggingAdapter logger = Context.GetLogger();

        private IActorRef consumer;

        public ThrottleWork(int element,int maxBust)
        {
            
            ReceiveAsync<SetTarget>(async target =>
            {
        private readonly       consumerILoggingAdapter logger = targetContext.RefGetLogger();

        private IActorRef   })consumer;

        public    ReceiveAsync<object>(async message =>
    ThrottleWork(int element,int maxBust)
        {
            
     if  (message is Batch batchMessage)
  ReceiveAsync<SetTarget>(async target =>
            {
                consumer = target.Ref;
   int Count = batchMessage.Obj.Count;
      });

            ReceiveAsync<object>(async  Context.IncrementMessagesReceived();message =>
            {
        Source<object, NotUsed> source = Source.From(batchMessage.Obj);

    if (message is Batch batchMessage)
            using (var materializer = Context.Materializer()) {
                    {
int Count = batchMessage.Obj.Count;
                    Context.IncrementMessagesReceived();
     var factorials = source;
            Source<object, NotUsed> source = Source.From(batchMessage.Obj);

        factorials
            using (var materializer = Context.Materializer())
             //.ZipWith(Source.From(Enumerable.Range(0, 100)), (num, idx) => $"{idx}! = {num}")
                        var factorials = source;
   .Throttle(element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping)
                  factorials
           .RunForeach(obj => {                //.ZipWith(Source.From(Enumerable.Range(0, 100)), (num, idx) => $"{idx}! = {num}")
          
                   .Throttle(element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping)
           var nowstr = DateTime.Now.ToString("mm:ss");
               .RunForeach(obj => {                if(obj is DelayMsg delayMsg)
              
                   {              var nowstr = DateTime.Now.ToString("mm:ss");
                    
             if(obj is DelayMsg delayMsg)
                     Console.WriteLine($"[{nowstr}] -  {delayMsg.Message}");
         {                            if (consumer != null) consumer.Tell(delayMsg);
     
                            }         Console.WriteLine($"[{nowstr}] -  {delayMsg.Message}");
                     
                if (consumer != null) consumer.Tell(delayMsg);
         }, materializer)
                       }      .Wait();
                    }
       
         }
            });
        }, materializer)
    }


    public class ThrottleActor : FSM<State, IData>
    {
        private readonly ILoggingAdapter logger = Context.GetLoggerWait();
        private int CollectSec;

        public ThrottleActor(int _CollectSec) }
        {
        }
          CollectSec = _CollectSec});

        }
    StartWith(State.Idle, Uninitialized.Instance);}


    public class ThrottleActor : FSM<State, IData>
   When(State.Idle, state => {
        private readonly ILoggingAdapter logger  {= Context.GetLogger();
        private int CollectSec;

      if  public (state.FsmEvent is SetTarget target && state.StateData is Uninitialized)ThrottleActor(int _CollectSec)
        {
            CollectSec = _CollectSec;

  {
          StartWith(State.Idle, Uninitialized.Instance);

          return Stay().Using(new Todo(target.Ref, ImmutableList<object>.Empty)); When(State.Idle, state =>
            {
       }

         if (state.FsmEvent is SetTarget target && state.StateData returnis null;Uninitialized)
            });

    {
         When(State.Active, state =>
         return Stay().Using(new  {Todo(target.Ref, ImmutableList<object>.Empty));
                if ((state.FsmEvent is Flush || state.FsmEvent is StateTimeout)
}

                return null;
        && state.StateData is Todo t});

            When(State.Active, state =>
        {
    {
                returnif GoTo(State.Idle).Using(t.Copy(ImmutableList<object>.Empty));
         (state.FsmEvent is Flush || state.FsmEvent is StateTimeout)
       }

             && state.StateData is returnTodo null;t)
            }, TimeSpan.FromSeconds(CollectSec));

    {
        WhenUnhandled(state =>
            {
return GoTo(State.Idle).Using(t.Copy(ImmutableList<object>.Empty));
                }

 if (state.FsmEvent is Queue q && state.StateData is Todo t)
      return null;
         {
   }, TimeSpan.FromSeconds(CollectSec));

            WhenUnhandled(state =>
   return GoTo(State.Active).Using(t.Copy(t.Queue.Add(q.Obj)));
        {
        }
        if (state.FsmEvent is Queue q && state.StateData is elseTodo t)
                {
                     logger.Warning($"Received unhandled request {state.FsmEvent} in state {StateName}/{state.StateData}");
return GoTo(State.Active).Using(t.Copy(t.Queue.Add(q.Obj)));
                }
         return Stay();       else
                }{
             });

       logger.Warning($"Received unhandled request {state.FsmEvent} in OnTransition((initialState, nextState) =>
state {StateName}/{state.StateData}");
              {
      return Stay();
         if (initialState == State.Active && nextState == State.Idle) }
            });

     {
       OnTransition((initialState, nextState) =>
           if (StateData{
 is Todo todo)
             if (initialState == State.Active && nextState  {== State.Idle)
                {
        todo.Target.Tell(new Batch(todo.Queue));
           if (StateData is Todo todo)
     }
               {
     else
                    {todo.Target.Tell(new Batch(todo.Queue));
                    }
    // nothing to do
             else
       }
               {
 }
            });

            Initialize();// nothing to do
        }
    }
}
Code Block
themeEmacs
title핵심코드 설명
// AkkaStream을 Actor내에 사용하기 위해 준비되는코드입니다.var materializer = Context.Materializer();

// FSM이 사용되었으며, 특정초마다 수집한 데이터를 처리합니다. (스트림 출발점)
When(State.Active, state =>
{
.......	
}, TimeSpan.FromSeconds(CollectSec));

// 빗물담당자는 빗물을 한꺼번에 흘려보냈지만
// 작업자는 안정적으로 초당 5개 처리 제약을두며 흘러들어온 버퍼를 꾸준하게 작업한만큼 비웁니다.
// 처리대상 소스 :한꺼번에 몰려욤
Source<object, NotUsed> source = Source.From(batchMessage.Obj);  

// 조절기를 달아서 천천히 처리,가속 기울기를 선택할수 있는것은 보너스입니다.
source.Throttle(element, TimeSpan.FromSeconds(1), maxBust, ThrottleMode.Shaping) 
 }
                }
            });

            Initialize();
        }
    }
}

클러스터와 함께 Stream처리는 클러스터와 함께 Stream은 Akka 에서 어려운 부분중 하나입니다.

Stream처리에 대하 더 자세한것은 아래 내용을 참고하세요

...