Асинхронное и параллельное программирование на C#

// Создание потока напрямую - тяжеловесный подход
Thread thread = new Thread(() => Console.WriteLine("Работаю в отдельном потоке"));
thread.Start();
 
// Создание задачи - современный подход
Task task = Task.Run(() => Console.WriteLine("Задача выполняется в пуле потоков"));

// Базовый пример асинхронного метода
public async Task<string> GetWebPageContentAsync(string url)
{
    using (HttpClient client = new HttpClient())
    {
        // Await "подписывается" на завершение задачи и временно 
        // освобождает текущий поток для выполнения другой работы
        string content = await client.GetStringAsync(url);
        return content.ToUpper(); // Какая-то обработка
    }
}

// CPU-bound асинхронная операция
public async Task<long> CalculateFactorialAsync(int n)
{
    return await Task.Run(() => {
        long result = 1;
        for (int i = 1; i <= n; i++)
            result *= i;
        return result;
    });
}

// Оптимизированная версия с ValueTask
public ValueTask<long> OptimizedFactorialAsync(int n)
{
    // Для маленьких чисел не создаем Task, возвращаем моментально
    if (n <= 1)
        return new ValueTask<long>(1);
        
    // Только для "тяжелых" случаев запускаем асинхронный расчет
    return new ValueTask<long>(CalculateFactorialAsync(n));
}

// В библиотечном коде так правильнее
public async Task<byte[]> ComputeHashAsync(string input)
{
    byte[] inputBytes = Encoding.UTF8.GetBytes(input);
    // Нет нужды возвращаться в оригинальный контекст
    return await Task.Run(() => MD5.Create().ComputeHash(inputBytes))
                     .ConfigureAwait(false);
}

// Превращение события в задачу
public Task<int> ReadFromSerialPortAsync(SerialPort port)
{
var tcs = new TaskCompletionSource<int>();
 
// Обработчики событий синхронного API
port.DataReceived += (s, e) => {
    try {
        int data = port.ReadByte();
        tcs.SetResult(data);  // Сигнализируем о завершении задачи
    } 
    catch (Exception ex) {
        tcs.SetException(ex);  // Или об ошибке
    }
};
 
port.ErrorReceived += (s, e) => tcs.SetException(
    new IOException("Serial port error"));
 
// Возвращаем задачу, которая завершится когда придут данные
return tcs.Task;
}

// Старый подход с колбэками
device.StartOperation(param, result => {
    if (result.Success) {
        device.GetData(data => {
            ProcessData(data);
            device.Cleanup(cleanup => {
                // Ад колбэков
            });
        });
    }
});

// С использованием асинхронных оберток
await device.StartOperationAsync(param);
var data = await device.GetDataAsync();
ProcessData(data);
await device.CleanupAsync();

// Операция с таймаутом
public async Task<string> FetchWithTimeoutAsync(string url, TimeSpan timeout)
{
using var client = new HttpClient();
    
// Создаем задачу, которая выбросит исключение после таймаута
var timeoutTask = Task.Delay(timeout).ContinueWith(_ => 
    throw new TimeoutException("The operation has timed out"));
    
// Основная задача запроса
var fetchTask = client.GetStringAsync(url);
    
// Ждем завершения любой из задач
var completedTask = await Task.WhenAny(fetchTask, timeoutTask);
    
// Если победил запрос, возвращаем результат, иначе исключение пробросится
return await completedTask;
}

Task.Run(() => ComputeFirstPart())
    .ContinueWith(prev => ComputeSecondPart(prev.Result))
    .ContinueWith(prev => ComputeThirdPart(prev.Result))
    .ContinueWith(prev => {
        Console.WriteLine("Final result: " + prev.Result);
    }, TaskScheduler.FromCurrentSynchronizationContext());

// Задача запускается один раз
var task = FetchDataAsync();
 
// Эти два await относятся к одной и той же операции
var result1 = await task;
var result2 = await task; // Не запускает второй запрос!

// Даёт шанс UI обновиться во время тяжелых вычислений
public async Task ProcessItemsAsync(IEnumerable<Item> items)
{
int count = 0;
foreach (var item in items)
{
    ProcessItem(item);
    count++;
    
    // Периодически "уступаем" поток UI
    if (count % 100 == 0)
        await Task.Yield();
}
}

// ОПАСНО! Может привести к дедлоку в UI-приложениях
string content = httpClient.GetStringAsync(url).Result;
 
// То же самое, но с обработкой исключений
try
{
    httpClient.GetStringAsync(url).Wait();
}
catch (AggregateException ae)
{
    // Исключения из задачи оборачиваются в AggregateException
}

