Регулирование асинхронных задач

Как было предложено, используйте поток данных TPL.

TransformBlock<TInput, TOutput> может быть тем, что ты ищешь.

Вы определяете MaxDegreeOfParallelism, чтобы ограничить количество строк, которые могут быть преобразованы (т. Е. Сколько URL-адресов могут быть загружены) параллельно. Затем вы отправляете URL-адреса в блок, и когда вы закончите, вы сообщаете блоку, что добавили элементы, и получаете ответы.

var downloader = new TransformBlock<string, HttpResponse>(
        url => Download(url),
        new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 50 }
    );

var buffer = new BufferBlock<HttpResponse>();
downloader.LinkTo(buffer);

foreach(var url in urls)
    downloader.Post(url);
    //or await downloader.SendAsync(url);

downloader.Complete();
await downloader.Completion;

IList<HttpResponse> responses;
if (buffer.TryReceiveAll(out responses))
{
    //process responses
}

Примечание: TransformBlock буферизует как ввод, так и вывод. Тогда зачем нам связывать его с BufferBlock?

Потому что TransformBlock не завершится, пока не будут израсходованы все элементы (HttpResponse), и await downloader.Completion зависнет. Вместо этого мы позволяем downloader пересылать весь свой вывод в выделенный буферный блок - затем мы ждем завершения downloader и проверяем этот буферный блок.

Допустим, у вас есть 1000 URL-адресов, и вы хотите, чтобы одновременно было открыто только 50 запросов; но как только один запрос завершается, вы открываете соединение со следующим URL-адресом в списке. Таким образом, одновременно всегда открыто ровно 50 подключений, пока список URL-адресов не будет исчерпан.

Следующее простое решение неоднократно появлялось здесь, на SO. Он не использует код блокировки и не создает потоки явно, поэтому очень хорошо масштабируется:

const int MAX_DOWNLOADS = 50;

static async Task DownloadAsync(string[] urls)
{
    using (var semaphore = new SemaphoreSlim(MAX_DOWNLOADS))
    using (var httpClient = new HttpClient())
    {
        var tasks = urls.Select(async url => 
        {
            await semaphore.WaitAsync();
            try
            {
                var data = await httpClient.GetStringAsync(url);
                Console.WriteLine(data);
            }
            finally
            {
                semaphore.Release();
            }
        });

        await Task.WhenAll(tasks);
    }
}

Дело в том, что обработка загруженных данных должна выполняться в другом конвейере с другим уровнем параллелизма, особенно если это Обработка с привязкой к ЦП.

Например, вы, вероятно, захотите иметь 4 потока, одновременно выполняющих обработку данных (количество ядер ЦП), и до 50 ожидающих запросов для дополнительных данных (которые вообще не используют потоки). AFAICT, это не то, что сейчас делает ваш код.

Вот где TPL Dataflow или Rx могут пригодиться в качестве предпочтительного решения. Тем не менее, безусловно, можно реализовать что-то подобное с простым TPL. Обратите внимание, единственный блокирующий код здесь - это тот, который выполняет фактическую обработку данных внутри Task.Run:

const int MAX_DOWNLOADS = 50;
const int MAX_PROCESSORS = 4;

// process data
class Processing
{
    SemaphoreSlim _semaphore = new SemaphoreSlim(MAX_PROCESSORS);
    HashSet<Task> _pending = new HashSet<Task>();
    object _lock = new Object();

    async Task ProcessAsync(string data)
    {
        await _semaphore.WaitAsync();
        try
        {
            await Task.Run(() =>
            {
                // simuate work
                Thread.Sleep(1000);
                Console.WriteLine(data);
            });
        }
        finally
        {
            _semaphore.Release();
        }
    }

    public async void QueueItemAsync(string data)
    {
        var task = ProcessAsync(data);
        lock (_lock)
            _pending.Add(task);
        try
        {
            await task;
        }
        catch
        {
            if (!task.IsCanceled && !task.IsFaulted)
                throw; // not the task's exception, rethrow
            // don't remove faulted/cancelled tasks from the list
            return;
        }
        // remove successfully completed tasks from the list 
        lock (_lock)
            _pending.Remove(task);
    }

    public async Task WaitForCompleteAsync()
    {
        Task[] tasks;
        lock (_lock)
            tasks = _pending.ToArray();
        await Task.WhenAll(tasks);
    }
}

// download data
static async Task DownloadAsync(string[] urls)
{
    var processing = new Processing();

    using (var semaphore = new SemaphoreSlim(MAX_DOWNLOADS))
    using (var httpClient = new HttpClient())
    {
        var tasks = urls.Select(async (url) =>
        {
            await semaphore.WaitAsync();
            try
            {
                var data = await httpClient.GetStringAsync(url);
                // put the result on the processing pipeline
                processing.QueueItemAsync(data);
            }
            finally
            {
                semaphore.Release();
            }
        });

        await Task.WhenAll(tasks.ToArray());
        await processing.WaitForCompleteAsync();
    }
}

