C# Academy · Урок

System.Threading.Channels

Создавайте конвейеры «производитель — потребитель» с помощью Channel, ChannelWriter и ChannelReader для управления противодавлением.

«System.Threading.Channels» — бесплатный урок C# Academy на CoddyKit. Это урок 2 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения C# Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс C# Academy содержит 4 уроков всего.

Что такое каналы?

System.Threading.Channels (представленный в .NET Core 3) предоставляет высокопроизводительную потокобезопасную очередь producer-consumer. В отличие от 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
});

Запись в канал

Producers записывают элементы с помощью WriteAsync (ожидает, если bounded channel заполнен) или TryWrite (немедленно возвращает false, если channel заполнен). Завершайте запись вызовом 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
});

Чтение из канала

Consumers читают данные с помощью ReadAllAsync() (самый простой подход) или ReadAsync/TryRead, если требуется больший контроль. ReadAllAsync завершается, когда writer вызывает Complete().

// 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);

Несколько producers

Каналы потокобезопасны. Несколько producers могут записывать данные одновременно — блокировки не требуются. Вызывайте Writer.Complete() только после завершения работы всех producers.

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();

Несколько consumers (распределение нагрузки)

Запустите несколько задач consumer, читающих из одного channel, чтобы распараллелить обработку. Каждый элемент доставляется ровно одному consumer (распределяется, а не рассылается всем).

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 channel автоматически применяет обратное давление: когда channel заполнен, WriteAsync приостанавливает producer, пока consumer не извлечёт элементы. Дополнительный код для ограничения скорости не нужен.

// 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), чтобы распространить ошибку на всех ожидающих readers. Readers увидят исключение при следующем чтении из channel.

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}");
}

Практический пример: очередь фоновых заданий

Очередь фоновых заданий, использующая hosted service и channel: HTTP-запросы помещают задания в очередь, а фоновый worker обрабатывает их по одному.

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);
    }
}

Быстрая проверка

Что происходит, когда producer вызывает WriteAsync для заполненного bounded channel?

Повторение: System.Threading.Channels

Основные выводы:

Каналы предоставляют высокопроизводительную очередь producer-consumer, изначально ориентированную на async
Unbounded: без ограничения; Bounded: конечная ёмкость с обратным давлением или стратегиями отбрасывания
Несколько producers и consumers безопасно работают без блокировок
ReadAllAsync() + await foreach — самый простой шаблон потребления
Объединяйте каналы в конвейеры для параллельной обработки по этапам
Вызывайте Writer.Complete() (или Complete(exception)), чтобы обозначить конец потока

Часто задаваемые вопросы

Урок «System.Threading.Channels» бесплатный?

Да — полный текст урока «System.Threading.Channels» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс C# Academy, подпишись на CoddyKit PRO. Курс C# Academy содержит 4 уроков всего.

Чему я научусь в уроке «System.Threading.Channels»?

Создавайте конвейеры «производитель — потребитель» с помощью Channel, ChannelWriter и ChannelReader для управления противодавлением. Ты практикуешь C# Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать C# Academy?

Предыдущий опыт не требуется. C# Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 2 из 4.

Сколько времени занимает урок «System.Threading.Channels»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке C# Academy?

Да. Каждый урок C# Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

← Назад к C# Academy