// Адаптер для APM к TAP
public Task<Stream> ReadFileAPMAsync(string filePath)
{
    // Создаём TaskCompletionSource для получения результата
    FileStream fs = new FileStream(filePath, FileMode.Open);
    
    // Волшебный метод превращает APM в Task
    return Task<Stream>.Factory.FromAsync(
        fs.BeginRead, fs.EndRead, new byte[100], 0, 100, null);
}

// Делайте так
public string GetData(string key)
{
    // Синхронная версия использует асинхронную
    return GetDataAsync(key).GetAwaiter().GetResult();
}
 
public async Task<string> GetDataAsync(string key)
{
    // Основная реализация асинхронна
    using var client = new HttpClient();
    return await client.GetStringAsync($"https://api.example.com/data/{key}");
}

downloadTask.ContinueWith(task => {
    // Выполнится только если операция завершилась ошибкой
    LogError(task.Exception);
}, TaskContinuationOptions.OnlyOnFaulted);
 
calculateTask.ContinueWith(task => {
    // Выполнится только если операция успешно завершилась
    SaveResult(task.Result);
}, TaskContinuationOptions.OnlyOnRanToCompletion);

// Захватываем текущий (UI) шедулер
var uiScheduler = TaskScheduler.FromCurrentSynchronizationContext();
 
Task.Run(() => PerformHeavyCalculation())
    .ContinueWith(task => {
        // Это продолжение выполнится в UI-потоке
        resultTextBox.Text = task.Result.ToString();
    }, uiScheduler);

// Запускаем несколько операций параллельно
Task<int>[] downloadTasks = urls.Select(url => DownloadAndCountBytesAsync(url)).ToArray();
 
// Ждём завершения всех загрузок
await Task.WhenAll(downloadTasks);
 
// Или так, если нужны все результаты в виде массива
int[] byteCounts = await Task.WhenAll(downloadTasks);

Task allTasks = Task.WhenAll(tasks);
try
{
    await allTasks;
}
catch
{
    // Внутри catch мы можем получить все исключения
    AggregateException allExceptions = allTasks.Exception;
    // Обработать каждое из исключений
}

// Создаем блок, который будет преобразовывать данные
var transformBlock = new TransformBlock<string, int>(
    data => Compute(data),  // Функция трансформации
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism = Environment.ProcessorCount
    }
);
 
// И блок, который будет принимать результаты
var actionBlock = new ActionBlock<int>(
    result => Console.WriteLine(result)
);
 
// Связываем блоки в конвейер
transformBlock.LinkTo(actionBlock, new DataflowLinkOptions { PropagateCompletion = true });
 
// Отправляем данные в конвейер
foreach (var item in items)
    transformBlock.Post(item);
 
// Сигнализируем, что входные данные закончились
transformBlock.Complete();
 
// Ждем полного завершения обработки
await actionBlock.Completion;

public async Task PollServiceUntilSuccessAsync(string endpoint, TimeSpan interval, CancellationToken token)
{
while (!token.IsCancellationRequested)
{
    try
    {
        var result = await CallServiceAsync(endpoint);
        if (result.IsSuccess)
            return; // Задача выполнена успешно
    }
    catch (Exception ex)
    {
        Log.Error($"Service call failed: {ex.Message}");
    }
    
    // Ждём заданный интервал перед следующей попыткой
    await Task.Delay(interval, token);
}
}

// НЕПРАВИЛЬНО! Исключение будет проглочено
Task.Run(() => { throw new Exception("Boom!"); });
 
// ПРАВИЛЬНО: всегда ожидайте задачи или обрабатывайте ошибки
try 
{
    await Task.Run(() => { throw new Exception("Boom!"); });
} 
catch (Exception ex) 
{
    // Обработка ошибки
}

Task[] tasks = new Task[10];
for (int i = 0; i < 10; i++)
{
    int taskNum = i; // Избегаем захвата переменной цикла
    tasks[i] = Task.Run(() => ProcessItem(taskNum));
}
 
try
{
    // Эта строка выбросит первое исключение, если оно было
    Task.WaitAll(tasks);
}
catch (AggregateException ae)
{
    // А здесь получим и обработаем все исключения
    foreach (var ex in ae.Flatten().InnerExceptions)
    {
        Console.WriteLine($"Task failed: {ex.Message}");
    }
}

public async Task ProcessFilesAsync(string[] files, CancellationToken token)
{
    foreach (var file in files)
    {
        // Проверяем токен отмены перед каждой тяжёлой операцией
        token.ThrowIfCancellationRequested();
        
        // Передаём токен во внутренние асинхронные вызовы 
        await ProcessFileAsync(file, token);
        
        // Комбинируем с таймаутом
        await Task.Delay(100, token);
    }
}
 
// Использование:
var cts = new CancellationTokenSource();
var task = ProcessFilesAsync(files, cts.Token);
 
// Где-то в другом потоке или через 5 секунд:
cts.CancelAfter(TimeSpan.FromSeconds(5));
// или cts.Cancel() для моментальной отмены

