c# - 如何打破 Parallel.ForEachAsync 循环，而不是取消它？

Question

在 .NET 5 中，我们有 Parallel.ForEach，您可以使用 ParallelLoopState.Break() 方法来停止处理额外的迭代。允许当前的完成处理。

但是新的 .NET 6 Parallel.ForEachAsync 没有 ParallelLoopState 类，所以我们不能像使用 Parallel.ForEach< 那样破坏它。那么有没有办法在 ForEachAsync 中执行相同的中断功能？ CancellationToken 传递给 func 我认为这不是正确的方法，因为您没有尝试取消正在运行的循环，而是阻止其他迭代开始。

类似于此功能但对于异步版本:

int count = 0;
Parallel.ForEach(enumerateFiles, new ParallelOptions() { CancellationToken = cancellationToken},
    (file, state) =>
    {
        Interlocked.Increment(ref count);
        if (count >= MaxFilesToProcess)
        {
            state.Break();
        }
...

作为一种解决方法，我可以在 TSource 将其传递到并行循环之前使用 .Take([xx]) 但这可能不是一个选项打破的复杂条件。

Answer 1

异步 ​​API Parallel.ForEachAsync 不提供 Stop/Break其同步副本的功能。

复制此功能的一种方法是使用 bool标志与 TakeWhile 组合LINQ 运算符:

bool breakFlag = false;
await Parallel.ForEachAsync(
    source.TakeWhile(_ => !Volatile.Read(ref breakFlag)),
    async (item, ct) =>
{
    // ...
    if (condition) Volatile.Write(ref breakFlag, true);
    // ...
});

Parallel.ForEachAsync不会积极缓冲来自源序列的元素，如 Parallel.ForEach会，所以只要满足条件，就不会启动更多的异步操作。

如果是 source是一个异步可枚举(IAsyncEnumerable<T>)，有一个兼容的 TakeWhile System.Linq.Async 中具有相同功能的运算符包。

¹ _{至少不是今天 (.NET 6)。此行为是 not documented or guaranteed .}