如何使用 CancellationToken 链接异步任务？

Question

如果我的CancellationToken 没有被解雇，我想链接一些任务，但有条件地继续执行。

我想要达到的目标相当于

var cts = new CancellationTokenSource();
var cancellationToken = cts.Token;
var t = Task.Run(async () => {
    if (cancellationToken.IsCancellationRequested) return;
    await t1();
    if (cancellationToken.IsCancellationRequested) return;
    await t2();
    if (cancellationToken.IsCancellationRequested) return;
    await t3();
    if (cancellationToken.IsCancellationRequested) return;
    await t4();
});
var timeout = Task.Delay(TimeSpan.FromSeconds(4));
var completedTask = await Task.WhenAny(t, timeout);
if (completedTask != t)
{
    cts.Cancel();
    await t;
}

这就是我现在所拥有的，它正在工作，虽然它也很冗长。

Answer 1

var cts = new CancellationTokenSource();
var t = Task.Run(async () => {
     await t1();
     await t2();
     await t3();
     await t4();
 }, cts.Token);

cts.CancelAfter(TimeSpan.FromSeconds(4));

try
{
     await t;
}
catch (OperationCanceledException)
{
     // The cancellation token was triggered, add logic for that
     ....
}

Answer 2

您的原始代码是正确的——它假定您始终希望单个任务运行完成，并且如果取消，您希望整个任务成功完成。这些东西都不是惯用的。

更正常的做法是：

var cts = new CancellationTokenSource();
var cancellationToken = cts.Token;
var t = Task.Run(async () => {
    cancellationToken.ThrowIfCancellationRequested();
    await t1(cancellationToken);
    cancellationToken.ThrowIfCancellationRequested();
    await t2(cancellationToken);
    cancellationToken.ThrowIfCancellationRequested();
    await t3(cancellationToken);
    cancellationToken.ThrowIfCancellationRequested();
    await t4(cancellationToken);
}, cancellationToken);

然后，在其他地方：

cts.Cancel();

您可以在此处省略对ThrowIfCancellationRequested 的调用，假设各个任务在进入后很快就会对其进行检查，但核心思想是您应该将令牌传递到正在执行工作的最内层循环，并且他们应该通过调用它来引发取消，这最终将任务设置为取消状态而不是成功状态。

（所以实际上你只需要实际上调用ThrowIfCancellationRequested，如果你遇到一个不接受CancellationToken参数的函数——这就是所有异步方法都应该这样做的原因，因为否则它的任务将是不可取消的。）

Answer 3

您的目标似乎是在整个操作花费超过 4 秒时停止执行。

如果您将CancellationToken 传递给您的t1/t2/等。方法，我想说你不能做得比你所拥有的更好。但正如你所拥有的，你可以只使用Stopwatch 而不是CancellationToken：

var timeout = TimeSpan.FromSeconds(4);
var stopwatch = new Stopwatch();
stopwatch.Start();

await t1();
if (stopwatch.Elapsed > timeout) return;
await t2();
if (stopwatch.Elapsed > timeout) return;
await t3();
if (stopwatch.Elapsed > timeout) return;
await t4();
stopwatch.Stop();

我假设这是在某个方法中，您可以在其中使用return，但您可以根据需要进行修改（返回值、抛出异常等）。

Answer 4

您的代码在功能上是正常的，但是一目了然地阅读它并不清楚它在做什么。所以我建议你将这个逻辑封装在一个带有描述性名称和参数的实用方法中：

public static async Task RunSequentially(IEnumerable<Func<Task>> taskFactories,
    int timeout = Timeout.Infinite, bool onTimeoutAwaitIncompleteTask = false)
{
    using (var cts = new CancellationTokenSource(timeout))
    {
        if (onTimeoutAwaitIncompleteTask)
        {
            await Task.Run(async () =>
            {
                foreach (var taskFactory in taskFactories)
                {
                    if (cts.IsCancellationRequested) throw new TimeoutException();
                    await taskFactory();
                }
            });
        }
        else // On timeout return immediately
        {
            var allSequentially = Task.Run(async () =>
            {
                foreach (var taskFactory in taskFactories)
                {
                    cts.Token.ThrowIfCancellationRequested();
                    var task = taskFactory(); // Synchronous part of task
                    cts.Token.ThrowIfCancellationRequested();
                    await task; // Asynchronous part of task
                }
            }, cts.Token);
            var timeoutTask = new Task(() => {}, cts.Token);
            var completedTask = await Task.WhenAny(allSequentially, timeoutTask);
            if (completedTask.IsCanceled) throw new TimeoutException();
            await completedTask; // Propagate any exception
        }
    }
}

此代码与您的代码不同，因为它会在超时时抛出 TimeoutException。我认为最好强制调用者明确处理这个异常，而不是隐藏操作超时的事实。调用者可以通过将 catch 块留空来忽略异常：

try
{
    await RunSequentially(new[] { t1, t2, t3, t4 },
        timeout: 4000,
        onTimeoutAwaitIncompleteTask: true);
}
catch (TimeoutException)
{
    // Do nothing
}

Answer 5

您应该考虑使用 Microsoft 的响应式框架（又名 Rx） - NuGet System.Reactive 并添加 using System.Reactive.Linq; - 然后您可以这样做：

IObservable<Unit> tasks =
    from x1 in Observable.FromAsync(() => t1())
    from x2 in Observable.FromAsync(() => t2())
    from x3 in Observable.FromAsync(() => t3())
    from x4 in Observable.FromAsync(() => t4())
    select Unit.Default;

IObservable<Unit> timer =
    Observable
        .Timer(TimeSpan.FromSeconds(4.0))
        .Select(x => Unit.Default)

IDisposable subscription =
    Observable
        .Amb(tasks, timer)
        .Subscribe();

如果计时器 observable 在任务完成之前触发，则整个管道将被取消。不会运行不必要的任务。

如果您想手动取消，只需致电subscription.Dispose()。

代码简洁美观。

Answer 6

由于这里没有人给出更好的答案，我会自己做。

在我看来，我在问题中提供的代码最有凝聚力，也足够通用，可以在其他情况下回收。