C# Academy · 강의

System.Threading.Channels

백프레셔를 제어하기 위해 Channel, ChannelWriter, ChannelReader로 생산자-소비자 파이프라인을 구축합니다.

System.Threading.Channels은(는) CoddyKit의 무료 C# Academy 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 C# Academy 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. C# Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

채널이란 무엇인가요

System.Threading.Channels(.NET Core 3에서 도입됨)은 고성능의 스레드 안전 생산자-소비자 큐를 제공합니다. BlockingCollection과 달리 채널은 완전히 비동기적이고 할당 효율이 높아 프로세스 내부 파이프라인에 적합합니다.

채널 만들기

채널은 팩토리를 사용해 만듭니다. unbounded(제한 없음) 또는 bounded(역압이 적용되는 제한된 용량)를 선택할 수 있습니다. 팩토리는 Channel<T>과 Writer 및 Reader를 반환합니다.

using System.Threading.Channels;

// Unbounded: unlimited capacity, no backpressure
var unbounded = Channel.CreateUnbounded<string>();

// Bounded: max 100 items; writer waits when full
var bounded = Channel.CreateBounded<string>(100);

// Bounded with drop-oldest strategy:
var dropping = Channel.CreateBounded<string>(new BoundedChannelOptions(50)
{
    FullMode = BoundedChannelFullMode.DropOldest
});

채널에 쓰기

생산자는 WriteAsync(bounded 채널이 가득 차면 기다림) 또는 TryWrite(가득 찬 경우 즉시 false 반환)를 사용해 항목을 씁니다. Writer.Complete()로 완료를 알립니다.

var channel = Channel.CreateUnbounded<int>();

// Producer task
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 100; i++)
    {
        await channel.Writer.WriteAsync(i);
        await Task.Delay(10);
    }
    channel.Writer.Complete(); // signal no more items
});

채널에서 읽기

소비자는 ReadAllAsync()(가장 간단한 방법) 또는 더 세밀하게 제어할 수 있는 ReadAsync/TryRead를 사용해 읽습니다. 작성자가 Complete()을 호출하면 ReadAllAsync가 완료됩니다.

// Consumer task
var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"Processed: {item}");
        // Naturally paced — waits for next item
    }
    Console.WriteLine("Channel completed");
});

await Task.WhenAll(producer, consumer);

여러 생산자

채널은 스레드 안전합니다. 여러 생산자가 동시에 쓸 수 있으므로 잠금이 필요하지 않습니다. 모든 생산자가 작업을 마친 후에만 Writer.Complete()을 호출합니다.

var channel = Channel.CreateUnbounded<WorkItem>();

var producers = Enumerable.Range(0, 4).Select(id =>
    Task.Run(async () =>
    {
        for (int i = 0; i < 25; i++)
            await channel.Writer.WriteAsync(new WorkItem(id, i));
    }));

// Wait for all producers before completing the writer
await Task.WhenAll(producers);
channel.Writer.Complete();

여러 소비자(팬아웃)

같은 채널 리더에서 여러 소비자 작업을 실행하면 처리를 병렬화할 수 있습니다. 각 항목은 정확히 하나의 소비자에게 전달되며, 모든 소비자에게 브로드캐스트되지는 않습니다.

var channel = Channel.CreateBounded<WorkItem>(100);

// 4 parallel consumers
var consumers = Enumerable.Range(0, 4).Select(id =>
    Task.Run(async () =>
    {
        await foreach (var item in channel.Reader.ReadAllAsync())
        {
            await ProcessItemAsync(item);
            Console.WriteLine($"Consumer {id} processed {item.Id}");
        }
    }));

await Task.WhenAll(consumers);

파이프라인 패턴

채널을 연결해 처리 파이프라인을 구성합니다. 각 단계는 한 채널에서 읽고, 항목을 처리한 다음, 다음 채널에 씁니다. 각 단계가 동시에 실행되며 자연스럽게 역압이 적용됩니다.

// Stage 1: raw data
var stage1 = Channel.CreateBounded<string>(50);
// Stage 2: parsed
var stage2 = Channel.CreateBounded<ParsedRecord>(50);
// Stage 3: enriched
var stage3 = Channel.CreateBounded<EnrichedRecord>(50);

