优化导致并行执行的并行选择

Question

我正在尝试找到执行以下流程的最佳方式

从流中获取 X 标识符列表
对于每个 X 标识符，转到 DB 并使用复杂查询获取 Y 标识符列表（应该并行）
合并列表并区分它
对于每个 Y 标识符执行一个长时间运行的任务（应该并行）

2-4 中的所有流程都应在与 1 不同的线程中完成，这样一旦操作 1 调用操作 2，它就会继续执行其他操作。

目前我的流程是这样的（这是针对 2-4 并在 1 之后调用的）：

private void PerformActionAsync(List<long> XIds)
{
     var affectedYIds = XIds.AsParallel().SelectMany(xId =>
     {
         return GetAffectedYIdsLongRunning(xId);   

     }).Distinct();

     affectedYIds.ForAll(yId=>
     {
         ExcuteLongRunningAction(yId);
     });
 }

这不起作用，因为 SelectMany 和 ForAll 仍然阻塞调用线程，我可以用新任务创建替换 ForAll，但 SelectMany 仍然会阻塞调用线程。如何以真正异步的方式执行 SelectMany？

目前我最好的解决方案是用Taks.Run包装整个方法实现，问题是是否有更好的方法。

Answer 1

您可以将您的代码包装在 Task.Run 和 return 该任务中。这将使所有内容运行在后台。

以下单元测试显示了用法：

[TestClass]
public class PLinqTests
{
    private static readonly Stopwatch Watch = Stopwatch.StartNew();

    [TestMethod]
    public async Task TestPerformAsync()
    {
        await PerformActionAsync(Enumerable.Range(0, 10));
    }

    private Task PerformActionAsync(IEnumerable<int> xIds)
    {
        return Task.Run(() =>
        {
            var affectedYIds = xIds
                .AsParallel()
                .WithDegreeOfParallelism(5)
                .SelectMany(this.GetAffectedYIdsLongRunning)
                .Distinct();

            affectedYIds.ForAll(this.ExcuteLongRunningAction);
        });
    }

    private void ExcuteLongRunningAction(int yId)
    {
        Thread.Sleep(1000);
        Console.WriteLine("Executed {0} at {1}.", yId, Watch.Elapsed.Seconds);
    }

    private IEnumerable<int> GetAffectedYIdsLongRunning(int xId)
    {
        Thread.Sleep(1000);
        Console.WriteLine("Getting Affected for {0} at {1}.", xId, Watch.Elapsed.Seconds);

        return Enumerable.Range(30, 10);
    }
}

输出：

Getting Affected for 0 at 1.
Getting Affected for 1 at 1.
Getting Affected for 2 at 1.
Getting Affected for 4 at 2.
Getting Affected for 3 at 2.
Getting Affected for 5 at 2.
Getting Affected for 6 at 2.
Getting Affected for 7 at 3.
Getting Affected for 8 at 3.
Getting Affected for 9 at 3.
Executed 32 at 3.
Executed 31 at 3.
Executed 30 at 4.
Executed 34 at 4.
Executed 33 at 4.
Executed 37 at 4.
Executed 36 at 4.
Executed 35 at 5.
Executed 39 at 5.
Executed 38 at 5.

优化导致并行执行的并行选择

Optimizing a Parallel selection leading to Parallel excution

c#

parallel-processing

plinq