J & J - 정성태의 닷넷 이야기

사용자

메뉴

최근 덧글

[정성태] 그냥 RSS Reader 기능과 약간의 UI 편의성 때문에 사용...[이종효] 오래된 소프트웨어는 보안 위협이 되기도 합니다. 혹시 어떤 기능...[정성태] @Keystroke IEEE의 문서를 소개해 주시다니... +_...[손민수 (Keystroke)] 괜히 듀얼채널 구성할 때 한번에 같은 제품 사라고 하는 것이 아...[정성태] 전각(Full-width)/반각(Half-width) 기능을 토...[정성태] Vector에 대한 내용은 없습니다. Vector가 닷넷 BCL...[orion] 글 읽고 찾아보니 디자인 타임에는 InitializeCompon...[orion] 연휴 전에 재현 프로젝트 올리자 생각해 놓고 여의치 않아서 못 ...[정성태] 아래의 글에 정리했으니 참고하세요. C# - Typed D...[정성태] 간단한 재현 프로젝트라도 있을까요? 저런 식으로 설명만 해...

글쓰기

제목

이름

암호

전자우편

HTML

홈페이지

유형

내용

<div style='display: inline'>
<h1 ad='csharp' style='font-family: Malgun Gothic, Consolas; font-size: 20pt; color: #006699; text-align: center; font-weight: bold'>C# - BufferBlock&lt;T&gt; 사용 예제</h1>

오호~~~ 이번 글은 다음의 트윗 덕분에 날로 먹으려고 합니다. ^^ 
 
<pre style='margin: 10px 0px 10px 10px; padding: 10px 0px 10px 10px; background-color: #fbedbb; overflow: auto; font-family: Consolas, Verdana;' >
BufferBlock&lt;T&gt;, 이제 알 것 같다. 태스크 간에 데이터가 들어올 때 까지 비동기로 흐름을 멈추고 기다릴 수 있다! <a href="https://t.co/dINyb2IOdC">pic.twitter.com/dINyb2IOdC</a>

&mdash; calci (@seonghwan_dev) <a href="https://twitter.com/seonghwan_dev/status/1444673927704289294?ref_src=twsrc%5Etfw">October 3, 2021</a>
</pre>
 
실려 있는 이미지가 <a target='tab' href='https://docs.microsoft.com/en-us/dotnet/api/system.threading.tasks.dataflow.bufferblock-1'>BufferBlock</a>의 사용 예를 아주 잘 나타내고 있는데요, 그냥 트윗으로만 묻히기에 아까워 제 글에서 이렇게 도용해 봅니다. ^^ 
 
<pre style='margin: 10px 0px 10px 10px; padding: 10px 0px 10px 10px; background-color: #fbedbb; overflow: auto; font-family: Consolas, Verdana;' >
using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace bufferblock_sample
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var foo = new Foo();
            await foo.Run();
        }
    }

public class Foo
    {
        BufferBlock&lt;int&gt; block1 = new BufferBlock&lt;int&gt;();
        BufferBlock&lt;int&gt; block2 = new BufferBlock&lt;int&gt;();

Task block1Observer;
        Task block2Observer;

public async Task Run()
        {
            block1Observer = Task.Run(async () =&gt; await Observe1());
            block2Observer = Task.Run(async () =&gt; await Observe2());

while (true)
            {
                var read = Console.ReadLine();

try
                {
                    var block = read.Split(' ');

var param1 = block[0];
                    var param2 = block[1];

int blockIndex = int.Parse(param1);
                    int inputValue = int.Parse(param2);

Task task = (blockIndex == 0) ? block1.SendAsync(inputValue)
                                                  : block2.SendAsync(inputValue);
                    await task;
                }
                catch (Exception)
                {
                    Console.Clear();
                    Console.WriteLine("Wrong Format!");
                }
            }
        }

private async Task Observe1()
        {
            try
            {
                while (true)
                {
                    int value = await block1.ReceiveAsync();
                    if (value == default)
                    {
                        break;
                    }

PrintConsole(ConsoleColor.Yellow, $"Observer 1 Received: {value}");
                }
            }
            catch (Exception) { }
        }

private async Task Observe2()
        {
            try
            {
                while (true)
                {
                    int value = await block2.ReceiveAsync();
                    if (value == default)
                    {
                        break;
                    }

PrintConsole(ConsoleColor.Blue, $"Observer 2 Received: {value}");
                }
            }
            catch (Exception) { }
        }

void PrintConsole(ConsoleColor color, string text)
 {
 var oldColor = Console.ForegroundColor;
 Console.ForegroundColor = color;
 Console.WriteLine(text);
 Console.ForegroundColor = oldColor;
 }
 }
}
</pre>
 
