RU
EN
ES
ZH
FR
DE
PT
AR
HI
Очереди задач для C# (.NET): System.Threading.Channels и TPL Dataflow
При разработке многопоточных приложений на C# (.NET) часто возникает необходимость организовать асинхронную передачу данных между производителями (producers) и потребителями (consumers). Классический паттерн Producer-Consumer решается с помощью очередей задач. В экосистеме .NET есть два мощных инструмента для этого: System.Threading.Channels (появился в .NET Core 3.0) и библиотека TPL Dataflow (часть .NET с 4.5).
В этой статье мы разберём оба подхода, их особенности и приведём практические примеры кода.
1. System.Threading.Channels — лёгкая и быстрая очередь
Пространство имён System.Threading.Channels предоставляет потокобезопасные, асинхронные очереди, оптимизированные для сценариев с высокой пропускной способностью. Основные типы:
Channel<T>— абстрактный класс канала.Channel.CreateUnbounded<T>()— неограниченная очередь (риск переполнения памяти).Channel.CreateBounded<T>(int capacity)— ограниченная очередь с контролем давления.ChannelWriter<T>— интерфейс для записи.ChannelReader<T>— интерфейс для чтения.
Пример: Producer-Consumer с Channel
Создадим простой пример, где один производитель генерирует числа, а один потребитель их обрабатывает.
using System.Threading.Channels;
using System.Threading.Tasks;
var channel = Channel.CreateUnbounded<int>();
// Producer
var producer = Task.Run(async () =>
{
for (int i = 0; i < 10; i++)
{
await channel.Writer.WriteAsync(i);
Console.WriteLine($"Произведено: {i}");
await Task.Delay(100);
}
channel.Writer.Complete(); // Сигнал об окончании
});
// Consumer
var consumer = Task.Run(async () =>
{
await foreach (var item in channel.Reader.ReadAllAsync())
{
Console.WriteLine($"Обработано: {item * 2}");
await Task.Delay(200);
}
});
await Task.WhenAll(producer, consumer);
Console.WriteLine("Готово!");Преимущества Channels:
- Минимальные накладные расходы (без выделения дополнительных объектов).
- Встроенная поддержка
async/awaitиIAsyncEnumerable. - Идеально для высоконагруженных систем (например, обработка сетевых пакетов).
2. TPL Dataflow — мощный конвейер обработки
Библиотека System.Threading.Tasks.Dataflow предоставляет более сложные блоки для построения конвейеров (pipeline) и сетей обработки данных. Основные блоки:
BufferBlock<T>— простая очередь.ActionBlock<T>— блок, выполняющий действие над каждым элементом.TransformBlock<TInput, TOutput>— преобразует каждый элемент.BatchBlock<T>— группирует элементы в батчи.
Пример: Конвейер с ActionBlock
Реализуем ту же задачу: производитель отправляет числа, потребитель обрабатывает.
using System.Threading.Tasks.Dataflow;
var actionBlock = new ActionBlock<int>(async item =>
{
await Task.Delay(200);
Console.WriteLine($"Обработано: {item * 2}");
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 2 });
// Producer
for (int i = 0; i < 10; i++)
{
await actionBlock.SendAsync(i);
Console.WriteLine($"Произведено: {i}");
await Task.Delay(100);
}
actionBlock.Complete();
await actionBlock.Completion;
Console.WriteLine("Готово!");Преимущества TPL Dataflow:
- Гибкая настройка параллелизма (
MaxDegreeOfParallelism). - Поддержка отмены через
CancellationToken. - Возможность связывать блоки в сложные графы (например,
TransformBlock→ActionBlock). - Встроенное управление давлением (bounded capacity).
