AsParallel() 和 Any() 是什么？

Question

我看到了这段代码，它使用AsParallel()和Any()来检查条件：

bool IsAnyDeviceConnected()
{
   return m_devices.Any(d => d.IsConnected);
}

为了让它更快：

bool IsAnyDeviceConnected()
{
   return m_devices.AsParallel().Any(d => d.IsConnected);
}

但是看看Any()：

 internal static bool Any<T>(this IEnumerable<T> source, Func<T, bool> predicate) {
            foreach (T element in source) {
                if (predicate(element)) {
                    return true;
                }
            }
            return false;
        }

我并没有看到（显然） - 它关心其他工作的取消 - 一旦发现。

然而 - 这段（其他的）代码确实会“尽快完成” + 取消其他未来的工作：

bool IsAnyDeviceConnected()
{
   var res = Parallel.ForEach(m_devices,
      (d,loopState) => {  
         if (d.IsConnected) 
            loopState.Stop();
      });
   return !res.IsCompleted;
}

问题：

我的诊断是否正确？在AsParallel上下文中，Any()一旦找到项目，不会取消其他线程吗？

注意，我担心我可能看错了源代码。

- Royi Namir

---

你正在查看错误的源代码。你需要查看ParalellEnumerable.Any，它比你看到的Enumerable.Any要复杂得多。 - Sriram Sakthivel

Answer 1

AsParallel()返回一个ParallelQuery，因此如果您调用AsParallel().Any(...)，您不是在调用Enumerable.Any，而是在调用ParallelEnumerable.Any。

ParallelEnumerable.Any的参考源代码在这里。

当您深入挖掘例如AnyAllSearchOperatorEnumerator类时，您会发现有一个名为resultFoundFlag的标志被用来告诉其他工作者找到结果，以便他们可以停止搜索。

- sloth

Answer 2

你正在查看错误的代码。 AsParallel 返回一个ParallelQuery<TSource>，而ParellelQuery还有另一个重载的Any。

'Any'创建一个新的AnyAllSearchOperator对象并对其进行聚合。如果你深入挖掘这些方法调用和对象的链式结构，你会发现QueryOpeningEnumerator确实支持取消操作。

---

不幸的是，参考源链接到那些特定成员函数的链接存在问题。

- Dirk

Answer 3

您正在查看错误的代码。 ParallelEnumerable.AsParallel 返回一个 ParallelQuery<>。ParallelEnumerable 还定义了自己的 Any 扩展方法。

为了指定取消、并行度等级等，您需要使用 ParallelEnumerable 的 WithXXX 扩展方法，例如 WithCancellation 和 WithDegreeOfParallelism。ParallelEnumerable.Any 不允许您指定这些选项，以保留与 Enumerable.Any 类似的签名。

- Panagiotis Kanavos