c# - 我应该如何从 .NET 4.0 中的延续任务传播任务异常？

Question

我有以下代码:

innerExceptions = dbconnByServer
    .AsParallel()
    .WithDegreeOfParallelism(dbconnByServer.Count)
    // A stream of groups of server connections proceeding in parallel per server
    .Select(dbconns => dbconns.Select(dbconn => m_sqlUtilProvider.Get(dbconn)))
    // A stream of groups of SqlUtil objects proceeding in parallel per server
    .Select(sqlUtils => GetTaskException(sqlUtils
        // Aggregate SqlUtil objects to form a single Task which runs the SQL asynchronously for the first SqlUtil, then upon completion
        // for the next SqlUtil and so long until all the SqlUtil objects are processed asynchronously one after another.
        .Aggregate<ISqlUtil, Task>(null, (res, sqlUtil) =>
        {
            if (res == null)
            {
                return sqlUtil.ExecuteSqlAsync(SQL, parameters);
            }
            return res.ContinueWith(_ => sqlUtil.ExecuteSqlAsync(SQL, parameters)).Unwrap();
        })))
    .Where(e => e != null)
    .ToList();

地点:

private static Exception GetTaskException(Task t)
{
    try
    {
        t.Wait();
        return null;
    }
    catch (Exception exc)
    {
        return exc;
    }
}

此代码的作用是跨多个数据库连接执行某些 SQL 语句，其中一些连接可能属于一个数据库服务器，而其他连接则属于另一个数据库服务器，依此类推。

代码确保满足两个条件:

1. SQL 语句在可用的数据库服务器上并行运行。
2. 在同一个数据库服务器中，SQL 语句异步但顺序运行。

给定每个数据库服务器的 N 个数据库连接，在聚合结束时将有一个 Task，执行它具有以下效果:

• 为 db conn 1 执行 SQL
  • 完成前面的操作后，为 db conn 2 执行 SQL
    • 完成前面的操作后，为 db conn 3 执行 SQL
    • ...
      • 完成前面的操作后，执行 db conn N 的 SQL

我的问题是现在除了第一个数据库连接外，异常都丢失了。我知道我应该检查 _ 参数并以某种方式处理延续函数内的 _.Exception 属性。我想知道是否有一种优雅的方式来做到这一点。

有什么想法吗？

Answer 1

1. 如果您打算使用异步方法，那么整个方法应该是异步的并且不应阻塞任何线程。
2. 如果您不打算阻塞，则没有太多理由使用 PLINQ，从单个线程设置延续应该足够快。
3. 如果你想在发生异常时继续，你将不得不自己将异常存储在某个地方。
4. 我对处理一组异常而不抛出它们感到不安，但我想这对于一个操作来说是可以接受的，它可以部分失败，部分成功。

这样，我执行此操作的代码将如下所示:

public Task<IEnumerable<Exception>> ExecuteOnServersAsync(
    IList<IEnumerable<Connection>> dbConnByServer,
    string sql, object parameters)
{
    var tasks = new List<Task>();
    var exceptions = new ConcurrentQueue<Exception>();

    Action<Task> handleException = t =>
    {
        if (t.IsFaulted)
            exceptions.Enqueue(t.Exception);
    };

    foreach (var dbConns in dbConnByServer)
    {
        Task task = null;

        foreach (var dbConn in dbConns)
        {
            var sqlUtil = m_sqlUtilProvider.Get(dbConn);

            if (task == null)
            {
                task = sqlUtil.ExecuteSqlAsync(sql, parameters);
            }
            else
            {
                task = task.ContinueWith(
                    t =>
                    {
                        handleException(t);
                        return sqlUtil.ExecuteSqlAsync(sql, parameters);
                    }).Unwrap();
            }
        }

        if (task != null)
        {
            task = task.ContinueWith(handleException);
            tasks.Add(task);
        }
    }

    return Task.Factory.ContinueWhenAll(
        tasks.ToArray(), _ => exceptions.AsEnumerable());
}