创建一个将多个 I/O 任务分组的任务

Question

首先，我不确定我是否在问一些愚蠢的问题，所以在开始之前道歉。

我有一个方法使用 SqlCommand 异步保存数据库中的实体集合，它返回一个代表异步操作的任务，到目前为止它运行良好。基本上它会为每个实体生成一个 SQL 命令和一堆参数，并将它们全部添加到一个 SqlCommand 实例（参数已编号）

但是，如果我尝试插入很多实体，当参数计数达到 2100 时，SQL 操作会失败，因为那是限制。然后我想分批拆分实体集合，并按顺序执行它们，然后返回一个直到所有子任务都完成后才会完成的任务。 （可能是最后一个）

每个子任务都会得到一个 Int32，说明有多少行发生了变化，最终任务必须返回所有行的总和。所以所有的任务都是Task。如果一个失败，它们都必须“回滚”，因此它们必须共享连接和事务对象。

另外，我想确保我使用 Task 和 SqlCommand 正确使用 I/O 完成端口，并且在 SQL Server 上的操作完成时没有线程等待/旋转，因为这个“保存" 操作是从 ASP.NET MVC 中的异步控制器调用的。

这里的正确方法是什么？

问候。

Answer 1

使用 C# 5（并假设基于任务的异步），这将非常简单：

async Task<IReadOnlyList<SomeType>> PerformUpdatesAsync(
    IEnumerable<AnotherType> data)
{
    var result = new List<SomeType>();

    foreach (var batch in data.SplitIntoBatches(BatchSize))
        result.Add(await PerformUpdateAsync(batch));

    return result;
}

在 C# 5 之前的版本中模拟异步 foreach 更加困难，但您可以使用批处理队列和“递归” lambda 进行处理，例如：

Task<SomeType[]> PerformUpdatesAsync(IEnumerable<AnotherType> data)
{
    var batches = new Queue<IEnumerable<AnotherType>>(
        data.SplitIntoBatches(BatchSize));

    var result = new List<SomeType>();

    var tcs = new TaskCompletionSource<SomeType[]>();

    AsyncCallback onEndUpdate = null;
    onEndUpdate =
        ar =>
        {
            result.Add(EndUpdate(ar));

            if (batches.Count == 0)
                tcs.SetResult(result.ToArray());
            else
                BeginUpdate(batches.Dequeue(), onEndUpdate, null);
        };

    BeginUpdate(batches.Dequeue(), onEndUpdate, null);

    return tcs.Task;
}

如果您不喜欢以这种方式使用闭包，您可以通过创建一个单独的类来做同样的事情，该类将在其字段中保存所有局部变量。

框架中没有SplitIntoBatches()，因此您必须自己编写（除非您已经这样做了）。

Answer 2

如果您要拆分插入，但按顺序运行它们，则无需为每个调用使用任务。你只需要正常的顺序代码。

创建一个启动任务并返回它的方法。在任务主体中，只需执行您想要的任何正常顺序代码。完成后，退出任务主体，单个任务将完成。

Answer 3

这是个好问题。这是一些代码：

Task<int>[] batchTasks = ...;//start the batches here. each task should return the count of changed rows
Task<int> sumTask = Task.Factory.ContinueWhenAll(batchTasks, _ => {
 return batchTasks.Sum(x => x.Result);
});

这将为您提供各个批次的无阻塞总和。

当您开始批处理时，请务必使用 BeginExecuteReader 等异步 SQL API。然后，您将完全无阻塞。