// Традиционный подход
var result1 = collection.Where(item => IsExpensive(item))
                       .Select(item => Transform(item))
                       .ToList();
 
// Параллельный подход с PLINQ
var result2 = collection.AsParallel()
                       .Where(item => IsExpensive(item))
                       .Select(item => Transform(item))
                       .ToList();

// Метод, возвращающий асинхронную последовательность
public async IAsyncEnumerable<string> GetLinesFromBigFileAsync(string path)
{
    using var file = new StreamReader(path);
    while (!file.EndOfStream)
    {
        // Асинхронно читаем строку
        string line = await file.ReadLineAsync();
        // И "возвращаем" её потребителю перед чтением следующей
        yield return line;
    }
}
 
// Использование с await foreach
await foreach (var line in GetLinesFromBigFileAsync("huge.txt"))
{
    await ProcessLineAsync(line);
}

// Этот код создаст deadlock в UI-приложении:
private void Button_Click(object sender, EventArgs e)
{
    string result = GetDataAsync().Result; // Deadlock!
    resultLabel.Text = result;
}
 
private async Task<string> GetDataAsync()
{
    await Task.Delay(1000); // Задача пытается вернуться в UI-поток
    return "Data";
}

private async void Button_Click(object sender, EventArgs e)
{
    string result = await GetDataAsync(); // Больше никаких dедлоков!
    resultLabel.Text = result;
}

private async Task<string> GetDataAsync()
{
    await Task.Delay(1000).ConfigureAwait(false); // Не возвращаемся в UI-поток
    return "Data";
}

// Изменение настроек пула потоков для масштабирования
ThreadPool.SetMinThreads(100, 100); // минимум рабочих/IO потоков
ThreadPool.SetMaxThreads(1000, 1000); // максимум рабочих/IO потоков

// Создаём канал с ограниченной емкостью
var channel = Channel.CreateBounded<WorkItem>(100);
 
// Производитель
async Task ProduceAsync(CancellationToken token)
{
    for (int i = 0; i < 1000 && !token.IsCancellationRequested; i++)
    {
        await channel.Writer.WriteAsync(new WorkItem { Id = i }, token);
    }
    channel.Writer.Complete(); // Сигнализируем о завершении записи
}
 
// Потребитель
async Task ConsumeAsync()
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        await ProcessItemAsync(item);
    }
}
 
// Запускаем оба процеса параллельно
await Task.WhenAll(ProduceAsync(cts.Token), ConsumeAsync());

// Запускаем несколько потребителей
var consumers = Enumerable.Range(0, Environment.ProcessorCount)
    .Select(_ => ConsumeAsync())
    .ToArray();
 
// Ждём завершения всех потребителей
await Task.WhenAll(consumers);

// Параллельный цикл с разделением данных между ядрами
Parallel.ForEach(
    bigDataCollection,
    new ParallelOptions { 
        MaxDegreeOfParallelism = Environment.ProcessorCount 
    },
    item => {
        // CPU-интенсивная обработка
        ProcessDataItem(item);
    }
);

// Параллельное суммирование большого массива
int[] numbers = GetLargeArray();
long sum = numbers.AsParallel().Aggregate(
    0L, // начальное значение для каждого потока
    (acc, num) => acc + num, // функция накопления в каждом потоке
    (total, threadResult) => total + threadResult, // объединение результатов
    result => result // финальное преобразование
);

ConcurrentBag<Result> results = new ConcurrentBag<Result>();
Parallel.ForEach(items, item =>
{
    // Локальный список для каждого потока
    List<Result> localResults = new List<Result>();
    
    // Обрабатываем партию элементов
    foreach (var subItem in item.SubItems)
    {
        var result = ProcessItem(subItem);
        localResults.Add(result);
    }
    
    // Одна операция с общим списком вместо многих
    foreach (var result in localResults)
        results.Add(result);
});

// Антипаттерн! Гонка данных практически гарантирована
private static string _cachedData;
 
public async Task<string> GetDataAsync()
{
if (_cachedData == null)
{
    // Проблема: два потока могут одновременно пройти эту проверку
    _cachedData = await FetchExpensiveDataAsync();
}
return _cachedData;
}

private static Task<string> _cachedDataTask;
private static readonly object _lock = new object();
 
public Task<string> GetDataAsync()
{
if (_cachedDataTask == null)
{
    lock (_lock)
    {
        // Двойная проверка для избежания блокировки на горячем пути
        if (_cachedDataTask == null)
        {
            _cachedDataTask = FetchExpensiveDataAsync();
        }
    }
}
return _cachedDataTask;
}

// Антипаттерн: UI зависнет!
private void SearchButton_Click(object sender, EventArgs e)
{
var results = SearchDatabase(searchQuery.Text);
resultsListBox.DataSource = results;
}

