如何加速 LINQ WHERE？答案

【问题标题】：How to speed up LINQ WHERE?如何加速 LINQ WHERE？
【发布时间】：2019-02-15 17:12:17
【问题描述】：

我在我的 .NET winforms 应用程序（使用 .NET 4.7.1 编译）上运行了一个分析器，它指出以下函数消耗了我应用程序 73% 的 CPU 时间，这对于一个简单的应用程序来说似乎太多了效用函数：

public static bool DoesRecordExist(string keyColumn1, string keyColumn2, string keyColumn3,
        string keyValue1, string keyValue2, string keyValue3, DataTable dt)
{
    if (dt != null && dt.Rows.Count > 0) {
        bool exists = dt.AsEnumerable()
            .Where(r =>
                string.Equals(SafeTrim(r[keyColumn1]), keyValue1, StringComparison.CurrentCultureIgnoreCase) &&
                string.Equals(SafeTrim(r[keyColumn2]), keyValue2, StringComparison.CurrentCultureIgnoreCase) &&
                string.Equals(SafeTrim(r[keyColumn3]), keyValue3, StringComparison.CurrentCultureIgnoreCase)
            )
            .Any();
        return exists;
    } else {
        return false;
    }
}

这个函数的作用是传入一些键列名和匹配键值，并检查内存中c#DataTable中是否存在匹配记录。

我的应用正在处理数十万条记录，对于每条记录，必须多次调用此函数。该应用程序正在执行大量插入，并且在任何插入之前，它必须检查该记录是否已存在于数据库中。我认为针对DataTable 的内存检查比每次都返回物理数据库要快得多，所以这就是我进行内存检查的原因。每次进行数据库插入时，我都会在DataTable中进行相应的插入，以便后续检查记录是否存在准确。

所以我的问题是：有更快的方法吗？（我认为我不能避免每次都检查记录是否存在，否则我会得到重复的插入和键违规。）

编辑 #1 除了尝试我现在正在尝试的建议之外，我还想到我应该只执行一次.AsEnumerable() 并传入EnumerableRowCollection<DataRow> 而不是DataTable。你认为这会有帮助吗？

编辑#2 我刚刚做了一个受控测试，发现直接查询数据库以查看记录是否已存在比进行内存查找要慢得多。

【问题讨论】：

您需要对内存中的这么多记录执行所有这些操作吗？ C# 不会像 SQL 那样高效。我最近遇到了一个类似的问题，一个存储过程将数十万条记录卸载到一个执行所有过滤的 C# 应用程序中。重构后，SQL 完成了大部分过滤，至少加快了 10 倍。
但是每次返回数据库会慢得多，不是吗？数据表已被过滤 - 那里没有不必要的记录。
没有想法是在 SQL 服务器上执行 WHERE 子句并在一次往返中获取结果集。您可以在实体框架、Linq2Sql 甚至 Dapper 中执行此操作。前 2 个（EF 和 L2S）将您的 LINQ 语句转换为表达式，这些表达式作为 TSQL 语句编组到服务器并在那里执行。 Dapper 希望您将查询编写为参数化 SQL。
@HerrimanCoder 您是否真正分析过在内存中运行查询或针对数据库运行查询之间的区别？
@HerrimanCoder 你会发现在数据库中运行查询，由于索引、大量内存、多核以及以 500MB/s 或更快的速度运行的数据连接的好处是指数级的比通过慢速网络连接拉取所有数据然后在没有任何索引的情况下扫描内存中的所有数据要快。如果三个键列是主键的一部分，则查询可能是即时的。

标签： c# performance linq lambda

【解决方案1】：

您应该尝试并行执行，这应该是一个很好的案例，因为您提到您正在处理一个巨大的集合，如果您只想检查记录是否已经存在，则不需要有序。

