停止Parallel.ForEachAsync

Question

在C#中，我有兴趣停止一个Parallel.ForEachAsync循环(考虑到differences between Stop and Break；对于Parallel.ForEach，我可以这样做：

Parallel.ForEach(items, (item, state) =>
{
    if (cancellationToken.IsCancellationRequested)
    {
        state.Stop();
        return;
    }

    // some process on the item
    Process(item);
});

但是，由于我有一个需要异步执行的进程，所以我切换到了Parallel.ForEachAsync。ForEachAsync没有Stop()方法，我可以按下面的方式对循环进行break，但我想知道这是否是打破循环的最有效方法(换句话说，当循环接收到取消请求时需要尽快停止)。

await Parallel.ForEachAsync(items, async (item, state) =>
{
    if (cancellationToken.IsCancellationRequested)
    {
        return;
    }

    // some async process on the item
    await ProcessAsync(item);
});

Answer 1

Parallel.ForEachAsync lambda的第二个参数是CancellationToken。这个令牌是由API提供的，它与您在ParallelOptions中传递的令牌不同。您可以将此令牌转发到lambda中调用的任何异步方法。如果您调用不可取消的方法，那么最好的方法是在lambda内部的策略位置调用ThrowIfCancellationRequested：

var cts = new CancellationTokenSource();
var options = new ParallelOptions() { CancellationToken = cts.Token };

try
{
    await Parallel.ForEachAsync(items, options, async (item, ct) =>
    {
        //...
        ct.ThrowIfCancellationRequested();
        //...
        await ProcessAsync(item, ct);
        //...
        ct.ThrowIfCancellationRequested();
        //...
    });
}
catch (OperationCanceledException ex)
{
    // ...
}

上面示例中的ct作为参数提供的令牌不仅在取消ParallelOptions.CancellationToken时被取消，而且在ProcessAsync操作失败时也被取消。这种机制允许更快地传播异常。当发生错误时，并行循环不会立即完成，因为它遵循不允许触发和遗忘操作的原则。由循环在内部启动的所有操作必须在整个循环成功完成或失败之前完成。lambda中的令牌使得将此延迟降到最小成为可能。

Answer 2

你需要这样的东西：

await Parallel.ForEachAsync(items, async (item, state) =>
{
    await ProcessAsync(item, cancellationToken);
});


async Task ProcessAsync(string item, CancellationToken ct)
{
    while (!ct.IsCancellationRequested)
    {
        //Process
    }
}