c# - 使用 TPL/async await 的递归异步调用

Question

我正在研究使用 C# 异步功能 (TPL/async/await) 以递归方式处理层次结构。这是我正在尝试做的事情的概述

我有一个作业集合要处理，如下所示。每个 Job 都有事情要做，并且可以选择有一个或多个 child 也有事情要做。所有的父作业和子作业都调用相同的函数来完成实际的“工作”，这个函数是“异步”的(下面的代码)

/*
 *  Jobs Collection
 *  |
 *  |__ Job1
 *  |    |__ Job4
 *  |    |     |__ Job7
 *  |    |
 *  |    |__ Job5
 *  |
 *  |__ Job2
 *  |    |__ Job6
 *  |
 *  |__ Job3
 *  |
 */

1. 层次结构中有 3 个级别。

2. 我想开始并行处理第一级(Job1、Job2、Job3)。

3. 一旦它们并行启动，每个单独的作业将开始处理本身，等待其处理完成(重要)，然后将继续递归地处理其子项，直到层次结构结束。子级依赖于父级处理的数据，因此他们等待父级处理完成。

4. 实际“作业”(由父项和子项调用)的处理是异步发生的，因为调用方法是异步工作的 - 因此不需要“新线程”(Task.StartNew())。

这是我用来演示场景的示例代码 -

public void Process()
{
    WebJob[] jobs = CreateWebJobs(); // dummy jobs

    // first level 
    Parallel.ForEach(jobs,
                new ParallelOptions { MaxDegreeOfParallelism = 2 }, // parallelism hardcoded for simplicity
                (job) => ExecuteJob(job));
}

private void ExecuteJob(WebJob job, [CallerMemberName] string memberName = "")
{
    Console.ForegroundColor = ConsoleColor.DarkYellow;
    Console.WriteLine("Caller> {0} :: {1} Job> {2} :: {3} Thread> {4}", memberName, "\t", job.Name, "\t", Thread.CurrentThread.ManagedThreadId);

    Task t = GetDataAsync(job);
    t.Wait(); // needed such that parent response is received before children start over (?).


    if (job.Children != null)
    {
        job.Children.ToList().ForEach((r) =>
        {
            r.ParentResponse = job.Response; // Children need parent's response
            ExecuteJob(r);
        });
    }
}

private async Task GetDataAsync(WebJob j)
{
    // This is just test code. Ideally it would be an external call to some "async" method
    await Task.Delay(1000);
    j.Response = string.Format("{0} complete", j.Name);
    Console.ForegroundColor = ConsoleColor.Cyan;
    Console.WriteLine("parentResp>> {0} :: {1} Job>> {2} :: {3} Thread>> {4}", j.ParentResponse, "\t", j.Name, "\t", Thread.CurrentThread.ManagedThreadId);
    Console.WriteLine("--------------");
}

private WebJob[] CreateWebJobs()
{
    return new WebJob[] {
        new WebJob() { Id=1, Name = "Job1", ExecURL = "http://url1", 
            Children = new WebJob[] 
            {
                new WebJob() 
                { 
                    Id=2, Name = "Job2", ExecURL = "http://url2", 
                    Children = new WebJob[] 
                    {
                        new WebJob() { Id=4, Name = "Job4", ExecURL = "http://url4" }
                    }
                },
                new WebJob() 
                { 
                    Id=3, Name = "Job3", ExecURL = "http://url3" 
                }
            }
        },
        new WebJob() { Id=5, Name = "Job5", ExecURL = "http://url5"}                
    };
}

• 流程方法首先启动所有“第一级”作业平行线。
• ExecuteJob 方法接管并递归地转到 child 完成所有处理

这工作正常，但我不确定这种递归异步模式是否是一种有效的方法。我想避免 t.Wait()。我已经在 t 上尝试了 ContinueWith，这在我的理解中似乎没有什么不同，我还阅读了有关 ForEachAsync 模式的信息，并且想知道这是否合适。该解决方案最终将成为 ASP.NET Web API 服务。对这种递归异步模式有什么想法吗？

Answer 1

如果 GetDataAsync 是您拥有的唯一阻塞操作，那么您可以自始至终使用异步编程，避免需要 Parallel.ForEach 调用或阻塞 Wait 调用。

public async Task Process()
{
    WebJob[] jobs = CreateWebJobs(); // dummy jobs

    await Task.WhenAll(jobs.Select(ExecuteJob));
}

private async Task ExecuteJob(WebJob job, [CallerMemberName] string memberName = "")
{
    Console.ForegroundColor = ConsoleColor.DarkYellow;
    Console.WriteLine("Caller> {0} :: {1} Job> {2} :: {3} Thread> {4}", memberName, "\t", job.Name, "\t", Thread.CurrentThread.ManagedThreadId);

    await GetDataAsync(job);

    if (job.Children != null)
    {
        var childTasks = job.Children.Select(r =>
        {
            r.ParentResponse = job.Response;
            return ExecuteJob(r);
        });

        await Task.WhenAll(childTasks);
    }
}

编辑:如果顶级方法应该阻塞(而不是冒着让消费者一发不可收拾的风险)，那么:

public void Process()
{
    WebJob[] jobs = CreateWebJobs(); // dummy jobs

    Task.WaitAll(jobs.Select(ExecuteJob));
}