Конкурентность и многопоточность в C#: продвинутый уровень

Конкурентность и многопоточность в C#: продвинутый уровень

В современной разработке на C# (.NET) умение эффективно управлять конкурентностью и многопоточностью отличает опытного инженера от новичка. Базовые знания о Thread и простых блокировках уже недостаточны для создания высокопроизводительных, отзывчивых и масштабируемых приложений. В этой статье мы разберем продвинутые техники: от глубинного понимания async/await до использования System.Threading.Channels и TPL Dataflow. Материал ориентирован на разработчиков, уже знакомых с основами многопоточности.

1. Глубокое погружение в async/await и SynchronizationContext

Многие разработчики используют async/await, но не до конца понимают, как именно планировщик задач (Task Scheduler) и SynchronizationContext управляют продолжениями (continuations). На продвинутом уровне важно контролировать, в каком потоке выполняется код после await.

По умолчанию await захватывает текущий SynchronizationContext (или TaskScheduler). В UI-приложениях (WPF, WinForms) это возвращает выполнение в главный поток. В ASP.NET Core и консольных приложениях контекст часто отсутствует, и продолжение может выполняться в любом потоке пула.

Для тонкой настройки используйте ConfigureAwait(false). Это говорит среде выполнения не пытаться вернуться в исходный контекст, что повышает производительность и предотвращает взаимные блокировки (deadlocks) в библиотечном коде.

public async Task<string> FetchDataAsync(HttpClient client)
{
    // Продолжение не будет захватывать контекст
    var result = await client.GetStringAsync("https://api.example.com")
                              .ConfigureAwait(false);
    // Этот код выполнится в любом потоке пула
    return ProcessResult(result);
}

public async Task<string> FetchDataWithContextAsync(HttpClient client)
{
    // Продолжение вернется в исходный контекст (например, UI поток)
    var result = await client.GetStringAsync("https://api.example.com");
    // Этот код гарантированно выполнится в исходном контексте
    return ProcessResult(result);
}

Продвинутый совет: Создавайте собственные SynchronizationContext для нестандартных сценариев, например, для организации очереди задач с фиксированным числом потоков.

2. Task Parallel Library (TPL) и параллелизм данных

TPL предоставляет высокоуровневые абстракции для параллелизма. Parallel.For и Parallel.ForEach автоматически разбивают коллекции на части (chunks) и распределяют их по потокам. Однако на продвинутом уровне нужно понимать, как настроить степень параллелизма и обработку исключений.

Используйте ParallelOptions для ограничения числа потоков, а также ParallelLoopState для досрочного выхода из цикла.

var options = new ParallelOptions
{
    MaxDegreeOfParallelism = Environment.ProcessorCount / 2 // Используем половину ядер
};

Parallel.ForEach(sourceData, options, (item, state) =>
{
    try
    {
        // Тяжелая операция
        ProcessItem(item);
    }
    catch (Exception ex)
    {
        // Логируем и прерываем выполнение
        Console.WriteLine($"Ошибка: {ex.Message}");
        state.Stop(); // Прерывает выполнение цикла
    }
});

Важно: Parallel.ForEach не гарантирует порядок обработки. Если порядок важен, используйте Parallel.ForEach с индексом или PLINQ (AsParallel().AsOrdered()).

3. Каналы (System.Threading.Channels) для Producer-Consumer

Класс Channel<T> появился в .NET Core 3.0 и стал мощным инструментом для сценариев производитель-потребитель. Он потокобезопасен, асинхронен и эффективен. Каналы бывают двух типов: неограниченные (Unbounded) и ограниченные (Bounded) с фиксированной емкостью.

Ограниченные каналы с BoundedChannelFullMode.Wait позволяют продюсеру ждать, пока потребитель освободит место — это естественная форма backpressure (обратного давления).

public class DataPipeline
{
    private readonly Channel<int> _channel;

    public DataPipeline()
    {
        // Ограниченный канал на 100 элементов с ожиданием
        var options = new BoundedChannelOptions(100)
        {
            FullMode = BoundedChannelFullMode.Wait
        };
        _channel = Channel.CreateBounded<int>(options);
    }

    public async Task ProducerAsync(IEnumerable<int> data)
    {
        var writer = _channel.Writer;
        foreach (var item in data)
        {
            await writer.WriteAsync(item);
            Console.WriteLine($"Произведено: {item}");
        }
        writer.Complete(); // Сигнал об окончании
    }

    public async Task ConsumerAsync(CancellationToken token)
    {
        var reader = _channel.Reader;
        await foreach (var item in reader.ReadAllAsync(token))
        {
            Console.WriteLine($"Потреблено: {item}");
            // Симуляция обработки
            await Task.Delay(50, token);
        }
    }
}

Продвинутый паттерн: Используйте несколько каналов для построения конвейера обработки, где каждый этап работает в своем пуле потоков.

4. TPL Dataflow: построение конвейеров обработки

Библиотека System.Threading.Tasks.Dataflow предоставляет блоки для создания сложных сетей обработки данных. Основные блоки: BufferBlock<T>, ActionBlock<T>, TransformBlock<TInput, TOutput> и BatchBlock<T>.

Dataflow автоматически управляет степенью параллелизма, буферизацией и обработкой ошибок. Вы можете настроить ExecutionDataflowBlockOptions для контроля.

public async Task ProcessWithDataflowAsync(IEnumerable<string> urls)
{
    var downloadBlock = new TransformBlock<string, string>(async url =>
    {
        using var client = new HttpClient();
        var content = await client.GetStringAsync(url);
        Console.WriteLine($"Скачано: {url}");
        return content;
    }, new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 4 // 4 параллельных скачивания
    });

    var processBlock = new ActionBlock<string>(data =>
    {
        // Обработка данных (синхронная)
        var result = data.ToUpper();
        Console.WriteLine($"Обработано: {result.Length} символов");
    });

    // Связываем блоки: выход downloadBlock -> вход processBlock
    downloadBlock.LinkTo