我如何保证继续按任务完成顺序运行？

Question

如果我的代码通过返回代表每个阶段的任务来抽象异步操作的分阶段序列，我如何确保继续按阶段顺序（即任务完成的顺序）执行？

请注意，这与简单的“不要浪费时间等待较慢的任务”不同。在调度中需要保证没有竞争条件的顺序。这个较宽松的要求可以通过以下问题的部分答案来解决：

1. Sort Tasks into order of completition
2. Is there default way to get first task that finished successfully?

我认为合乎逻辑的解决方案是使用自定义 TaskScheduler（例如基于 SynchronizationContext 的）附加延续。但是，我无法保证调度在任务完成时同步执行。

在代码中，这可能类似于

class StagedOperationSource
{
    public TaskCompletionSource Connect = new TaskCompletionSource();
    public TaskCompletionSource Accept = new TaskCompletionSource();
    public TaskCompletionSource Complete = new TaskCompletionSource();
}
class StagedOperation
{
    public Task Connect, Accept, Complete;
    public StagedOperation(StagedOperationSource source)
    {
        Connect = source.Connect.Task;
        Accept = source.Accept.Task;
        Complete = source.Complete.Task;
    }
}
...
private StagedOperation InitiateStagedOperation(int opId)
{
    var source = new StagedOperationSource();
    Task.Run(GetRunnerFromOpId(opId, source));
    return new StagedOperation(source);
}
...
public RunOperations()
{
    for (int i=0; i<3; i++)
    {
        var op = InitiateStagedOperation(i);
        op.Connect.ContinueWith(t => Console.WriteLine("{0}: Connected", i));
        op.Accept.ContinueWith(t => Console.WriteLine("{0}: Accepted", i));
        op.Complete.ContinueWith(t => Console.WriteLine("{0}: Completed", i));
    }
}

应该产生类似于

的输出

0: Connected
1: Connected
0: Accepted
2: Connected
0: Completed
1: Accepted
2: Accepted
2: Completed
1: Completed

显然，该示例缺少详细信息，例如如果早期阶段失败，则将异常转发到（或取消）后期阶段，但这只是一个示例。

Answer 1

在进入下一个阶段之前，只需await...

public static async Task ProcessStagedOperation(StagedOperation operation, int i)
{
    await operation.Connect;
    Console.WriteLine("{0}: Connected", i);
    await operation.Accept;
    Console.WriteLine("{0}: Accepted", i);
    await operation.Complete;
    Console.WriteLine("{0}: Completed", i);
}

然后您可以在 for 循环中调用该方法。

Answer 2

如果您使用 TAP（任务异步编程），即 async 和 await，您可以使处理流程更加明显。在这种情况下，我将创建一个新方法来封装操作顺序：

public async Task ProcessStagedOperation(StagedOperation op, int i)
{
    await op.Connect;
    Console.WriteLine("{0}: Connected", i);

    await op.Accept;
    Console.WriteLine("{0}: Accepted", i)

    await op.Complete;
    Console.WriteLine("{0}: Completed", i)
}

现在你的处理循环被简化了一点：

public async Task RunOperations()
{
    List<Task> pendingOperations = new List<Task>();

    for (int i=0; i<3; i++)
    {
        var op = InitiateStagedOperation(i);
        pendingOperations.Add(ProcessStagedOperation(op, i));
    }

    await Task.WhenAll(pendingOperations); // finish
}

您现在有一个对任务对象的引用，您可以显式等待或简单地从另一个上下文中 await。 （或者你可以简单地忽略它）。我修改 RunOperations() 方法的方式允许您创建一个大型待处理任务队列，但在等待它们全部完成时不会阻塞。