bool exists = dt.AsEnumerable().AsParallel().Any((r =>
            string.Equals(SafeTrim(r[keyColumn1]), keyValue1, StringComparison.CurrentCultureIgnoreCase) &&
            string.Equals(SafeTrim(r[keyColumn2]), keyValue2, StringComparison.CurrentCultureIgnoreCase) &&
            string.Equals(SafeTrim(r[keyColumn3]), keyValue3, StringComparison.CurrentCultureIgnoreCase)
        )

【讨论】：

干得好 iSpain17，Any()（而不是 Where()...Any()）和 AsParallele() 的组合带来了显着的性能提升！
使用更多的核来分发低效的算法不是答案。更好地思考创建更高效的算法......
Ramon 在上下文中是绝对正确的，如果存在更快的算法，它可能会更快。

【解决方案2】：

您的解决方案找到在条件中评估为真的所有出现，然后您询问是否有任何出现。而是直接使用 Any。将 Where 替换为 Any。当条件的第一次真正评估时，它将停止处理。

bool exists = dt.AsEnumerable().Any(r => condition);

【讨论】：

是的，使用 list.Where().FirstOrDefault() 时也一样，删除 where 并将条件放入 Any
我建议你查看 Any 的源代码（你可以这样做here）。省略 Where 只是简写
这应该不会有太大的不同，如果有的话。 Where 只会处理元素，只要它们都没有到达Any。这就是 Linq 的强大之处。
IEnumerable 是懒惰的，因此它只会在您要求时获取新记录。如果Any() 调用返回true，那么它将不再要求新记录。您的建议虽然更简洁一些，但不会对性能产生太大影响。
仅此一项更改就将性能提高了 15%，这很好。如果这是我能做的范围，我会接受你的回答。 Lukas，我还能做些什么来加快速度吗？

【解决方案3】：

我建议您将现有记录的关键列保存在HashSet 中。我在这里使用元组，但您也可以通过覆盖 GetHashCode 和 Equals 来创建自己的 Key 结构或类。

private HashSet<(string, string, string)> _existingKeys =
    new HashSet<(string, string, string)>();

然后你可以很快地测试一个密钥的存在

if (_existingKeys.Contains((keyValue1, keyValue2, keyValue3))) {
    ...
}

不要忘记将此HashSet 与您的添加和删除保持同步。请注意，元组不能与CurrentCultureIgnoreCase 进行比较。因此，要么将所有键转换为小写，要么使用自定义结构方法，您可以使用所需的比较方法。

public readonly struct Key
{
    public Key(string key1, string key2, string key3) : this()
    {
        Key1 = key1?.Trim() ?? "";
        Key2 = key2?.Trim() ?? "";
        Key3 = key3?.Trim() ?? "";
    }

    public string Key1 { get; }
    public string Key2 { get; }
    public string Key3 { get; }

    public override bool Equals(object obj)
    {
        if (!(obj is Key)) {
            return false;
        }

        var key = (Key)obj;
        return
            String.Equals(Key1, key.Key1, StringComparison.CurrentCultureIgnoreCase) &&
            String.Equals(Key2, key.Key2, StringComparison.CurrentCultureIgnoreCase) &&
            String.Equals(Key3, key.Key3, StringComparison.CurrentCultureIgnoreCase);
    }

    public override int GetHashCode()
    {
        int hashCode = -2131266610;
        unchecked {
            hashCode = hashCode * -1521134295 + StringComparer.CurrentCultureIgnoreCase.GetHashCode(Key1);
            hashCode = hashCode * -1521134295 + StringComparer.CurrentCultureIgnoreCase.GetHashCode(Key2);
            hashCode = hashCode * -1521134295 + StringComparer.CurrentCultureIgnoreCase.GetHashCode(Key3);
        }
        return hashCode;
    }
}

另一个问题是在比较数据库键时使用当前文化是否是个好主意。具有不同文化的用户可能会得到不同的结果。最好明确指定数据库使用的相同文化。

【讨论】：

请注意，OP 小心地使用CurrentCultureIgnoreCase 进行比较，他会在这里丢失。不过，使用自定义结构很容易返回。
那行不通——StringComparison 不能比较元组
实施 IEquatable<T> 否则你的结构将被装箱，这否定了它是一个结构的意义（参见我的答案作为参考）
我不确定 HashSet 方法是否可行，因为我的 DataTables 几乎总是包含比键更多的列，而且我有时需要访问这些列的值。

【解决方案4】：

您可能想要转置您的数据结构。而不是每行都有keyColumn1、keyColumn2和keyColumn3的DataTable，而是有3个HashSet<string>，其中第一个包含所有keyColumn1值等。

这样做应该比遍历每一行要快得多：

var hashSetColumn1 = new HashSet<string>(
    dt.Rows.Select(x => x[keyColumn1]),
   StringComparison.CurrentCultureIgnoreCase);

var hashSetColumn2 = new HashSet<string>(
    dt.Rows.Select(x => x[keyColumn2]),
   StringComparison.CurrentCultureIgnoreCase);

var hashSetColumn3 = new HashSet<string>(
    dt.Rows.Select(x => x[keyColumn3]),
   StringComparison.CurrentCultureIgnoreCase);

显然，创建一次，然后维护它们（就像您当前维护 DataTable 一样）。创建它们的成本很高，但查询起来却很便宜。

然后：

bool exists = hashSetColumn1.Contains(keyValue1) &&
    hashSetColumn2.Contains(keyValue2) &&
    hashSetColumn3.Contains(keyValue3);

或者（更简洁），您可以定义自己的结构，其中包含来自 3 列的值，并使用单个 HashSet：

public struct Row : IEquatable<Row>
{
    // Convenience
    private static readonly IEqualityComparer<string> comparer = StringComparer.CurrentCultureIngoreCase;

    public string Value1 { get; }
    public string Value2 { get; }
    public string Value3 { get; }

    public Row(string value1, string value2, string value3)
    {
        Value1 = value1;
        Value2 = value2;
        Value3 = value3;
    }

    public override bool Equals(object obj) => obj is Row row && Equals(row);

    public bool Equals(Row other)
    {
        return comparer.Equals(Value1, other.Value1) &&
               comparer.Equals(Value2, other.Value2) &&
               comparer.Equals(Value3, other.Value3);
    }

    public override int GetHashCode()
    {
        unchecked
        {
            int hash = 17;
            hash = hash * 23 + comparer.GetHashCode(Value1);
            hash = hash * 23 + comparer.GetHashCode(Value2);
            hash = hash * 23 + comparer.GetHashCode(Value3);
            return hash;
        }
    }

    public static bool operator ==(Row left, Row right) => left.Equals(right);
    public static bool operator !=(Row left, Row right) => !(left == right);
}

然后你可以做一个：

var hashSet = new HashSet<Row>(dt.Select(x => new Row(x[keyColumn1], x[keyColumn2], x[keyColumn3]));

并缓存它。像这样查询：

hashSet.Contains(new Row(keyValue1, keyValue2, keyValue3));

【讨论】：

需要注意的是，HashSets 应该只在需要时创建，而不是在每次调用方法时创建
本质上，您正在为您的表创建一个内存索引，每次获取该表时都会重新创建该索引。如果表是只读的并且需要经常调用DoesRecordExist 方法，我认为这个解决方案是一个合适的选择。
同意，我建议在对 OP 的评论中直接针对数据库进行分析，我仍然认为它可能会比所有这些都快...
如果你真的需要，内存中的 SQLite 数据库也能很好地处理这样的事情

【解决方案5】：

在某些情况下，使用 LINQ 的优化不如顺序查询那么好，因此您最好用老式的方式编写查询：

public static bool DoesRecordExist(string keyColumn1, string keyColumn2, string keyColumn3,
        string keyValue1, string keyValue2, string keyValue3, DataTable dt)
{
    if (dt != null) 
    {
        foreach (var r in dt.Rows)
        {
            if (string.Equals(SafeTrim(r[keyColumn1]), keyValue1, StringComparison.CurrentCultureIgnoreCase) &&
                string.Equals(SafeTrim(r[keyColumn2]), keyValue2, StringComparison.CurrentCultureIgnoreCase) &&
                string.Equals(SafeTrim(r[keyColumn3]), keyValue3, StringComparison.CurrentCultureIgnoreCase)
            {
                return true;
            }
        }
    }
    return false;
}

但可能会有更多结构上的改进，但这取决于你是否可以使用它。

选项 1：在数据库中进行选择 您使用的是DataTable，因此您有可能从数据库中获取数据。如果您有很多记录，那么将此检查移至数据库可能更有意义。如果使用正确的索引，它可能会比内存中的表扫描更快。

选项 2：将 string.Equals+SafeTrim 替换为自定义方法 您每行最多使用 SafeTrim 三次，这会创建很多新字符串。当您创建自己的方法来比较两个字符串 (string.Equals) 与前导/尾随空格 (SafeTrim)，但没有创建一个新字符串时，这可能会更快，减少内存负载和减少垃圾收集。如果实现足以内联，那么您将获得很多性能。

选项 3：以正确的顺序检查列 确保使用正确的顺序并将匹配概率最小的列指定为keyColumn1。这将使 if 语句的结果更快地变为 false。如果 keyColumn1 在 80% 的情况下匹配，那么您需要执行更多的比较。

【讨论】：

三票否决，没有评论为什么......简单的答案（使用更多内核）适用于懒惰的程序员。更好地考虑更有效的算法......
我不知道你为什么被否决：从热路径中删除 linq 是一个完全有效的答案