在冗长的while循环中应用异步和等待

Question

我们不得不从以我们无法控制的方式提供给我们的极其有限的 Web API 调用数据。逻辑流程如下：

For each product represented by a list of ID values
  Get each batch of sub-categories of type FOO (100 records max. per call)
    Keep calling the above until no records remain
  Get each batch of sub-categories of type BAR (100 records max. per call)
    Keep calling the above until no records remain

目前这会产生近 100 个 Web API 调用（我们已经询问了提供商，没有改进 Web API 来缓解这种情况）。

我担心性能会因此而受到严重影响，因此我试图了解异步替代方案，希望这会有所帮助。一个主要问题是数据只能调用一次。之后，它会被锁定并且不会重新发送，这是我们测试的主要限制。

我已阅读 here、here 和 here，但我正在努力适应我的代码，因为我认为我需要两个 await 调用而不是一个，并且担心事情会搞砸。

任何人都可以将等待和异步逻辑应用到这个伪代码中，以便我可以阅读并尝试遵循正在发生的事情的流程吗？

public class DefaultController : Controller
{
    public ActionResult Index()
    {
        var idlist = new List<String>() {"123", "massive list of strings....", "789"};

        var xdoc = new XDocument();
        xdoc.Declaration = new XDeclaration("1.0", Encoding.Unicode.WebName, "yes");
        var xroot = new XElement("records");
        xdoc.Add(xroot);

        foreach (string id in idlist)
        {
            // Get types FOO -----------------------------------
            Boolean keepGoingFOO = true;
            while (keepGoingFOO)
            {
                // 100 records max per call
                var w = new WebServiceClient();
                request.enumType = enumType.FOO;
                var response = w.response();
                foreach (ResultItem cr in response.ResultList)
                {
                    var xe = new XElement("r");
                    // create XML 
                    xroot.Add(xe);
                }               
                keepGoingFOO = response.moreRecordsExist;
            }

            // Get types BAR -----------------------------------
            Boolean keepGoingBAR = true;
            while (keepGoingBAR)
            {
                // 100 records max per call
                var w = new WebServiceClient();
                request.enumType = enumType.BAR;
                var response = w.response();
                foreach (ResultItem cr in response.ResultList)
                {
                    var xe = new XElement("r");
                    // create XML 
                    xroot.Add(xe);
                }               
                keepGoingBAR = response.moreRecordsExist;
            }                           
        }


        return View(xdoc);
    }
}

Answer 1

为了使解决方案更优雅，我会将批处理隐藏在一个可枚举对象后面，就像 weston 建议的那样，但不是将所有项目放在一个列表中，而是在它们可用时立即使用它们（以最小化内存利用率）。

使用AsyncEnumerator NuGet Package，您可以编写如下代码：

public class DefaultController : Controller
{
    public async ActionResult Index()
    {
        var idlist = new List<String>() { "123", "massive list of strings....", "789" };

        var xdoc = new XDocument();
        xdoc.Declaration = new XDeclaration("1.0", Encoding.Unicode.WebName, "yes");
        var xroot = new XElement("records");
        xdoc.Add(xroot);

        foreach (string id in idlist) {

            // Get types FOO -----------------------------------
            var foos = EnumerateItems(enumType.FOO);

            await foos.ForEachAsync(cr => {
                var xe = new XElement("r");
                // create XML 
                xroot.Add(xe);
            });

            // Get types BAR -----------------------------------
            var bars = EnumerateItems(enumType.BAR);

            await foos.ForEachAsync(cr => {
                var xe = new XElement("r");
                // create XML 
                xroot.Add(xe);
            });
        }

        return View(xdoc);
    }

    public IAsyncEnumerable<ResultItem> EnumerateItems(enumType itemType)
    {
        return new AsyncEnumerable<ResultItem>(async yield => {

            Boolean keepGoing = true;
            while (keepGoing) {

                // 100 records max per call
                var w = new WebServiceClient();
                request.enumType = itemType;

                // MUST BE ASYNC CALL
                var response = await w.responseAsync();

                foreach (ResultItem cr in response.ResultList)
                    await yield.ReturnAsync(cr);

                keepGoing = response.moreRecordsExist;
            }
        });
    }
}

注意 1：通过使所有内容 async 实际上不会提高单个例程的性能，实际上会使其速度变慢一点（由于异步状态机和 TPL 任务的开销）。但是，它有助于更​​好地利用应用程序中的工作线程。

