【优化】：LINQ to Objects 查询慢

Question

我有一个困扰我的问题；它被封装为一个新的查询运算符，我做了两个版本，试图看看哪个版本更好。两者的表现都很糟糕。

第一次尝试；声明式风格

public static IEnumerable<IEnumerable<α>> Section<α>(this IEnumerable<α> source, int length)
{
    return source.Any()
        ? source.Take(length).Cons(source.Skip(length).Section(length))
        : Enumerable.Empty<IEnumerable<α>>();
}

第二次尝试：命令式“收益回报”风格

public static IEnumerable<IEnumerable<α>> Section<α>(this IEnumerable<α> source, int length)
{
    var fst = source.Take(length);
    var rst = source.Skip(length);

    yield return fst;

    if (rst.Any())
        foreach (var section in rst.Section(length))
            yield return section;
}

事实上，第二次尝试更糟糕，无论是在可读性、组合性还是速度方面。

关于如何优化它的任何线索？

Answer 1

只要有可能，我会尝试只在操作符内迭代一次源。如果源类似于Reverse() 运算符的结果，调用Any、Take 和Skip 可能会导致很多令人讨厌的性能。

目前尚不完全清楚您的操作员要做什么，但如果您可以在不多次阅读源代码的情况下做到这一点，那可能会有所帮助 - 尽管这在很大程度上取决于输入的内容。

Answer 2

如果我正确理解您的问题，您正在尝试构建一个惰性实现的枚举器，它将较大的项目集合拆分为较小的可枚举项目集合。

例如，一百万个数字的序列可以分成“部分”，每个部分只产生 100 个，而您希望这一切都懒惰地完成，即。在生成它们之前不要将 100 个项目收集到一个列表中。

首先，您的尝试将多次重复该集合，这很糟糕，因此会出现性能问题。

如果您尝试构建纯惰性实现，则应考虑以下问题：

• 您只想遍历基础集合一次
• 您应该返回可重用底层枚举器的枚举
• 您需要处理您返回的部分未完全枚举（例如，调用代码只需要这 100 项中的前 50 项）。

---

编辑：在我进入我的简单解决方案之前，这里有一些关于它的警告：

• 您不能保存每个部分以供以后使用，即。你不能这样做：collection.Sequence(10).ToArray() 来获取一组部分。
• 您不能对每个部分进行多次枚举，因为这样做会更改隐藏的底层数据结构。

基本上：我的解决方案不是通用的。如果你需要这个，你应该使用@LBushkin关于MoreLinq Batch 的评论，我会犹豫将我的代码放入类库中，它必须在需要它的地方，或者重命名为清楚的东西警告您有关它的问题。

---

这是一个简单的实现，我很确定这里有错误，所以你可能想看看为边缘情况实现大量的单元测试：

using System;
using System.Collections.Generic;
using System.Linq;

namespace ConsoleApplication20
{
    class SectionEnumerable<T> : IEnumerable<T>
    {
        private readonly IEnumerator<T> _Enumerator;

        public SectionEnumerable(IEnumerator<T> enumerator, int sectionSize)
        {
            _Enumerator = enumerator;
            Left = sectionSize;
        }

        public IEnumerator<T> GetEnumerator()
        {
            while (Left > 0)
            {
                Left--;
                yield return _Enumerator.Current;
                if (Left > 0)
                    if (!_Enumerator.MoveNext())
                        break;
            }
        }

        System.Collections.IEnumerator System.Collections.IEnumerable.GetEnumerator()
        {
            return GetEnumerator();
        }

        public int Left { get; private set; }
    }

    static class SequenceExtensions
    {
        public static IEnumerable<IEnumerable<T>> Section<T>(this IEnumerable<T> collection, int sectionSize)
        {
            if (collection == null)
                throw new ArgumentNullException("collection");
            if (sectionSize < 1)
                throw new ArgumentOutOfRangeException("sectionSize");

            using (IEnumerator<T> enumerator = collection.GetEnumerator())
            {
                while (enumerator.MoveNext())
                {
                    SectionEnumerable<T> enumerable = new SectionEnumerable<T>(enumerator, sectionSize);
                    yield return enumerable;
                    for (int index = 0; index < enumerable.Left; index++)
                        if (!enumerator.MoveNext())
                            yield break;
                }
            }
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            var sequence = Enumerable.Range(0, 100);
            var sections = sequence.Section(10);
            foreach (var section in sections)
            {
                Console.WriteLine(
                    String.Join(", ",
                    section.Take(5).ToArray().Select(i => i.ToString()).ToArray()));
            }
            Console.ReadLine();
        }
    }
}

输出：

0, 1, 2, 3, 4
10, 11, 12, 13, 14
20, 21, 22, 23, 24
30, 31, 32, 33, 34
40, 41, 42, 43, 44
50, 51, 52, 53, 54
60, 61, 62, 63, 64
70, 71, 72, 73, 74
80, 81, 82, 83, 84
90, 91, 92, 93, 94

你应该单元测试的东西：

