时间戳

首页 > 解决方案 > 如何在项目准备好后立即解析 IAsyncEnumerables 列表

c# - 如何在项目准备好后立即解析 IAsyncEnumerables 列表

问题描述

public async IAsyncEnumerable<Entity> FindByIds(List<string> ids)
    {
        List<List<string>> splitIdsList = ids.Split(5);

        var entityList = splitIdsList.Select(x => FindByIdsQuery(x)).ToList();

        foreach (var entities in entityList)
        {
            await foreach (var entity in entities)
            {
                yield return entity;
            }
        }
    }

private async IAsyncEnumerable<Entity> FindByIdsQuery(List<string> ids)
    {
        var result = await Connection.QueryAsync(query, new {ids})

        foreach (var entity in result)
        {
            yield return entity;
        }
    }

如果我向这个函数发送 25 个 ID。第一个 FindByIdsQuery 需要 5000 毫秒。其他 4 个 FindByIdsQuery 需要 100 毫秒。然后这个解决方案不会输出任何实体，直到 5000 毫秒之后。是否有任何解决方案可以在有任何人输出时立即开始输出实体。或者，如果您可以在 Task 中执行类似操作，则使用Task.WhenAny.

需要明确的是：5 个查询中的任何一个都可能需要 5000 毫秒。

标签： c#.net.net-coreiasyncenumerable

解决方案

从您的评论中，我理解了您的问题。您基本上要寻找的是某种“ SelectMany”运算符。该运算符将开始等待所有的IAsyncEnumerables并按它们来的顺序返回项目，而不管源异步枚举的顺序是什么。

我希望，默认值AsyncEnumerable.SelectMany会这样做，但我发现这不是真的。它遍历源枚举，然后遍历整个内部枚举，然后继续下一个。所以我破解了同时SelectMany正确等待所有内部异步枚举的变体。请注意，我不保证正确性，也不保证安全。零错误处理。

/// <summary>
/// Starts all inner IAsyncEnumerable and returns items from all of them in order in which they come.
/// </summary>
public static async IAsyncEnumerable<TItem> SelectManyAsync<TItem>(IEnumerable<IAsyncEnumerable<TItem>> source)
{
    // get enumerators from all inner IAsyncEnumerable
    var enumerators = source.Select(x => x.GetAsyncEnumerator()).ToList();

    List<Task<(IAsyncEnumerator<TItem>, bool)>> runningTasks = new List<Task<(IAsyncEnumerator<TItem>, bool)>>();

    // start all inner IAsyncEnumerable
    foreach (var asyncEnumerator in enumerators)
    {
        runningTasks.Add(MoveNextWrapped(asyncEnumerator));
    }

    // while there are any running tasks
    while (runningTasks.Any())
    {
        // get next finished task and remove it from list
        var finishedTask = await Task.WhenAny(runningTasks);
        runningTasks.Remove(finishedTask);

        // get result from finished IAsyncEnumerable
        var result = await finishedTask;
        var asyncEnumerator = result.Item1;
        var hasItem = result.Item2;

        // if IAsyncEnumerable has item, return it and put it back as running for next item
        if (hasItem)
        {
            yield return asyncEnumerator.Current;

            runningTasks.Add(MoveNextWrapped(asyncEnumerator));
        }
    }

    // don't forget to dispose, should be in finally
    foreach (var asyncEnumerator in enumerators)
    {
        await asyncEnumerator.DisposeAsync();
    }
}

/// <summary>
/// Helper method that returns Task with tuple of IAsyncEnumerable and it's result of MoveNextAsync.
/// </summary>
private static async Task<(IAsyncEnumerator<TItem>, bool)> MoveNextWrapped<TItem>(IAsyncEnumerator<TItem> asyncEnumerator)
{
    var res = await asyncEnumerator.MoveNextAsync();
    return (asyncEnumerator, res);
}

然后，您可以使用它来合并所有可枚举而不是第一个 foreach：

    var entities = SelectManyAsync(splitIdsList.Select(x => FindByIdsQuery(x)));

    return entities;

推荐阅读