간단히 말해서, 비동기로 동작하는 Producer/Consumer의 예제라고 보시면 됩니다. 예전에 AutoReset + ManualReset 이벤트로 동기식 버전을 구현한 것과 비교해 보셔도 좋을 듯합니다. ^^ 
 
<pre style='margin: 10px 0px 10px 10px; padding: 10px 0px 10px 10px; background-color: #fbedbb; overflow: auto; font-family: Consolas, Verdana;' >
AutoReset, ManualReset, Monitor.Wait의 차이
 - 4. 개선 방법 = AutoReset + ManualReset
; <a target='tab' href='https://www.sysnet.pe.kr/2/0/1015#4'>https://www.sysnet.pe.kr/2/0/1015#4</a>
</pre>
 
물론 동기식 버전으로 구현하면 개별 Consumer마다 고정적으로 스레드를 만들어 할당해 두어야 하지만, BufferBlock을 이용하게 되면 비동기로 처리되므로 그럴 필요가 없습니다. 
 
<hr style='width: 50%' /> 
 
BufferBlock 내부에는 당연히 큐잉 구현도 되어 있으므로 ReceiveAsync 호출 이전에 다중으로 SendAsync가 호출이 되어도 순차적으로 잘 처리합니다. 해당 큐에 대한 max 값은 다음의 옵션으로 전달할 수 있는데요, 
 
<pre style='margin: 10px 0px 10px 10px; padding: 10px 0px 10px 10px; background-color: #fbedbb; overflow: auto; font-family: Consolas, Verdana;' >
static DataflowBlockOptions dfbo = new DataflowBlockOptions {
 BoundedCapacity = 2,
};

BufferBlock&lt;int&gt; block1 = new BufferBlock&lt;int&gt;(dfbo);
</pre>
 
만약 위와 같이 2개로 제한한 상황에서 SendAsync를 (ReceiveAsync 수신을 할 수 없는 상태에서) 3번을 하게 되면, 
 
<pre style='margin: 10px 0px 10px 10px; padding: 10px 0px 10px 10px; background-color: #fbedbb; overflow: auto; font-family: Consolas, Verdana;' >
public async Task Run()
{
 block1Observer = Task.Run(async () =&gt; await Observe1());
 block2Observer = Task.Run(async () =&gt; await Observe2());

// 3개의 workitem 전달
 Console.WriteLine(await block1.SendAsync(11));
 Console.WriteLine(await block1.SendAsync(12));
 Console.WriteLine(await block1.SendAsync(13));

while (true)
    {
        // ...[생략]...
    }
}

private async Task Observe1()
{
 Thread.Sleep(5000); // 수신을 할 수 없도록 임시로 5초 지연

// ...[생략]...
}
</pre>
 
처음 2개는 곧바로 SendAsync의 호출이 반환하지만 3번째 호출에서는 5초 이후로 실행이 지연됩니다. 
 
(<a target='tab' href='https://www.sysnet.pe.kr/bbs/DownloadAttachment.aspx?fid=1869&boardid=331301885'>첨부 파일은 이 글의 예제 코드를 포함</a>합니다.)

<hr />[이 글에 대해서 여러분들과 의견을 공유하고 싶습니다. 틀리거나 미흡한 부분 또는 의문 사항이 있으시면 언제든 댓글 남겨주십시오.]

</div>

첨부파일

스팸 방지용 인증 번호

3230 (왼쪽의 숫자를 입력해야 합니다.)