如何并行运行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);
}