并行地对多个IQueryable运行相同的linq查询
本文关键字:linq 查询 运行 IQueryable 并行 | 更新日期: 2023-09-27 17:58:08
情况:我有一个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);
但我不知道该怎么做。有什么想法吗?
因此,您想要的第一件事是获取一系列查询,执行所有查询,然后获得扁平的结果列表。这很简单:
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
扩展方法。此外,如果要使用它们,您真的最好将它们重命名为更好的名称。
首先,考虑到您当前的实现,没有理由使用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);
}