var parse   = ParseStageAsync(stage1.Reader, stage2.Writer);
var enrich  = EnrichStageAsync(stage2.Reader, stage3.Writer);
var persist = PersistStageAsync(stage3.Reader);

await Task.WhenAll(parse, enrich, persist);

bounded 채널을 사용한 역압

bounded 채널은 자동으로 역압을 적용합니다. 채널이 가득 차면 소비자가 항목을 꺼낼 때까지 WriteAsync가 생산자를 일시 중단합니다. 수동으로 조절하는 코드를 작성할 필요가 없습니다.

// Bounded channel: max 10 items
var channel = Channel.CreateBounded<string>(10);

// Fast producer
var producer = Task.Run(async () =>
{
    for (int i = 0; i < 1000; i++)
    {
        // WriteAsync waits when channel has 10 items
        await channel.Writer.WriteAsync($"item-{i}");
        // Producer is naturally slowed to consumer speed
    }
    channel.Writer.Complete();
});

// Slow consumer
var consumer = Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        await Task.Delay(50); // simulate slow processing
        Console.WriteLine(item);
    }
});

채널의 오류 처리

Writer.Complete(exception)에 예외를 전달하면 대기 중인 모든 리더에게 오류를 전파할 수 있습니다. 리더는 다음에 채널을 읽을 때 예외를 확인합니다.

var channel = Channel.CreateUnbounded<int>();

var producer = Task.Run(async () =>
{
    try
    {
        for (int i = 0; i < 100; i++)
        {
            if (i == 50) throw new Exception("Producer failed at 50");
            await channel.Writer.WriteAsync(i);
        }
        channel.Writer.Complete();
    }
    catch (Exception ex)
    {
        channel.Writer.Complete(ex); // propagate to reader
    }
});

try
{
    await foreach (var item in channel.Reader.ReadAllAsync())
        Console.WriteLine(item);
}
catch (Exception ex)
{
    Console.Error.WriteLine($"Channel error: {ex.Message}");
}

실전: 백그라운드 작업 큐

호스팅된 서비스와 채널을 사용하는 백그라운드 작업 큐의 예입니다. HTTP 요청이 작업을 큐에 넣으면 백그라운드 작업자가 작업을 한 번에 하나씩 처리합니다.

public class JobQueue
{
    private readonly Channel<Func<CancellationToken, Task>> _queue
        = Channel.CreateBounded<Func<CancellationToken, Task>>(100);

    public ChannelWriter<Func<CancellationToken, Task>> Writer => _queue.Writer;
    public ChannelReader<Func<CancellationToken, Task>> Reader => _queue.Reader;
}

public class JobProcessor : BackgroundService
{
    private readonly JobQueue _queue;
    public JobProcessor(JobQueue q) => _queue = q;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await foreach (var job in _queue.Reader.ReadAllAsync(ct))
            await job(ct);
    }
}

간단 확인

생산자가 가득 찬 bounded 채널에서 WriteAsync를 호출하면 어떻게 되나요?

복습: System.Threading.Channels

핵심 요점:

채널은 고성능의 비동기 중심 생산자-소비자 큐를 제공합니다
unbounded: 제한 없음, bounded: 역압 또는 삭제 전략이 적용되는 유한한 용량
여러 생산자와 소비자를 잠금 없이 안전하게 사용할 수 있습니다
ReadAllAsync() + await foreach는 가장 깔끔한 소비 패턴입니다
채널을 파이프라인으로 연결하면 단계별 동시 처리를 구현할 수 있습니다
스트림의 끝을 알리려면 Writer.Complete()(또는 Complete(exception))을 호출합니다

자주 묻는 질문

“System.Threading.Channels” 강의는 무료인가요?

네 — “System.Threading.Channels” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 C# Academy 강의 전체를 잠금 해제할 수 있습니다. C# Academy 강의에는 총 4개의 강의가 포함되어 있습니다.

“System.Threading.Channels”에서 뭘 배우나요?

백프레셔를 제어하기 위해 Channel, ChannelWriter, ChannelReader로 생산자-소비자 파이프라인을 구축합니다. 브라우저에서 직접 실행하는 실습 코드로 C# Academy을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

C# Academy을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 C# Academy은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.

“System.Threading.Channels” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 C# Academy 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 C# Academy 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← C# Academy(으)로 돌아가기