Ограничение количества одновременных System.Threading.Tasks.Task

Question

Мне нужно выполнить обширные вставки данных в мою базу данных. Я могу реализовать код многопоточным способом с дросселированным планировщиком, который ограничивает количество параллельных операций. На каждом M строках блок формируется и вводится в базу данных как атомная операция. Несколько параллельных операций должны происходить из-за того, что база данных работает медленнее, чем чтение и анализ файла данных. Я часто реализую эту модель с помощью многопоточности.

Если вместо этого я решил реализовать свой код, используя Await/асинхронный (Entity Framework поддерживает асинхронное программирование), как я могу убедиться, что не более чем N параллельных задач выполнения (т.е. перейти в базу данных), в то же время?

В моем первоначальном дизайне я создал экземпляр List<Task>, добавив новые задачи, как только я прочитал блок данных, который нужно вставить атомарно, а затем позвольте моему методу вернуться после await всей задачи. Проблема времени разработки заключается в том, что количество параллельных Task s (и, следовательно, объем памяти) будет взорваться, потому что задачи будут загружаться быстрее, чем они завершаются для больших файлов данных.

Я думал об использовании SemaphoreSlim, но у меня мало опыта с асинхронным программированием (в отличие от многопоточного). Поэтому я задаю этот вопрос, чтобы получить отзывы о лучших практиках, если они есть.

Answer 1

вопрос время разработки является то, что число параллельных задач (и, таким образом память след) собирается взорвать, потому что задачи подаются быстрее, чем они завершаются для больших файлов данных. Я думал об использовании SemaphoreSlim

Да, SemaphoreSlim является подходящим выбором для дросселирования параллельных асинхронных операций:

private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(10); 

async Task ThrottledWorkAsync() 
{ 
    await _semaphore.WaitAsync(); 
    try 
    { 
    await WorkAsync(); 
    } 
    finally 
    { 
    _semaphore.Release(); 
    } 
} 

Однако ...

Если вместо этого я решил осуществить свою код с использованием await/async (Entity Framework поддерживает асинхронное программирование), как я могу убедиться, что одновременно выполняются не более N одновременных задач (т.е. перейдите в базу данных)?

Одна вещь, о которой следует знать, заключается в том, что Entity Framework - при поддержке асинхронных API - по-прежнему требует одного соединения для каждого запроса. Таким образом, вы не можете иметь несколько одновременных асинхронных запросов с тем же DbContext; вам необходимо создать отдельное соединение для каждого параллельного запроса (или, по крайней мере, N соединений, которые «заимствованы» параллельными запросами).

Answer 2

Я использую этот кусок кода, чтобы выполнить мои темы:

public static async Task WhenAll(this List<Func<Task>> actions, int threadCount) 
{ 
    var _countdownEvent = new CountdownEvent(actions.Count); 
    var _throttler = new SemaphoreSlim(threadCount); 

    foreach (Func<Task> action in actions) 
    { 
     await _throttler.WaitAsync(); 

     Task.Run(async() => 
     { 
      try 
      { 
       await action(); 
      } 
      finally 
      { 
       _throttler.Release(); 
       _countdownEvent.Signal(); 
      } 
     }); 
    } 

    _countdownEvent.Wait(); 
} 

Этот код основан на коде, представленной в этой теме: How to limit the amount of concurrent async I/O operations?

Calling это должно выглядеть примерно так. Она будет выполнять все задачи, но только 40 (в данном случае) одновременно:

var tasks = new List<Func<Task>>(); 
tasks.Add(() => saveAsync()); 
//add more 
await tasks.WhenAll(40); 

Answer 3

Если у вас есть по крайней мере n значения для вставки первоначально (n быть максимальное количество одновременных задач), вы можете принять следующий подход:

1. Invoke InsertAsync()n раз с разными значениями.
2. Когда каждая из задач завершена, продолжайте новый вызов до InsertAsync() (повторите 2).

Таким образом, вам не нужно будет контролировать уровень параллелизма с помощью семафора и будет неблокировать.

I've just published a package, которые могут быть полезны для этого сценария, он предоставляет 2 метода Times() и Map(): https://github.com/jorgebay/concurrent-utils

Например:

// Execute MyMethodAsync() 1,000,000 times limiting the maximum amount 
// of parallel async operations to 512 
await ConcurrentUtils.Times(1000000, 512, (index) => MyMethodAsync(index));