在多个IQueryable上并行运行相同的linq查询？

Question

情况:我有一个List<IQueryable<MyDataStructure>>.我想并行地对它们中的每一个运行一个linq查询,然后加入结果.

问题:如何创建我可以作为参数传递的linq查询？

示例代码:

这是一些简化的代码.首先,我有以下集合IQueryable<string>:

    public List<IQueryable<string>> GetQueries()
    {
        var set1 = (new List<string> { "hello", "hey" }).AsQueryable();
        var set2 = (new List<string> { "cat", "dog", "house" }).AsQueryable();
        var set3 = (new List<string> { "cat", "dog", "house" }).AsQueryable();
        var set4 = (new List<string> { "hello", "hey" }).AsQueryable();

        var sets = new List<IQueryable<string>> { set1, set2, set3, set4 };

        return sets;
    }

我想找到所有以字母"h"开头的单词.只需一个,IQueryable<string>这很容易:

query.Where(x => x.StartsWith("h")).ToList()

但我希望对所有IQueryable<string>对象并行运行相同的查询,然后合并结果.这是一种方法:

        var result = new ConcurrentBag<string>();
        Parallel.ForEach(queries, query =>
        {
            var partOfResult = query.Where(x => x.StartsWith("h")).ToList();

            foreach (var word in partOfResult)
            {
                result.Add(word);
            }
        });

        Console.WriteLine(result.Count);

但我希望这是一个更通用的解决方案.这样我就可以单独定义linq操作并将其作为参数传递给方法.像这样的东西:

        var query = Where(x => x.FirstName.StartsWith("d") && x.IsRemoved == false)
            .Select(x => x.FirstName)
            .OrderBy(x => x.FirstName);

        var queries = GetQueries();

        var result = Run(queries, query);

但我不知道如何做到这一点.有任何想法吗？

Answer 1

首先-给定您当前的实现，没有理由使用IQueryable<T>-您可以使用IEnumerable<T>。

然后，您可以编写一个采用IEnumerable<IEnumerable<T>>和的方法Func<IEnumerable<T>, IEnumerable<U>>来生成结果：

IEnumerable<IEnumerable<U>> QueryMultiple<T,U>(IEnumerable<IEnumerable<T>> inputs, Func<IEnumerable<T>,IEnumerable<U>> mapping)
{
     return inputs.AsParallel().Select(i => mapping(i));
}

然后，您可以将其用作：

void Run()
{
    IEnumerable<IEnumerable<YourType>> inputs = GetYourObjects();

    Func<IEnumerable<YourType>, IEnumerable<YourType>> query = i => 
       i.Where(x => x.FirstName.StartsWith("d") && x.IsRemoved == false)
        .Select(x => x.FirstName)
        .OrderBy(x => x.FirstName);

    var results = QueryMultiple(inputs, query);
}

Answer 2

所以你想要的第一件事就是采用一系列查询,执行所有查询,然后获得扁平的结果列表.这很简单:

public static IEnumerable<T> Foo<T>(IEnumerable<IQueryable<T>> queries)
{
    return queries.AsParallel()
            .Select(query => query.ToList())
            .SelectMany(results => results);
}

对于每个查询,我们执行它(调用ToList它)并且它是并行完成的,这要归功于AsParallel,然后将结果平铺成单个序列SelectMany.

您要做的另一件事是在一系列查询中为每个查询添加许多查询操作.这并不需要进行并行(由于延迟执行,在两个电话Where,OrderBy等采取几乎没有时间),并且可以通过刚刚完成Select:

var queries = GetQueries().Select(query =>
    query.Where(x => x.FirstName.StartsWith("d")
        && !x.IsRemoved)
    .Select(x => x.FirstName)
    .OrderBy(x => x.FirstName));

var results = Foo(queries);

我个人认为没有必要将这两种方法结合起来.你可以创建一个同时执行这两个方法的方法,但它们实际上是相当独立的概念,因此我认为不需要它.如果你确实想要它们的组合,这里是:

public static IEnumerable<TResult> Bar<TSource, TResult>(
    IEnumerable<IQueryable<TSource>> queries,
    Func<IQueryable<TSource>, IQueryable<TResult>> selector)
{

    return queries.Select(selector)
        .AsParallel()
        .Select(query => query.ToList())
        .SelectMany(results => results);
}

如果您愿意,可以随意制作任何一种Foo或Bar扩展方法.此外,你真的胜过重命名为更好的东西,如果你打算使用它们.