异步/等待和多处理

Question

我有包含 50 000 000 个唯一域名的 txt 文件，起初我试图只打开每个站点。我正在使用异步 HttpClient 并尝试了 3 种不同的方法来拆分任务：

1

IEnumerable<string> lines = File.ReadLines("file.txt");
try
{
    DataSet allData;
    var downloadData = new TransformBlock<string,byte[]>(
    async line =>
    {
        HttpClientHandler httpClientHandler = new HttpClientHandler();
        HttpClient client = new HttpClient(httpClientHandler);
        try
        {
            HttpResponseMessage responseMessage =
            await client.GetAsync(line).ConfigureAwait(false);
            return
            await responseMessage.Content.ReadAsByteArrayAsync().ConfigureAwait(false);
        }
        catch (Exception ex)
        {
            //catch all to reduce code for testing
            return null;
        }
        finally
        {
            Interlocked.Increment(ref finishedUrls);
        }
    },
    new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 500,
    });
    foreach (var line in lines)
    downloadData.Post(line);
    downloadData.Complete();
    await downloadData.Completion;

2

List<Task> allTasks = new List<Task>();
SemaphoreSlim throttler = new SemaphoreSlim(initialCount: DataflowBlockOptions.Unbounded);
foreach (var line in lines)
{

    await throttler.WaitAsync().ConfigureAwait(false);
    allTasks.Add(Task.Run(async () =>
    {

            try
            {
                HttpClientHandler httpClientHandler = new HttpClientHandler();
                HttpClient client = new HttpClient(httpClientHandler);
                try
                {
                    HttpResponseMessage responseMessage = await client.GetAsync(line).ConfigureAwait(false);
                    var byteArray = await responseMessage.Content.ReadAsByteArrayAsync().ConfigureAwait(false);
                }
                catch (Exception ex)
                {
                }
                Interlocked.Increment(ref finishedUrls);
            }
            catch (Exception ex)
            {
            }
        }
        finally
        {
            throttler.Release();
        }
    }));
}
await Task.WhenAll(allTasks);

3

await lines.ForEachAsync(500,cancellationToken,async line =>
{
    HttpClientHandler httpClientHandler = new HttpClientHandler();
    HttpClient client = new HttpClient(httpClientHandler);
    try
    {
        HttpResponseMessage responseMessage = await client.GetAsync(line).ConfigureAwait(false);
        var byteArray = await responseMessage.Content.ReadAsByteArrayAsync().ConfigureAwait(false);
    }
    catch (Exception ex)
    {
    }
    Interlocked.Increment(ref finishedUrls);
}
);

public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, CancellationToken cancellationToken,
            Func<T, Task> body)
        {
            return Task.WhenAll(
                from partition in Partitioner.Create(source).GetPartitions(dop).AsParallel()
                select Task.Run(async delegate
                {
                    using (partition)
                        while (partition.MoveNext())
                            await body(partition.Current).ConfigureAwait(false);
                }, cancellationToken));
        }

我从 #3 解决方案中获得的最佳速度结果 - 在性能监视器中建立了大约 12 000 个 url/分钟和 10 000 个连接 - cpu 使用了 ~1%

但是，当我将 txt 文件拆分为 5 个部分 = 每个文件中有 10 000 000 个 url 并运行我的程序的 5 个实例时，汇总速度为 25 000 个 url / 分钟并建立了 30 000 个连接 - cpu 使用率为 3%。我正在玩将分区数从 500 增加到更多，但这并没有带来很大的变化。所以我的问题是 - 如何运行一个可以处理 25 000 个 URL / 分钟的程序实例？如何划分异步作业以获得尽可能高的速度？

对进程有任何 .NET 限制吗？

程序在64位windows server 2012上运行 500Mb 网络、64GB RAM、SSD 磁盘、E5-1620-v2 CPU

更新 1 不同“dop”和 4 个实例的速度结果同时： http://pastebin.com/ab3UQPAC

Answer 1

删除外部任务可能会有所帮助？

有点离题（减去异常处理？）

List<Task> allTasks = new List<Task>();
foreach (var line in lines)
{
            HttpClientHandler httpClientHandler = new HttpClientHandler();
            HttpClient client = new HttpClient(httpClientHandler);
            try
            {
            allTasks.Add(client.GetAsync(line).
            ContinueWith(t => t.Result.Content.ReadAsByteArrayAsync(), TaskContinuationOptions.OnlyOnRanToCompletion));
            }
            catch
            {
            }
}
await Task.WhenAll(allTasks);

如果有一个外部任务正在等待响应，是否可以想象您正在消耗过多的 ThreadPool 资源？不确定调度程序将如何处理此问题的具体细节，但外部任务对我来说似乎是多余的。

Answer 2

尝试将System.Net.ServicePointManager.DefaultConnectionLimit 设置为一个非常高的数字，例如int.MaxValue。

Answer 3

我认为您的问题与此有关：Limit of outgoing connections for one process (.Net) 尝试将最大连接数增加到您同时运行的任务数（可能是内核数）。