.Net Core 3.1 异步控制台应用程序在 VS 之外运行时挂起

Question

目标：对于 S3 URI 列表中的每个项目，获取对象数。

当从 VS 2019 运行时，我的 .Net Core 3.1 控制台应用程序运行良好，但一旦列表大小超过 5000 项左右，从 cmd（或任务计划程序、.bat 文件等）运行时就会出现问题。

在剩余的任务减少到大约 500-1000 个之前，一切似乎都还不错。然后，大约 75% 的时间，剩余的任务似乎永远不会完成，应用程序永远挂起......尽管 RAM 使用量在任务管理器中减少到几乎为零。

我对 Async 还很陌生，我已经尝试根据我看到的无数解决方案重构一堆，但似乎无法弄清楚。

注意事项：

• 在 VS 中，随着时间的推移，任务似乎恢复得更快，所以我的前 1000 任务可能需要 10 秒，下一个需要 9 秒，等等。似乎在 VS 之外 相反，它们会随着时间的推移而变慢
• 我在 AWS EC2 上运行这个应用程序，这是一个 t3a.2xlarge w/32GB RAM
• 当我使用 PowerShell 运行它时，有时在运行过程中，它会断开我与 RDP 的连接，有时会多次断开。
• 在 VS 中，应用程序使用大约 75MB 的空间和一个小 URI 列表，大约 600MB 和 150k 的列表。在 VS 之外，它使用大约 4 倍的 RAM。
• 我尝试编译为 32 位和 64 位

代码：

namespace MyNamespace
{
    public class MyClass
    {
        private static DataTable dt;
        private static IAmazonS3 clientS3;

        static async Task Main(string[] args)
        {
            dt = <Call DB, get S3 URIs>;
            clientS3 = new AmazonS3Client();

            IEnumerable<Task<int>> callApiTasksQuery = from row in dt.AsEnumerable() select GetS3DataAsync(row);
            List<Task<int>> apiTasks = callApiTasksQuery.ToList();

            int total = 0;
            while (apiTasks.Any())
            {
                // if (apiTasks.Count % 100 == 0) await Console.Out.WriteLineAsync($"{apiTasks.Count} remaining.");
                Task<int> finishedTask = await Task.WhenAny(apiTasks);
                apiTasks.Remove(finishedTask);
                total += await finishedTask;
            }
        }
        
        static async Task<int> GetS3DataAsync(DataRow row)
        {
            var response = await clientS3.ListObjectsV2Async(new ListObjectsV2Request { BucketName = row[0], Prefix = row[1] });
            // Console.WriteLine(response.S3Objects.Count().ToString());  
            return 1;
        }
    }
}

Answer 1

我看到的唯一问题是这段代码，它在 O(n^2) 时间内运行：

int total = 0;
while (apiTasks.Any())
{
  // if (apiTasks.Count % 100 == 0) await Console.Out.WriteLineAsync($"{apiTasks.Count} remaining.");
  Task<int> finishedTask = await Task.WhenAny(apiTasks);
  apiTasks.Remove(finishedTask);
  total += await finishedTask;
}

如果不需要输出，则将其替换为单个Task.WhenAll：

var totals = await Task.WhenAll(apiTasks);
var total = totals.Sum();

如果您确实需要输出，那么您可以通过完成一次然后await 每一个来重新排序。有some blogs on how to do that，也可以使用Nito.AsyncEx：

int total = 0;
var orderedApiTasks = apiTasks.OrderByCompletion();
for (int i = 0; i != orderedApiTasks.Count; ++i)
{
  total += await orderedApiTasks[i];
  if (i % 100 == 0) await Console.Out.WriteLineAsync($"{orderedApiTasks.Count - i} remaining.");
}

Answer 2

以下批处理解决方案有效。它会在 2-3 秒内恢复每个批次（如果在调试器中运行，则大约 10 秒）

感谢https://www.michalbialecki.com/2018/04/19/how-to-send-many-requests-in-parallel-in-asp-net-core/ 并感谢大家的协助！

using System;
using System.Threading.Tasks;
using System.Collections.Generic;
using System.Data;
using Amazon.S3;
using System.Linq;
using Amazon.S3.Model;

namespace MyNamespace
{
    public class S3PrefixGrabber
    {
        private static IAmazonS3 clientS3;

        static async Task Main(string[] args)
        {
            var query = "SELECT bucket,prefix from myTable";
            DataTable dt = GetStuffFromDB(query);
            List<S3Prefix> unpopulatedList = (from DataRow dr in dt.Rows select new S3Prefix() { B = dr[0].ToString(), P = dr[1].ToString() }).ToList();

            var batchSize = 1000;
            int numberOfBatches = (int)Math.Ceiling((double)unpopulatedList.Count() / batchSize);
            List<S3Prefix> populatedList = new List<S3Prefix>();

            for (int i = 0; i < numberOfBatches; i++)
            {
                var currentItems = unpopulatedList.Skip(i * batchSize).Take(batchSize);
                var tasks = currentItems.Select(id => GetS3DataAsync(id));
                populatedList.AddRange(await Task.WhenAll(tasks));
            }
        }

        static async Task<S3Prefix> GetS3DataAsync(S3Prefix s3Item)
        {
            clientS3 = new AmazonS3Client();
            var response = await clientS3.ListObjectsV2Async(new ListObjectsV2Request { BucketName = s3Item.B, Prefix = s3Item.P });
            s3Item.O = response.S3Objects.Count;

            return s3Item;
        }
    }

    public class S3Prefix
    {
        public string B { get; set; }
        public string P { get; set; }
        public int O { get; set; }
    }
}

运行 10k 条记录，RAM 为 75MB，CPU 为 40%
运行 300k 条记录，RAM 为 700MB，CPU 为 40%

日志中的 sn-p（我没有包含在上面的代码中）仅供参考：

06:32:52.310: ================= 开始 =================
06:32:52.795：查询：SELECT bucket,prefix FROM myTable
06:32:52.874：打开连接
06:32:54.205：灌装适配器
06:33:06.309: 从 DB 返回 313863 行
06:33:07.647：批处理...批处理大小：1000 批处理：314
06:33:07.647：开始批次 1/314...在 02.84 秒内完成。
06:33:10.492：开始批次 2/314... 在 02.48 秒内完成。
06:33:12.977：开始批次 3/314...在 02.19 秒内完成。
...
06:38:55.435：开始批次 150/314... 在 02.32 秒内完成。
06:38:57.761：开始批次 151/314... 在 02.17 秒内完成。
06:38:59.936：开始批次 152/314... 在 02.27 秒内完成。
...
06:45:13.579：开始批次 312/314... 在 02.17 秒内完成。
06:45:15.751：开始批次 313/314... 在 02.35 秒内完成。
06:45:18.105：开始批次 314/314... 02.10 秒完成。
06:45:20.211：将 313863 行写入 CSV... 完成。
06:45:23.086：数据库行：313863 CSV 行：313863 NotInS3：0 InS3ButNotFound：0
06:45:23.087：在 12:30.77 秒内完成。
06:45:23.092: ================= 结束 =================