注意 2：您的客户端请求必须是异步的，否则优化没有意义 - 如果您同步执行它会阻塞线程并在服务器响应时等待。

注意 3：请向每个 Async 方法传递一个 CancellationToken - 这是一个很好的做法。

注 4：从代码来看，您有一个 Web 服务器，所以我不建议并行运行所有项目的所有批次并执行 Task.WhenAll 等待它们 - 如果您的服务客户端是同步的（@ 987654325@ call) 那么它只会阻塞很多线程并且你的整个网络服务可能会变得无响应。

在建议的解决方案中，您可以使用一种技巧来并行执行操作 - 您可以在处理当前批次的同时提前读取下一批：

    public IAsyncEnumerable<ResultItem> EnumerateItemsWithReadAhead(enumType itemType)
    {
        return new AsyncEnumerable<ResultItem>(async yield => {

            Task<Response> nextBatchTask = FetchNextBatch(itemType);
            Boolean keepGoing = true;

            while (keepGoing) {

                var response = await nextBatchTask;

                // Kick off the next batch request (read ahead)
                keepGoing = response.moreRecordsExist;
                if (keepGoing)
                    nextBatchTask = FetchNextBatch(itemType);

                foreach (ResultItem cr in response.ResultList)
                    await yield.ReturnAsync(cr);
            }
        });
    }

    private Task<Response> FetchNextBatch(enumType itemType)
    {
        // 100 records max per call
        var w = new WebServiceClient();
        request.enumType = itemType;

        // MUST BE ASYNC
        return w.responseAsync();
    }

Answer 2

您的代码存在多个问题，可以通过重构大大改进它，因为在 for 循环之间更改的唯一项目是 request.EnumType。并且通过正确使用异步等待可以大大提高性能 - 只要 id 是独立的，问题不是并行化两个 - 而是尽可能多地并行化。

减慢您时间的部分不是 xml 访问 - 它是 Web api 调用。

我会将它重构为

async Task<Tuple<string, enumType, XElement>> SendRequest(string id, enumType input){
    ..
}

并将for循环替换为

List<Tuple<string, enumType>> tupleList = idList.Select(id => Tuple.Create(id, enumType.BAR)).ToList();

tupleList.Concat(idList.Select(id => Tuple.Create(id, enumType.FOO)).ToList());

Task<Tuple<string, enumType, XElement>>[] all = tupleList
    .Select(c => SendRequest(c.Item1, c.Item2))
    .ToArray();

var res = await Task.WhenAll(tasks);

res 变量将包含您要添加的所有 XElement 值，这些值应该很快。您可以改为使用键值对，其中 id-enumType 的元组也是键，但想法是相同的。

Answer 3

应该让你开始：

public async ActionResult Index()
{
    var idlist = new List<string>() { "123", "massive list of strings....", "789" };
    IEnumerable<XElement> list = await ProcessList(idlist);
    //sort the list as it will be completely out of order
    return View(xdoc);
}

public async Task<IEnumerable<XElement>> ProcessList(IEnumerable<string> idlist)
{
    IEnumerable<XElement>[] processList = await Task.WhenAll(idlist.Select(FooBar));
    return processList.Select(x => x.ToList()).SelectMany(x => x);
}

private async Task<IEnumerable<XElement>> FooBar(string id)
{
    Task<IEnumerable<XElement>> foo = Foo(id);
    Task<IEnumerable<XElement>> bar = Bar(id);
    return ((await bar).Concat(await foo));
}

private async Task<IEnumerable<XElement>> Bar(string id)
{
    var localListOfElements = new List<XElement>();
    var keepGoingFoo = true;
    while (keepGoingFoo)
    {
        var response = await ServiceCallAsync(); //make sure you use the async version
        localListOfElements.Add(new XElement("r"));
        keepGoingFoo = response.moreRecordsExist;
    }
    return localListOfElements;
}

private async Task<IEnumerable<XElement>> Foo(string id)
{
    var localListOfElements = new List<XElement>();
    var keepGoingFoo = true;
    while (keepGoingFoo)
    {
        var response = await ServiceCallAsync(); //make sure you use the async version
        localListOfElements.Add(new XElement("r"));
        keepGoingFoo = response.moreRecordsExist;
    }
    return localListOfElements;
}

private async Task<Response> ServiceCallAsync()
{
    await Task.Delay(1000);//simulation
    return new Response();
}