Microsoft MVP성태의 닷넷 이야기
.NET Framework: 1120. C# - BufferBlock<T> 사용 예제 [링크 복사], [링크+제목 복사]
조회: 1159
글쓴 사람
정성태 (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 좋은 정보 감사드립니다. ^^ (이래서 꾸준히 공부해야 하나 봅니다. ^^;)
정성태
2021-12-09 02시55분
[서영준] Channel과는 또 다르네요?
[손님]

[1]  2  3  4  5  6  7  8  9  10  11  12  13  14  15  ...
NoWriterDateCnt.TitleFile(s)
12938정성태1/24/202230개발 환경 구성: 633. Docker Desktop + k8s 환경에서 local 이미지를 사용하는 방법
12937정성태1/24/202237.NET Framework: 1139. C# - ffmpeg(FFmpeg.AutoGen)를 이용해 오디오(mp2) 인코딩하는 예제(encode_audio.c)파일 다운로드1
12936정성태1/22/202235.NET Framework: 1138. C# - ffmpeg(FFmpeg.AutoGen)를 이용해 멀티미디어 파일의 메타데이터를 보여주는 예제(metadata.c)파일 다운로드1
12935정성태1/22/202230.NET Framework: 1137. ffmpeg의 파일 해시 예제(ffhash.c)를 C#으로 포팅파일 다운로드1
12934정성태1/22/202217오류 유형: 788. Warning C6262 Function uses '65564' bytes of stack: exceeds /analyze:stacksize '16384'. Consider moving some data to heap.
12933정성태1/21/202231.NET Framework: 1136. C# - ffmpeg(FFmpeg.AutoGen)를 이용해 MP2 오디오 파일 디코딩 예제(decode_audio.c)파일 다운로드1
12932정성태1/20/202282.NET Framework: 1135. C# - ffmpeg(FFmpeg.AutoGen)로 하드웨어 가속기를 이용한 비디오 디코딩 예제(hw_decode.c) [2]파일 다운로드1
12931정성태1/20/202289개발 환경 구성: 632. ASP.NET Core 프로젝트를 AKS/k8s에 올리는 과정
12930정성태1/19/202237개발 환경 구성: 631. AKS/k8s의 Volume에 파일 복사하는 방법
12929정성태1/19/202259개발 환경 구성: 630. AKS/k8s의 Pod에 Volume 연결하는 방법
12928정성태1/18/202256개발 환경 구성: 629. AKS/Kubernetes에서 호스팅 중인 pod에 shell(/bin/bash)로 진입하는 방법
12927정성태1/18/202263개발 환경 구성: 628. AKS 환경에 응용 프로그램 배포 방법
12926정성태1/17/202239오류 유형: 787. AKS - pod 배포 시 ErrImagePull/ImagePullBackOff 오류
12925정성태1/17/202274개발 환경 구성: 627. AKS의 준비 단계 - ACR(Azure Container Registry)에 docker 이미지 배포
12924정성태1/15/2022167.NET Framework: 1134. C# - ffmpeg(FFmpeg.AutoGen)를 이용한 비디오 디코딩 예제(decode_video.c) [2]파일 다운로드1
12923정성태1/15/202286개발 환경 구성: 626. ffmpeg.exe를 사용해 비디오 파일을 MPEG1 포맷으로 변경하는 방법
12922정성태1/14/202295개발 환경 구성: 625. AKS - Azure Kubernetes Service 생성 및 SLO/SLA 변경 방법
12921정성태1/14/202256개발 환경 구성: 624. Docker Desktop에서 별도 서버에 설치한 docker registry에 이미지 올리는 방법
12920정성태1/14/202223오류 유형: 786. Camtasia - An error occurred with the camera: Failed to Add Video Sampler.
12919정성태1/13/202283Windows: 199. Host Network Service (HNS)에 의해서 점유되는 포트
12918정성태1/13/202286Linux: 47. WSL - shell script에서 설정한 환경 변수가 스크립트 실행 후 반영되지 않는 문제
12917정성태1/12/202250오류 유형: 785. C# - The type or namespace name '...' could not be found (are you missing a using directive or an assembly reference?)
12916정성태1/12/202230오류 유형: 784. TFS - One or more source control bindings for this solution are not valid and are listed below.
12915정성태1/11/202267오류 유형: 783. Visual Studio - We didn't find any interpreters
12914정성태1/11/2022180VS.NET IDE: 172. 비주얼 스튜디오 2022의 파이선 개발 환경 지원
[1]  2  3  4  5  6  7  8  9  10  11  12  13  14  15  ...