Как я и просил, вот код, который я выбрал.

Работа настроена в конфигурации «мастер-деталь», и каждый мастер обрабатывается как пакет. Каждая единица работы выстраивается в очередь следующим образом:

var success = true;

// Start processing all the master records.
Master master;
while (null != (master = await StoredProcedures.ClaimRecordsAsync(...)))
{
    await masterBuffer.SendAsync(master);
}

// Finished sending master records
masterBuffer.Complete();

// Now, wait for all the batches to complete.
await batchAction.Completion;

return success;

Мастера помещаются в буфер по одному, чтобы сохранить работу для других внешних процессов. Детали для каждого мастера отправляются на работу через masterTransform TransformManyBlock. BatchedJoinBlock также создается для сбора деталей в одной партии.

Фактическая работа выполняется в detailTransform TransformBlock асинхронно, по 150 за раз. BoundedCapacity установлен на 300, чтобы гарантировать, что слишком много мастеров не будут помещены в буфер в начале цепочки, а также оставляют место для постановки в очередь достаточного количества подробных записей, позволяющих обрабатывать 150 записей за один раз. Блок выводит object для своих целей, потому что он фильтруется по ссылкам в зависимости от того, Detail или Exception.

batchAction ActionBlock собирает выходные данные из всех пакетов и выполняет массовые обновления базы данных, ведение журнала ошибок и т. Д. Для каждого пакета.

Будет несколько BatchedJoinBlock, по одному на каждого мастера. Поскольку каждый ISourceBlock выводится последовательно и каждый пакет принимает только то количество подробных записей, которое связано с одним мастером, пакеты будут обрабатываться по порядку. Каждый блок выводит только одну группу и по завершении отключается. Только последний пакетный блок передает свое завершение на последний ActionBlock.

Сеть потока данных:

// The dataflow network
BufferBlock<Master> masterBuffer = null;
TransformManyBlock<Master, Detail> masterTransform = null;
TransformBlock<Detail, object> detailTransform = null;
ActionBlock<Tuple<IList<object>, IList<object>>> batchAction = null;

// Buffer master records to enable efficient throttling.
masterBuffer = new BufferBlock<Master>(new DataflowBlockOptions { BoundedCapacity = 1 });

// Sequentially transform master records into a stream of detail records.
masterTransform = new TransformManyBlock<Master, Detail>(async masterRecord =>
{
    var records = await StoredProcedures.GetObjectsAsync(masterRecord);

    // Filter the master records based on some criteria here
    var filteredRecords = records;

    // Only propagate completion to the last batch
    var propagateCompletion = masterBuffer.Completion.IsCompleted && masterTransform.InputCount == 0;

    // Create a batch join block to encapsulate the results of the master record.
    var batchjoinblock = new BatchedJoinBlock<object, object>(records.Count(), new GroupingDataflowBlockOptions { MaxNumberOfGroups = 1 });

    // Add the batch block to the detail transform pipeline's link queue, and link the batch block to the the batch action block.
    var detailLink1 = detailTransform.LinkTo(batchjoinblock.Target1, detailResult => detailResult is Detail);
    var detailLink2 = detailTransform.LinkTo(batchjoinblock.Target2, detailResult => detailResult is Exception);
    var batchLink = batchjoinblock.LinkTo(batchAction, new DataflowLinkOptions { PropagateCompletion = propagateCompletion });

    // Unlink batchjoinblock upon completion.
    // (the returned task does not need to be awaited, despite the warning.)
    batchjoinblock.Completion.ContinueWith(task =>
    {
        detailLink1.Dispose();
        detailLink2.Dispose();
        batchLink.Dispose();
    });

    return filteredRecords;
}, new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });

// Process each detail record asynchronously, 150 at a time.
detailTransform = new TransformBlock<Detail, object>(async detail => {
    try
    {
        // Perform the action for each detail here asynchronously
        await DoSomethingAsync();

        return detail;
    }
    catch (Exception e)
    {
        success = false;
        return e;
    }

}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 150, BoundedCapacity = 300 });

// Perform the proper action for each batch
batchAction = new ActionBlock<Tuple<IList<object>, IList<object>>>(async batch =>
{
    var details = batch.Item1.Cast<Detail>();
    var errors = batch.Item2.Cast<Exception>();

    // Do something with the batch here
}, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

masterBuffer.LinkTo(masterTransform, new DataflowLinkOptions { PropagateCompletion = true });
masterTransform.LinkTo(detailTransform, new DataflowLinkOptions { PropagateCompletion = true });