将 Task 异常与 Task<T> 响应相关联

Question

我刚刚开始处理任务。我们有一个使用请求/响应的系统设置。运行任务的服务接受具有请求对象列表的主请求并返回具有响应对象列表的主响应。所以它看起来像这样

    var MasterRequest = new MasterRequest;
    MasterRequest.Requests.Add(new BlueRequest);
    MasterRequest.Requests.Add(new RedRequest);
    MasterRequest.Requests.Add(new YellowRequest);

请求实现了一个简单的 IRequest 接口，每种颜色都是一个具体的类。该服务具有设置为能够根据具体请求对象分别和同时处理每个请求的具体类（请求处理器）。服务上的每个具体类都有一个 GetTask 方法，其签名如下：

    Task<IResponse> GetTask(IRequest);
    {
       // some setup stuff
       return Task.Factory.StartNew<IResponse>(() =>
        {
           // do task stuff
           return response; // implements IResponse
         });            
    }

我的服务接受传入的 MasterRequest 并通过在具体请求处理器上调用上面列出的 GetTask 调用来构建任务列表。然后我使用列表中的 Parallel.ForEach 来处理任务。

    // this is what is returned from the service.
    // it has a List<IResponse> on it to hold  the resposnes
    MasterResposne resposne = new MasterResponse();

    List<Task<IResponse>> tasks = new List<Task<IResponse>>();

    foreach(IRequest req in MasterRequest.Requests)
    {
        // factory to get the proper request processor
        RequestProcessor p  = rp.GetProcessor(req);

        tasks.add(p.GetTask(req));
     }

     Parallel.ForEach(tasks, t =>
        {
             t.Wait();

              // check for faulted and cancelled 
              // this is where I need help

              response.Responses.Add(t.Result);
         }

这一切都很好。但是，如果任务引发异常，我不知道如何将其与触发它的特定具体请求联系起来。我需要知道，这样我才能将正确构建的响应传回给调用者。

我的第一个想法是对 Task 进行子类化，但这带来了我不想处理的一系列问题。

我读了这篇 SO 文章，似乎我想做这样的事情

Is this ok to derive from TPL Task to return more details from method?

我认为 Reed 的第二个示例是我的解决方案，但我仍然看不到如何同时运行任务并能够将异常与请求联系起来，以便我可以返回正确构建的响应列表。

提前致谢。

Answer 1

所以我能够从我提供的链接中使用 Reed 的解决方案。我处理请求的服务代码变成了这个

// this is what is returned from the service.
// it has a List<IResponse> on it to hold  the resposnes
MasterResposne resposne = new MasterResponse();

List<ExecutionResult> tasks = new List<ExecutionResult>();

foreach(IRequest req in MasterRequest.Requests)
{
    // factory to get the proper request processor
    RequestProcessor p  = rp.GetProcessor(req);

    tasks.add(p.GetResult(req));
 }

 Parallel.ForEach(tasks, t =>
    {
         t.task.Wait();

          response.Responses.Add(t.Result);
     }

ExecutionResult 是这样定义的

    class ExecutionResult
    {
       public IResult Result;
       public Task<IResponse> task;

    }

这使我可以访问预构建的响应对象，以便我可以将其传回给调用者。

编辑： 因此，我查看了我的 Parallel.ForEach 并能够重做我的代码并按照建议使用 await Task.WhenAll。新代码看起来更像这样：

    // this is what is returned from the service.
    // it has a List<IResponse> on it to hold  the resposnes
    MasterResposne resposne = new MasterResponse();

    List<ExecutionResult> tasks = new List<ExecutionResult>();
    List<ExecutionResult> executionResults = new List<ExecutionResult>();

    foreach(IRequest req in MasterRequest.Requests)
    {
        // factory to get the proper request processor
        RequestProcessor p  = rp.GetProcessor(req);

        ExecutionResult er = engine.GetResult(req);
        executionResults.Add(er);
        tasks.Add(er.Task);
     }

      await Task.WhenAll<IResponse>(tasks);

      foreach (ExecutionResult r in executionResults)
      {
          if (r.Task.IsCompleted)
          {
              response.AddResponse(r.Task.Result);
          }
          else
          {
              r.Response.Status = false;
              AggregateException flat = r.Task.Exception.Flatten();

              foreach (Exception e in flat.InnerExceptions)
              {
                  Log.ErrorFormat("Reqest [{0}] threw [{1}]", r.Response.RequestId, e);
                  r.Response.StatusReason.AppendLine(e.Message);
              }
          }
      }

这使我可以将我的请求信息与我的任务联系起来，并获得我需要返回给我的调用者的响应。

感谢您的指导。

Answer 2

然后我使用列表中的 Parallel.ForEach 来处理任务。

这实际上很糟糕。它把 ton 的线程混在一起，只是为了阻止任务的完成。

但如果任务引发异常，我不知道如何将其与触发它的具体具体请求联系起来。我需要知道，这样我才能将正确构建的响应传回给调用者。

每当您遇到“完成后处理任务”这类问题时，通常最好的解决方案是更高级别的异步操作：

private async Task<IResponse> ProcessAsync(IRequest request)
{
  try
  {
    return await engine.GetResult(request);
  }
  catch (Exception ex)
  {
    IResponse result = /* create error response */;
    return result;
  }
}

这允许一个更简单的主函数：

MasterResposne resposne = new MasterResponse();

var tasks = MasterRequest.Requests.Select(req => ProcessAsync(req));
response.AddRange(await Task.WhenAll(tasks));