• 空输入集合不会产生任何部分
• 元素数量恰到好处的集合，只产生一个部分
• 包含多个节大小元素（即 10、20、30 等，节大小为 5 或 10 的元素）的集合在所有预期的
• 那其实是懒惰的，如果你枚举了前10个元素的section，但是只枚举了第二个section的前5个，那么就只枚举了底层集合的前15个元素

Answer 3

这样更快吗？应该是，因为它只需要对源序列进行一次迭代。

public static IEnumerable<IEnumerable<T>> Section<T>(
    this IEnumerable<T> source, int length)
{
    return source
        .Select((x, i) => new { Value = x, Group = i / length })
        .GroupBy(x => x.Group, y => y.Value);
}

Answer 4

这是另一种不使用 linq 的方法，它比您的第二种方法快得多：

 public static IEnumerable<IEnumerable<a>> Section<a>(this IEnumerable<a> source, int length)
        {


            var enumerator = source.GetEnumerator();
            var continueLoop = true;
            do
            {
                var list = new List<a>();
                var index = 0;
                for (int i = 0; i < length; i++)
                {
                    if (enumerator.MoveNext())
                    {
                        list.Add(enumerator.Current);
                        index++;
                    }
                    else
                    {
                        continueLoop = false;
                        break;
                    }
                }
                if (list.Count > 0)
                {
                    yield return list;
                }
            } while (continueLoop);


        }

Answer 5

我怀疑您遇到的问题与枚举最终结果至少是 O(n^2) 操作这一事实有关，可能更糟；我还没有把这一切都想清楚。

这是为什么呢？好吧，假设您有 [1, 2, 3, 4, 5, 6] 并将其拆分为您认为的 { { 1, 2 }, {3, 4}, {5, 6} }

这不是你所做的。实际上，您已将其拆分为 { 取前两个，取前两个并丢弃它们，然后取接下来的两个，取前两个并丢弃，然后取下两个并丢弃它们，然后取第三个两个}

注意沿途的每一步如何重新计算结果？这是因为数组可能在对枚举的调用之间发生变化。LINQ 旨在始终为您提供最新的结果。您编写的查询意味着“跳过前四个并迭代接下来的两个”，这正是您得到的——一个在您枚举时执行该代码的查询。

原始序列是否足够小且足够快，以至于您可以将整个内容读入内存并立即将其全部拆分，而不是懒惰地尝试这样做？或者，序列是否可索引？如果你得到的只是对序列的前向访问，并且一次读入内存太大或太慢，那么你可以在这里做很多事情。但是，如果您拥有这些属性中的一个或两个，那么您至少可以将其设为线性。

Answer 6

我今天有个主意；看看这个

public static IEnumerable<α> Take<α>(this IEnumerator<α> iterator, int count)
{
    for (var i = 0; i < count && iterator.MoveNext(); i++)
        yield return iterator.Current;
}

public static IEnumerable<IEnumerable<α>> Section<α>(this IEnumerator<α> iterator, int length)
{
    var sct = Enumerable.Empty<α>();
    do
    {
        sct = iterator.Take(length).ToArray();
        if (sct.Any())
            yield return sct;
    }
    while (sct.Any());
}

这仍然不是超级优雅，但至少实现非常短且可读。

通过 IEnumerator 研究查询运算符可能会很有趣。

为了方便

public static IEnumerable<IEnumerable<α>> Section<α>(this IEnumerable<α> source, int length)
{
    using (var iterator = source.GetEnumerator())
        foreach (var e in iterator.Section(length))
            yield return e;
}

Answer 7

随着你的进步，你是否需要保留你的原始来源？如果不是，你为什么不使用递归并使用 hd :: tl 风格来拉头，将 tl 传递给递归调用，然后在任何偶数递归上将你坐在的两个部分合并为一个部分？

随着实验性 Ix 扩展的更新版本，您可以使用 Window 或 Buffer 运算符来创建 sliding window，这应该可以实现您的目标。

Answer 8

扩展方法怎么样

public static class IEnumerableExtensions
{
    public static IEnumerable<List<T>> InSetsOf<T>(this IEnumerable<T> source, int max)
    {
        List<T> toReturn = new List<T>();
        foreach(var item in source)
        {
                toReturn.Add(item);
                if (toReturn.Count == max)
                {
                        yield return toReturn;
                        toReturn = new List<T>();
                }
        }
        if (toReturn.Any())
        {
                yield return toReturn;
        }
    }
}

一些测试：

[TestFixture]
public class When_asked_to_return_items_in_sets
{
    [Test]
    public void Should_return_the_correct_number_of_sets_if_the_input_contains_a_multiple_of_the_setSize()
    {
        List<string> input = "abcdefghij".Select(x => x.ToString()).ToList();
        var result = input.InSetsOf(5);
        result.Count().ShouldBeEqualTo(2);
        result.First().Count.ShouldBeEqualTo(5);
        result.Last().Count.ShouldBeEqualTo(5);
    }

    [Test]
    public void Should_separate_the_input_into_sets_of_size_requested()
    {
        List<string> input = "abcdefghijklm".Select(x => x.ToString()).ToList();
        var result = input.InSetsOf(5);
        result.Count().ShouldBeEqualTo(3);
        result.First().Count.ShouldBeEqualTo(5);
        result.Last().Count.ShouldBeEqualTo(3);
    }
}