我可以在后台运行多个慢进程,因此可以并行运行多个任务吗？

Question

我有一个使用C#Core .NET 2.2框架顶部编写的控制台应用程序.

我的应用程序允许我使用Windows任务调度程序触发长时间运行的管理作业.

其中一个管理作业进行Web-API调用,在将文件上载到Azure Blob存储之前下载大量文件.以下是我的代码完成工作所需执行的逻辑步骤

1. 使用Mime消息调用远程API,其中每条消息代表一个文件.
2. 解析Mime消息并将每条消息转换为MemoryStream创建MemoryStream的集合

一旦我拥有多个1000+的集合MemoryStream,我想将每个集合写入StreamAzure Blob存储.由于对远程存储的写入速度很慢,我希望我可以使用自己的进程或线程执行每次写入迭代.这将允许我同时并行运行1000+线程,而不必等待每次写入操作的结果.每个线程都将负责记录在写入/上载过程中可能发生的任何错误.任何记录的错误都将使用不同的作业处理,因此我不必担心重试.

我的理解是调用异步写入/上传流的代码就是这样做的.换句话说,我会说" Stream执行它并运行它只需要它.只要任务完成,我就不关心结果."

在测试时,我发现我对调用的理解async有些无效.我的印象是,当调用一个定义的方法时,async将在后台线程/ worker中执行,直到该过程完成.但是,当我测试代码时,我的理解失败了.我的代码告诉我,如果不添加关键字await,async代码就永远不会被执行.同时,当await添加关键字时,代码将等待,直到该过程在继续之前完成执行.换句话说,await为我的需要添加将失去异步调用方法的目的.

这是我的代码的精简版本,以便解释我想要完成的任务

public async Task Run()
{
    // This gets populated after calling the web-API and parsing out the result
    List<Stream> files = new List<MemoryStream>{.....};

    foreach (Stream file in files)
    {
        // This code should get executed in the background without having to await the result
        await Upload(file);
    }
}

// This method is responsible of upload a stream to a storage and log error if any
private async Task Upload(Stream stream)
{
    try
    {
        await Storage.Create(file, GetUniqueName());
    } 
    catch(Exception e)
    {
        // Log any errors
    }
}

从上面的代码中,调用await Upload(file);工作并将按预期上传文件.但是,由于我await在调用Upload()方法时使用,因此在上传代码完成之前,我的循环不会跳转到下一次迭代.同时,删除await关键字后,循环不等待上传过程,但Stream实际上从未实际写入存储,就好像我从未调用过代码一样.

如何Upload并行执行多个方法,以便在后台每次上传一个线程？

Answer 1

将列表转换为"上传"任务列表,并等待它们全部Task.WhenAll():

public async Task Run()
{
    // This gets populated after calling the web-API and parsing out the result
    List<Stream> files = new List<MemoryStream>{.....};
    var tasks = files.Select(Upload);

    await Task.WhenAll(tasks);
}

有关tasks/await的更多信息,请参阅此文章.