ParallelEnumerable.GroupBy：它有多懒？

Question

MSDN 表示，ParallelEnumerable.GroupBy 根据指定的键选择器函数对序列的元素进行并行分组。 所以我的问题是：它有多懒？

很明显，ParallelQuery> 是惰性的。但是 IGrouping 本身呢，它也很懒吗？

因此，如果我执行以下操作：

var entities = sites.AsParallel()
                         .Select(x => GetDataItemsFromWebsiteLazy(x))
                         .SelectMany(x => x)
                         .GroupBy(dataItem => dataItem.Url.Host)
                         .AsParallel()
                             .SelectMany(x => TransformToEntity(x));

在所有站点获取结果后是否会首次调用 TransformToEntity ？
或者一旦第一个 GetDataItemsFromWebsiteLazy() 方法将产生返回一个元素？

所有这一切的要点是并行向不同主机发出请求。
数据处理如下。对于集合中的每个网站：

1. 请求网站
2. 解析响应并提取另一个网站 url
3. 通过提取的 url 请求网站
4. 解析响应并根据获得的数据创建实体

Answer 1

GroupBy 运算符，均位于 PLINQ 和 LINQ，是通过使用延迟执行来实现的。此方法表示的查询在枚举之前不会执行。但执行查询不会产生您想要的行为。 GroupBy 产生分组，并且在完全枚举源序列之前不会产生分组。 IGrouping 作为一个结构，其内部只是一个带有 Key 属性的不可变数组。它是一个物化集合，而不是一个延迟枚举。当 GroupBy 生成分组时，它包含源序列中具有特定键的所有元素。未来不会再添加其他元素。

显然你想要实现的是所谓的“任务并行性”。这意味着对相同的数据执行不同的操作，这些操作彼此并行运行。相比之下，“数据并行”是对数据的不同子集并行执行相同的操作。 PLINQ 库旨在支持数据并行性。实现任务并行性的首选工具是 TPL 数据流 库。如果您确实想使用 PLINQ 执行此操作，这是可行的，但很棘手。您可能仍然会错过诸如反压之类的有用功能，该功能在 TPL Dataflow 中通过 BoundedCapacity 选项。

Answer 2

事实上，GroupBy 扩展根本不懒惰（或者更准确地说，根本不延迟），这可以通过以下测试程序轻松演示

void Main()
{
    var source = new[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 }.AsParallel();
    var groupEven = GetEvenNumbersUsingGroupBy(source);
    // foreach (int num in groupEven) { }
}

IEnumerable<int> GetEvenNumbersUsingGroupBy(IEnumerable<int> source)
{
    Console.WriteLine("Method called: GetEvenNumbersUsingGroupBy");
    var grouped = source.GroupBy(i => i % 2);
    return grouped.Where(g => g.Key == 0).Single();
}

：程序输出以下内容：

调用方法：GetEvenNumbersUsingGroupBy

这意味着即使我们从未真正迭代 GetEvenNumbersUsingGroupBy 方法的结果，它仍然会被执行。

这与使用 yield 语句的普通延迟枚举形成对比，如下所示：

void Main()
{
    var yieldEven = GetEvenNumbersUsingYield(source);
    foreach (int num in yieldEven) { }
    foreach (int num in yieldEven) { }
}

IEnumerable<int> GetEvenNumbersUsingYield(IEnumerable<int> source)
{
    Console.WriteLine("Method called: GetEvenNumbersUsingYield");
    foreach (int i in source)
        if ((i % 2) == 0)   
            yield return i;
}

这将打印以下内容：

调用方法：GetEvenNumbersUsingYield
调用方法：GetEvenNumbersUsingYield

换句话说，每次迭代结果时，都会重新计算结果，这是延迟计算的典型特征（与直接延迟加载相反，后者在第一次计算后缓存结果）。

请注意，无论您是否使用 AsParallel，这都是一样的；它是 GroupBy 扩展的一个特征（根据定义，它需要构建哈希表或其他类型的查找来存储各个组）并且完全独立于并发性。

如果您考虑如何实现延迟分组函数，就很容易理解为什么会出现这种情况；为了迭代单个组的所有元素，您必须迭代整个序列以确保您实际上已经覆盖了该组的所有元素。因此，虽然技术上可以推迟整个序列的一次性迭代，但在大多数情况下可能不值得，因为它将具有与急切的完全相同的内存和 CPU 特性。 - 加载版本。