Как ограничить количество используемых потоков?

Ограничение количества потоков в C# Backend

В C# существуют различные подходы для ограничения количества одновременно выполняемых потоков, которые выбираются в зависимости от конкретной задачи и контекста.

Основные подходы к ограничению потоков

1. SemaphoreSlim для синхронизации

Наиболее гибкий и рекомендуемый подход для современных приложений:

// Создаем семафор с максимальным количеством одновременно выполняемых потоков
private static readonly SemaphoreSlim _semaphore = new SemaphoreSlim(5); // Максимум 5 потоков

public async Task ProcessItemsAsync(IEnumerable<Item> items)
{
    var tasks = items.Select(async item =>
    {
        // Ожидаем доступного слота
        await _semaphore.WaitAsync();
        
        try
        {
            // Критическая секция - выполняется не более 5 потоков одновременно
            await ProcessItemAsync(item);
        }
        finally
        {
            // Освобождаем слот
            _semaphore.Release();
        }
    });
    
    await Task.WhenAll(tasks);
}

2. ParallelOptions для параллельных операций

Для ограничения потоков в параллельных циклах:

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

Parallel.For(0, 1000, options, i =>
{
    // Выполняется не более MaxDegreeOfParallelism потоков одновременно
    ProcessData(i);
});

3. ThreadPool.SetMaxThreads для глобального ограничения

Внимание: Этот подход влияет на весь пул потоков и может иметь негативные последствия:

// Устанавливаем максимальное количество рабочих потоков и потоков ввода-вывода
ThreadPool.SetMaxThreads(50, 50);
ThreadPool.SetMinThreads(10, 10);

4. ActionBlock/TPL Dataflow для асинхронной обработки

Идеально подходит для pipeline-архитектуры:

var options = new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 3, // Максимум 3 параллельных обработки
    BoundedCapacity = 100       // Максимальная очередь
};

var actionBlock = new ActionBlock<Data>(async data =>
{
    await ProcessDataAsync(data);
}, options);

// Отправляем данные на обработку
foreach (var data in dataCollection)
{
    await actionBlock.SendAsync(data);
}

actionBlock.Complete();
await actionBlock.Completion;

5. ConcurrentExclusiveSchedulerPair для планировщиков задач

Позволяет создавать собственные планировщики с ограничениями:

var schedulerPair = new ConcurrentExclusiveSchedulerPair(
    TaskScheduler.Default, 
    maxConcurrencyLevel: 4);

var factory = new TaskFactory(schedulerPair.ConcurrentScheduler);

var tasks = Enumerable.Range(0, 20).Select(i => factory.StartNew(() =>
{
    // Выполняется не более 4 задач одновременно
    ProcessTask(i);
}));

await Task.WhenAll(tasks);

Практические рекомендации

Когда что использовать:

• SemaphoreSlim - для общего случая ограничения параллельных операций
• ParallelOptions - для параллельных циклов и обработки коллекций
• TPL Dataflow - для сложных асинхронных конвейеров обработки данных
• ConcurrentExclusiveSchedulerPair - когда нужен кастомный планировщик задач

Ключевые принципы:

1. Не злоупотребляйте глобальными ограничениями ThreadPool - это может привести к дедлокам
2. Учитывайте тип нагрузки - CPU-bound vs I/O-bound операции требуют разных подходов
3. Используйте асинхронные версии (WaitAsync у SemaphoreSlim) для избежания блокировок
4. Мониторьте производительность с помощью PerformanceCounters или Application Insights

Пример комплексного решения

public class ThrottledProcessor
{
    private readonly SemaphoreSlim _throttler;
    private readonly ILogger<ThrottledProcessor> _logger;
    
    public ThrottledProcessor(int maxConcurrentOperations, ILogger<ThrottledProcessor> logger)
    {
        _throttler = new SemaphoreSlim(maxConcurrentOperations);
        _logger = logger;
    }
    
    public async Task ProcessBatchAsync<T>(IEnumerable<T> items, Func<T, Task> processor)
    {
        var tasks = items.Select(async item =>
        {
            await _throttler.WaitAsync();
            
            try
            {
                await processor(item);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Ошибка обработки элемента");
            }
            finally
            {
                _throttler.Release();
            }
        });
        
        await Task.WhenAll(tasks);
    }
    
    // Метод с таймаутом ожидания
    public async Task<bool> TryProcessWithTimeoutAsync(Func<Task> operation, TimeSpan timeout)
    {
        if (await _throttler.WaitAsync(timeout))
        {
            try
            {
                await operation();
                return true;
            }
            finally
            {
                _throttler.Release();
            }
        }
        
        _logger.LogWarning("Таймаут ожидания свободного слота");
        return false;
    }
}

Производительность и отладка

Всегда мониторьте:

• Количество потоков в ThreadPool (ThreadPool.ThreadCount)
• Загрузку процессора
• Размер очереди ожидания
• Частоту коллизий при ожидании семафора

Оптимальное количество потоков зависит от конкретной задачи: для I/O-bound операций можно использовать больше потоков, для CPU-bound - обычно ограничиваются количеством ядер процессора.