c# - 等待列表中的所有任务，然后根据任务结果更新列表

Question

我想用 WhenAll 运行所有任务(不是一个接一个)。但之后我需要根据结果更新列表(LastReport 属性)。

我想我有解决方案，但我想看看是否有更好的方法。

想法是:

• 运行所有任务
• 记住配置和任务之间的关系
• 更新配置

我的解决方案是:

var lastReportAllTasks = new List<Task<Dictionary<string, string>>>();
var configurationTaskRelation = new Dictionary<int, Task<Dictionary<string, string>>>();
foreach (var configuration in MachineConfigurations)
{
    var task = machineService.GetReports(configuration);
    lastReportAllTasks.Add(task);
    configurationTaskRelation.Add(configuration.Id, task);
}

await Task.WhenAll(lastReportAllTasks);
foreach (var configuration in MachineConfigurations)
{
    var lastReportTask = configurationTaskRelation[configuration.Id];
    configuration.LastReport = await lastReportTask;
}

Answer 1

Select函数本身可以是异步的。您可以等待报告并返回配置和结果相同的结果对象(匿名类型或元组，无论您喜欢什么):

var tasks=MachineConfigurations.Select(async conf=>{
                       var report= await machineService.GetReports(conf);
                       return new {conf,report});
var results=await Task.WhenAll(tasks);
foreach(var pair in results)
{
    pair.conf.LastReport=pair.report;
}

编辑 - 循环和错误处理

按照 Servy 的建议，Task.WhenAll可以省略并且等待可以在循环内移动:

foreach(var task in tasks)
{   
    var pair=await task;
    pair.conf.LastReport=pair.report;
}

任务仍将并发执行。但在异常情况下，一些配置对象将被修改，而另一些则不会。

一般来说，这将是一个丑陋的情况，需要额外的异常处理代码来清理修改的对象。当修改就地完成并在快乐路径完成时完成/应用时，异常处理会容易得多。这就是更新 Select() 中的配置对象的原因之一。需要仔细考虑。

在这种特殊情况下虽然“跳过”失败的报告可能更好，但可能会将它们移至错误队列并在以后重新处理它们。有部分结果可能比没有结果要好，只要这种行为是预期的:

foreach(var task in tasks)
{   
    try
    {
        var pair=await task;
        pair.conf.LastReport=pair.report;
    }
    catch(Exception exc)
    {
        //Make sure the error is logged
        Log.Error(exc);
        ErrorQueue.Enqueue(new ProcessingError(conf,ex);
    }
}
//Handle errors after the loop

编辑 2 - 数据流

为了完整起见，我确实每天要生成几千份票务报告，而且每次 GDS 调用(每个旅行社通过该服务售票)都需要相当长的时间。我无法同时运行所有 请求 - 如果我尝试超过 10 个并发请求，我将开始收到服务器序列化错误。我也不能重试一切。

在这种情况下，我结合使用了 TPL DataFlow 和一些 Railway oriented programming tricks . DOP 为 8 的 ActionBlock 处理票证请求。结果包装在 Success 中类并发送到下一个 block 。失败的请求和异常包含在 Failure 中类并发送到另一个 block 。这两个类都继承自具有 Successful 的 IFlowEnvelope旗帜。是的，这就是 F# Discriminated Union 嫉妒。

这与超时等的一些重试逻辑相结合。

在伪代码中，管道看起来像这样:

var reportingBlock=new TransformBlock<Ticket,IFlowEnvelope<TicketReport>(reportFunc,dopOptions);
var happyBlock = new ActionBlock<IFlowEnvelope<TicketReport>>(storeToDb);
var errorBlock = new ActionBlock<IFlowEnvelope<TicketReport>>(logError);

reportingBlock.LinkTo(happyBlock,linkOptions,msg=>msg.Success);
reportingBlock.LinkTo(errorBlock,linkOptions,msg=>!msg.Success);

foreach(var ticket in tickets)
{
    reportingBlock.Post(ticket);
}

reportFunc捕获任何异常并将它们包装为 Failure<T>对象:

async Task<IFlowEnvelope<Ticket,TicketReport>> reportFunc(Ticket ticket)
{
    try
    {
        //Do the heavy processing
        return new Success<TicketReport>(report);
    }
    catch(Exception exc)
    {
        //Construct an error message, msg
        return new Failure<TicketReport>(report,msg);
    }
}

真正的流水线包括解析每日报告和个人工单的步骤。每次调用 GDS 需要 1-6 秒，因此管道的复杂性是合理的。