c# - 如何并行运行 LINQ 'let' 语句？

Question

我有这样的代码:

var list = new List<int> {1, 2, 3, 4, 5};

var result = from x in list.AsParallel()
             let a = LongRunningCalc1(x)
             let b = LongRunningCalc2(x)
             select new {a, b};

假设 LongRunningCalc 方法每个都需要 1 秒。上面的代码运行大约需要 2 秒，因为虽然 5 个元素的列表是并行操作的，但从 let 语句调用的两个方法是顺序调用的。

但是，这些方法也可以安全地并行调用。它们显然需要为 select 合并回来，但在那之前应该并行运行 - select 应该等待它们。

有什么办法可以实现吗？

Answer 1

您将无法使用查询语法或 let 操作，但您可以编写一个方法来为每个项目并行执行多个操作:

public static ParallelQuery<TFinal> SelectAll<T, TResult1, TResult2, TFinal>(
    this ParallelQuery<T> query,
    Func<T, TResult1> selector1,
    Func<T, TResult2> selector2,
    Func<TResult1, TResult2, TFinal> resultAggregator)
{
    return query.Select(item =>
    {
        var result1 = Task.Run(() => selector1(item));
        var result2 = Task.Run(() => selector2(item));
        return resultAggregator(result1.Result, result2.Result);
    });
}

这将允许你写:

var query = list.AsParallel()
    .SelectAll(LongRunningCalc1, 
        LongRunningCalc2, 
        (a, b) => new {a, b})

您也可以为额外的并行操作添加重载:

public static ParallelQuery<TFinal> SelectAll<T, TResult1, TResult2, TResult3, TFinal>
    (this ParallelQuery<T> query,
    Func<T, TResult1> selector1,
    Func<T, TResult2> selector2,
    Func<T, TResult3> selector3,
    Func<TResult1, TResult2, TResult3, TFinal> resultAggregator)
{
    return query.Select(item =>
    {
        var result1 = Task.Run(() => selector1(item));
        var result2 = Task.Run(() => selector2(item));
        var result3 = Task.Run(() => selector3(item));
        return resultAggregator(
            result1.Result,
            result2.Result,
            result3.Result);
    });
}

可以编写一个版本来处理许多在编译时未知的选择器，但要做到这一点，它们都需要计算相同类型的值:

public static ParallelQuery<IEnumerable<TResult>> SelectAll<T, TResult>(
    this ParallelQuery<T> query,
    IEnumerable<Func<T, TResult>> selectors)
{
    return query.Select(item => selectors.AsParallel()
            .Select(selector => selector(item))
            .AsEnumerable());
}
public static ParallelQuery<IEnumerable<TResult>> SelectAll<T, TResult>(
    this ParallelQuery<T> query,
    params Func<T, TResult>[] selectors)
{
    return SelectAll(query, selectors);
}