Microsoft MVP성태의 닷넷 이야기
.NET Framework: 1120. C# - BufferBlock<T> 사용 예제 [링크 복사], [링크+제목 복사]
조회: 894
글쓴 사람
정성태 (techsharer at outlook.com)
홈페이지
첨부 파일

C# - BufferBlock<T> 사용 예제

오호~~~ 이번 글은 다음의 트윗 덕분에 날로 먹으려고 합니다. ^^

BufferBlock<T>, 이제 알 것 같다.

태스크 간에 데이터가 들어올 때 까지 비동기로 흐름을 멈추고 기다릴 수 있다! pic.twitter.com/dINyb2IOdC — calci (@seonghwan_dev) October 3, 2021

실려 있는 이미지가 BufferBlock의 사용 예를 아주 잘 나타내고 있는데요, 그냥 트윗으로만 묻히기에 아까워 제 글에서 이렇게 도용해 봅니다. ^^

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<int> block1 = new BufferBlock<int>();
        BufferBlock<int> block2 = new BufferBlock<int>();

        Task block1Observer;
        Task block2Observer;

        public async Task Run()
        {
            block1Observer = Task.Run(async () => await Observe1());
            block2Observer = Task.Run(async () => 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;
        }
    }
}

간단히 말해서, 비동기로 동작하는 Producer/Consumer의 예제라고 보시면 됩니다. 예전에 AutoReset + ManualReset 이벤트로 동기식 버전을 구현한 것과 비교해 보셔도 좋을 듯합니다. ^^

AutoReset, ManualReset, Monitor.Wait의 차이
   - 4. 개선 방법 = AutoReset + ManualReset
; https://www.sysnet.pe.kr/2/0/1015#4

물론 동기식 버전으로 구현하면 개별 Consumer마다 고정적으로 스레드를 만들어 할당해 두어야 하지만, BufferBlock을 이용하게 되면 비동기로 처리되므로 그럴 필요가 없습니다.




BufferBlock 내부에는 당연히 큐잉 구현도 되어 있으므로 ReceiveAsync 호출 이전에 다중으로 SendAsync가 호출이 되어도 순차적으로 잘 처리합니다. 해당 큐에 대한 max 값은 다음의 옵션으로 전달할 수 있는데요,

static DataflowBlockOptions dfbo = new DataflowBlockOptions {
        BoundedCapacity = 2,
};

BufferBlock<int> block1 = new BufferBlock<int>(dfbo);

만약 위와 같이 2개로 제한한 상황에서 SendAsync를 (ReceiveAsync 수신을 할 수 없는 상태에서) 3번을 하게 되면,