private async void SearchButton_Click(object sender, EventArgs e)
{
searchButton.Enabled = false;
progressBar.Visible = true;
 
try
{
    var results = await Task.Run(() => SearchDatabase(searchQuery.Text));
    resultsListBox.DataSource = results;
}
catch (Exception ex)
{
    MessageBox.Show($"Error: {ex.Message}");
}
finally
{
    searchButton.Enabled = true;
    progressBar.Visible = false;
}
}

private CancellationTokenSource _cts;
 
private async void StartProcessingButton_Click(object sender, EventArgs e)
{
_cts = new CancellationTokenSource();
startButton.Visible = false;
cancelButton.Visible = true;
 
try
{
    await ProcessDataWithProgressAsync(data, UpdateProgressBar, _cts.Token);
    resultLabel.Text = "Processing complete!";
}
catch (OperationCanceledException)
{
    resultLabel.Text = "Operation cancelled.";
}
finally
{
    startButton.Visible = true;
    cancelButton.Visible = false;
}
}
 
private void CancelButton_Click(object sender, EventArgs e)
{
_cts?.Cancel();
}

// Создаём буферизованный конвейер обработки с ограничением одновременных операций
var downloadBlock = new TransformBlock<string, byte[]>(
    async url => await DownloadDataAsync(url),
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism = 16, // Ограничение числа одновременных загрузок
        BoundedCapacity = 100 // Ограничение буфера для предотвращения утечек памяти
    }
);
 
var processBlock = new TransformBlock<byte[], ProcessedData>(
    data => ProcessData(data),
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism = Environment.ProcessorCount // Используем все ядра
    }
);
 
var saveBlock = new ActionBlock<ProcessedData>(
    async data => await SaveToDbAsync(data),
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism = 32 // Оптимизировано под конкретную БД
    }
);
 
// Связываем блоки в конвейер
downloadBlock.LinkTo(processBlock, new DataflowLinkOptions { PropagateCompletion = true });
processBlock.LinkTo(saveBlock, new DataflowLinkOptions { PropagateCompletion = true });
 
// Запускаем конвейер
foreach (var url in urlsToProcess)
    await downloadBlock.SendAsync(url); // Ожидаем, если буфер полон (backpressure)
 
downloadBlock.Complete(); // Сигнализируем о завершении ввода
await saveBlock.Completion; // Ждём завершения всей работы

// Разбиваем крупную задачу на мелкие части
var batchResults = inputData
    .Batch(100) // Расширение, группирующее элементы по 100 штук
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .Select(async batch => {
        // Для каждой партии асинхронно получаем данные
        var enrichedData = await FetchAdditionalDataAsync(batch);
        // И выполняем CPU-интенсивную обработку локально
        return enrichedData.Select(ProcessLocally).ToList();
    })
    .ToArray();
 
// Ждём завершения всех задач и объединяем результаты
var results = (await Task.WhenAll(batchResults)).SelectMany(x => x).ToList();

// Создаём источник событий
var eventSource = Observable.FromEventPattern<DataEventArgs>(
    h => source.DataReceived += h,
    h => source.DataReceived -= h
);
 
// Строим реактивный пайплайн
var subscription = eventSource
    .Select(e => e.EventArgs.Data)
    .Buffer(TimeSpan.FromSeconds(5), 100) // Группируем данные по времени или количеству
    .Where(buffer => buffer.Any())
    .SelectMany(async buffer => {
        // Асинхронно обрабатываем каждую группу данных
        return await ProcessBatchAsync(buffer);
    })
    .Retry(3) // Автоматически повторяем при ошибках
    .Subscribe(
        result => Console.WriteLine($"Processed: {result}"),
        ex => Console.WriteLine($"Error: {ex.Message}"),
        () => Console.WriteLine("Completed")
    );
 
// Когда нужно остановить обработку
subscription.Dispose();

// Опасный код! Может привести к исчерпанию соединений
var tasks = Enumerable.Range(0, 10000)
    .Select(i => httpClient.GetStringAsync($"https://api.example.com/item/{i}"))
    .ToList();
 
await Task.WhenAll(tasks);

// Контролируемая параллельность с SemaphoreSlim
using var throttler = new SemaphoreSlim(initialCount: 100);
 
var tasks = new List<Task>();
foreach (var id in Enumerable.Range(0, 10000))
{
    await throttler.WaitAsync(); // Ждём, если достигнут лимит параллельных задач
    
    tasks.Add(Task.Run(async () => {
        try
        {
            return await httpClient.GetStringAsync($"https://api.example.com/item/{id}");
        }
        finally
        {
            throttler.Release(); // Важно освобождать даже при ошибках
        }
    }));
}
 
await Task.WhenAll(tasks);