如何将任务参数包含在任务异常处理数组中 [英] How to include task parameter In array of tasks exception handling

问题描述

线程在Parallel.ForEach中嵌套等待有一个答案建议使用Task.WhenAll并行运行多个(MaxDegreeOfParallelism)异步任务，而不要等到上一个任务完成.

public static Task ForEachAsync<T>(
      this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll( 
        from partition in Partitioner.Create(source).GetPartitions(dop) 
        select Task.Run(async delegate { 
            using (partition) 
                while (partition.MoveNext()) 
                    await body(partition.Current).ContinueWith(t => 
                          {
                              //observe exceptions
                          });
})); 
}

并像

那样称呼它

ids.ForEachAsync(10,  async id =>
{
    ICustomerRepo repo = new CustomerRepo();  
    var cust = await repo.GetCustomer(id);  
    customers.Add(cust);  
});

如果body有一个参数，我想在处理异常时知道参数值，例如如果任务主体的id失败，那么我需要记录异常，并指定它是针对特定id发生的.

我看过 访问Task.ContinueWith 中的值，但在t时无法访问参数. IsFaulted.

最后，我在lambda主体中添加了try/catch，它似乎可以工作

ids.ForEachAsync(10,  async id =>
{
    try
    {
        ICustomerRepo repo = new CustomerRepo();
        var cust = await repo.GetCustomer(id);
        customers.Add(cust);
    }
    catch(Exception e)
    {
        _logger.LogError(e, id=+ id);
    }
});

但是我不确定它是否可以正常工作(即异步运行，没有阻塞).

后来原始答案的作者建议使用var当前=分区.当前在等待正文之前，然后在继续中使用current(ContinueWith(t => {...}).

任何人都可以确认哪种方法更好吗?每种方法有什么缺点?

解决方案

在try/catch中包装await很好，请参见:async/await之前).

请注意，您的示例将异常处理的负担转移给了调用方(在本例中为_logger.LogError).您需要确保这就是您想要的，而不是嵌入在ForEachAsync代码本身中的包罗万象"，以处理调用者确实允许异常通过的情况.像这样:

while (partition.MoveNext()) 
{
    try
    {
        await body(partition.Current)
    }
    catch (Exception e)
    {
        // of course here you don't know the type of T (partition.Current)
        // or anything else about the operation for that matter
        LogError("error processing: " + partition.Current + ": " + e); 
    }
}

The thread Nesting await in Parallel.ForEach has an answer suggested to use Task.WhenAll to run multiple (MaxDegreeOfParallelism) asynchronous tasks in parallel, not waiting until previous task is completed.

public static Task ForEachAsync<T>(
      this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll( 
        from partition in Partitioner.Create(source).GetPartitions(dop) 
        select Task.Run(async delegate { 
            using (partition) 
                while (partition.MoveNext()) 
                    await body(partition.Current).ContinueWith(t => 
                          {
                              //observe exceptions
                          });
})); 
}

And call it like

ids.ForEachAsync(10,  async id =>
{
    ICustomerRepo repo = new CustomerRepo();  
    var cust = await repo.GetCustomer(id);  
    customers.Add(cust);  
});

If body has a parameter, I want to know parameter value when handling exceptions.e.g. If task body failed for id , I need to log exception, specifying, that it happened for particular id.

I’ve looked at Accessing values in Task.ContinueWith, but wasn’t able to access parameters when t.IsFaulted.

Finally I’ve added try/catch inside lambda body and it seems to work

ids.ForEachAsync(10,  async id =>
{
    try
    {
        ICustomerRepo repo = new CustomerRepo();
        var cust = await repo.GetCustomer(id);
        customers.Add(cust);
    }
    catch(Exception e)
    {
        _logger.LogError(e," id="+ id);
    }
});

However I am not sure, does it work correctly(i.e asynchronously, without blocking).

Later the author of the original answer suggested to use var current = partition.Current before await body and then use current in the continuation (ContinueWith(t => { ... }). –

Can anyone confirm, which approach is better? Any disadvantages of each of approaches?

解决方案

Wrapping await in try/catch is fine, see: Catch an exception thrown by an async method. No big difference from my suggestion (capturing partition.Current and injecting into the ContinueWith continuation), except maybe it's a bit more efficient since no capturing is involved. Also it's a bit more readable and elegant I think, ContinueWith is kind of the "old" way of doing things (pre async/await).

Note that your example passes the burden of exception handling to the caller (who in this case calls _logger.LogError). You need to make sure that's what you want, as opposed to a catch-all embedded in the ForEachAsync code itself to handle the case where he caller does let an exception slip through. Something like:

while (partition.MoveNext()) 
{
    try
    {
        await body(partition.Current)
    }
    catch (Exception e)
    {
        // of course here you don't know the type of T (partition.Current)
        // or anything else about the operation for that matter
        LogError("error processing: " + partition.Current + ": " + e); 
    }
}

这篇关于如何将任务参数包含在任务异常处理数组中的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！