Microsoft MVP성태의 닷넷 이야기
.NET Framework: 1120. C# - BufferBlock<T> 사용 예제 [링크 복사], [링크+제목 복사]
조회: 2185
글쓴 사람
정성태 (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초 이후로 실행이 지연됩니다.

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




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

[연관 글]






[최초 등록일: ]
[최종 수정일: 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 좋은 정보 감사드립니다. ^^ (이래서 꾸준히 공부해야 하나 봅니다. ^^;)
정성태
2021-12-09 02시55분
[서영준] Channel과는 또 다르네요?
[손님]

1  2  3  4  5  6  7  8  9  [10]  11  12  13  14  15  ...
NoWriterDateCnt.TitleFile(s)
12894정성태12/30/20211289.NET Framework: 1125. C# - DefaultObjectPool<T>의 IDisposable 개체에 대한 풀링 문제 [3]파일 다운로드1
12893정성태12/27/20211538.NET Framework: 1124. C# - .NET Platform Extension의 ObjectPool<T> 사용법 소개파일 다운로드1
12892정성태12/26/20211110기타: 83. unsigned 형의 이전 값이 최댓값을 넘어 0을 지난 경우, 값의 차이를 계산하는 방법
12891정성태12/23/20211000스크립트: 38. 파이썬 - uwsgi의 --master 옵션
12890정성태12/23/20211089VC++: 152. Golang - (문자가 아닌) 바이트 위치를 반환하는 strings.IndexRune 함수
12889정성태12/22/20211604.NET Framework: 1123. C# - (SharpDX + DXGI) 화면 캡처한 이미지를 빠르게 JPG로 변환하는 방법파일 다운로드1
12888정성태12/21/20211373.NET Framework: 1122. C# - ImageCodecInfo 사용 시 System.Drawing.Image와 System.Drawing.Bitmap에 따른 Save 성능 차이파일 다운로드1
12887정성태12/21/20211509오류 유형: 777. OpenCVSharp4를 사용한 프로그램 실행 시 "The type initializer for 'OpenCvSharp.Internal.NativeMethods' threw an exception." 예외 발생
12886정성태12/20/20211185스크립트: 37. 파이썬 - uwsgi의 --enable-threads 옵션
12885정성태12/20/2021959오류 유형: 776. uwsgi-plugin-python3 환경에서 MySQLdb 사용 환경
12884정성태12/20/2021974개발 환경 구성: 620. Windows 10+에서 WMI root/Microsoft/Windows/WindowsUpdate 네임스페이스 제거
12883정성태12/19/20211273오류 유형: 775. uwsgi-plugin-python3 환경에서 "ModuleNotFoundError: No module named 'django'" 오류 발생
12882정성태12/18/20211012개발 환경 구성: 619. Windows Server에서 WSL을 위한 리눅스 배포본을 설치하는 방법
12881정성태12/17/20211249개발 환경 구성: 618. WSL Ubuntu 20.04에서 파이썬을 위한 uwsgi 설치 방법 (2)
12880정성태12/16/20211031VS.NET IDE: 170. Visual Studio에서 .NET Core/5+ 역어셈블 소스코드 확인하는 방법
12879정성태12/16/20214365오류 유형: 774. Windows Server 2022 + docker desktop 설치 시 WSL 2로 선택한 경우 "Failed to deploy distro docker-desktop to ..." 오류 발생
12878정성태12/15/20211203개발 환경 구성: 617. 윈도우 WSL 환경에서 같은 종류의 리눅스를 다중으로 설치하는 방법
12877정성태12/15/20211002스크립트: 36. 파이썬 - pymysql 기본 예제 코드
12876정성태12/14/20211039개발 환경 구성: 616. Custom Sources를 이용한 Azure Monitor Metric 만들기
12875정성태12/13/20211057스크립트: 35. python - time.sleep(...) 호출 시 hang이 걸리는 듯한 문제
12874정성태12/13/2021868오류 유형: 773. shell script 실행 시 "$'\r': command not found" 오류
12873정성태12/12/20211301오류 유형: 772. 리눅스 - PATH에 등록했는데도 "command not found"가 나온다면?
12872정성태12/12/20211090개발 환경 구성: 615. GoLang과 Python 빌드가 모두 가능한 docker 이미지 만들기
12871정성태12/12/20212025오류 유형: 771. docker: Error response from daemon: OCI runtime create failed
12870정성태12/9/2021992개발 환경 구성: 614. 파이썬 - PyPI 패키지 만들기 (4) package_data 옵션
12869정성태12/8/20211203개발 환경 구성: 613. git clone 실행 시 fingerprint 묻는 단계를 생략하는 방법
1  2  3  4  5  6  7  8  9  [10]  11  12  13  14  15  ...