public async Task Run()
{
    block1Observer = Task.Run(async () => await Observe1());
    block2Observer = Task.Run(async () => 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초 지연

    // ...[생략]...
}

처음 2개는 곧바로 SendAsync의 호출이 반환하지만 3번째 호출에서는 5초 이후로 실행이 지연됩니다.

(첨부 파일은 이 글의 예제 코드를 포함합니다.)




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



donaricano-btn



[최초 등록일: ]
[최종 수정일: 10/6/2021]

Creative Commons License
이 저작물은 크리에이티브 커먼즈 코리아 저작자표시-비영리-변경금지 2.0 대한민국 라이센스에 따라 이용하실 수 있습니다.
by SeongTae Jeong, mailto:techsharer at outlook.com

비밀번호

댓글 쓴 사람
 



2021-10-06 10시08분
[dimohy] 오 ManualReset을 같은 목적을 위해 종종 사용하는데 값을 주거니 받거니 하는 용도로 BufferBlock<T>가 매우 편리하겠군요! 감사합니다.
[손님]
2021-10-06 11시41분
문서에 보면 .NET Core 1부터 지원하고 있었는데, 이제서야 알게 되는군요. ^^ 뭐가 많이 추가된 것 같은데... 이런 식으로라도 하나씩 정리를 해야겠습니다.
정성태
2021-10-24 09시28분
[tb] 나중에 나온 System.Threading.Channels 도 좋더라구요. IAsyncEnumerable 로 구현했고 Cancellation 도 지원하구요.
[손님]
2021-10-25 09시24분
@tb 좋은 정보 감사드립니다. ^^ (이래서 꾸준히 공부해야 하나 봅니다. ^^;)
정성태

[1]  2  3  4  5  6  7  8  9  10  11  12  13  14  15  ...
NoWriterDateCnt.TitleFile(s)
12869정성태12/8/202112개발 환경 구성: 613. git clone 실행 시 fingerprint 묻는 단계를 생략하는 방법
12868정성태12/7/202122오류 유형: 770. twine 업로드 시 "HTTPError: 400 Bad Request ..." 오류
12867정성태12/7/202133개발 환경 구성: 612. 파이썬 - PyPI 패키지 만들기 (3) entry_points 옵션
12866정성태12/7/202118오류 유형: 769. "docker build ..." 시 "failed to solve with frontend dockerfile.v0: failed to read dockerfile ..." 오류
12865정성태12/6/202130개발 환경 구성: 611. 파이썬 - PyPI 패키지 만들기 (2) long_description, cmdclass 옵션
12864정성태12/6/202141Linux: 46. WSL 환경에서 find 명령을 사용해 파일을 찾는 방법
12863정성태12/4/202154개발 환경 구성: 610. 파이썬 - PyPI 패키지 만들기
12862정성태12/3/202147오류 유형: 768. Golang - 빌드 시 "cmd/go: unsupported GOOS/GOARCH pair linux /amd64" 오류
12861정성태12/3/202163개발 환경 구성: 609. 파이썬 - "Windows embeddable package"로 개발 환경 구성하는 방법
12860정성태12/1/2021124오류 유형: 767. SQL Server - 127.0.0.1로 접속하는 경우 "Access is denied"가 발생한다면?
12859정성태12/1/2021139개발 환경 구성: 608. Hyper-V 가상 머신에 Console 모드로 로그인하는 방법
12858정성태11/30/2021342개발 환경 구성: 607. 로컬의 USB 장치를 원격 머신에 제공하는 방법 - usbip-win
12857정성태11/24/2021113개발 환경 구성: 606. WSL Ubuntu 20.04에서 파이썬을 위한 uwsgi 설치 방법
12856정성태11/23/2021339.NET Framework: 1121. C# - 동일한 IP:Port로 바인딩 가능한 서버 소켓 [2]
12855정성태11/13/2021112개발 환경 구성: 605. Azure App Service - Kudu SSH 환경에서 FTP를 이용한 파일 전송
12854정성태11/13/2021231개발 환경 구성: 604. Azure - 윈도우 VM에서 FTP 여는 방법
12853정성태11/10/2021137오류 유형: 766. Azure App Service - JBoss 호스팅 생성 시 "This region has quota of 0 PremiumV3 instances for your subscription. Try selecting different region or SKU."
12851정성태11/1/2021237스크립트: 34. 파이썬 - MySQLdb 기본 예제 코드
12850정성태10/27/2021516오류 유형: 765. 우분투에서 pip install mysqlclient 실행 시 "OSError: mysql_config not found" 오류
12849정성태10/17/2021519스크립트: 33. JavaScript와 C#의 시간 변환
12848정성태10/17/2021475스크립트: 32. 파이썬 - sqlite3 기본 예제 코드
12847정성태10/14/2021429스크립트: 31. 파이썬 gunicorn - WORKER TIMEOUT 오류 발생
12846정성태10/7/2021525스크립트: 30. 파이썬 __debug__ 플래그 변수에 따른 코드 실행 제어
12845정성태10/6/2021894.NET Framework: 1120. C# - BufferBlock<T> 사용 예제 [4]파일 다운로드1
12844정성태10/3/2021446오류 유형: 764. MSI 설치 시 "... is accessible and not read-only." 오류 메시지
[1]  2  3  4  5  6  7  8  9  10  11  12  13  14  15  ...