c# - 如何异步查询两个 IAsyncEnumerables

Question

我有两种方法连接到 Foo 的两个不同来源s 返回两个 IAsyncEnumerable<Foo> .我需要获取所有 Foo s 来自两个来源，然后才能处理它们。
问题 :我想同时(异步)查询两个源，即。不等Source1在开始枚举之前完成枚举Source2 .根据我的理解，这就是方法 SequentialSourcesQuery 中发生的情况。下面的例子，我说得对吗？
对于常规任务，我只会开始第一个任务，然后是第二个，然后调用 await Task.WhenAll .但我对如何处理 IAsyncEnumerable 有点困惑.

public class FoosAsync
{
    public async IAsyncEnumerable<Foo> Source1() { }

    public async IAsyncEnumerable<Foo> Source2() { }

    public async Task<List<Foo>> SequentialSourcesQuery()
    {
        List<Foo> foos = new List<Foo>();

        await foreach (Foo foo1 in Source1())
        {
            foos.Add(foo1);
        }

        await foreach (Foo foo2 in Source2())
        { //doesn't start until Source1 completed the enumeration? 
            foos.Add(foo2);
        }

        return foos;
    }
}

Answer 1

您可以利用图书馆 System.Linq.Async和 System.Interactive.Async (由属于 .NET 基金会的 RxTeam 所有)。它们包含像 Merge 这样的操作符和 ToListAsync 这可以轻松解决您的问题。

// Merges elements from all of the specified async-enumerable sequences
// into a single async-enumerable sequence.
public static IAsyncEnumerable<TSource> Merge<TSource>(
    params IAsyncEnumerable<TSource>[] sources);

// Creates a list from an async-enumerable sequence.
public static ValueTask<List<TSource>> ToListAsync<TSource>(
    this IAsyncEnumerable<TSource> source,
    CancellationToken cancellationToken = default);

把所有东西放在一起:

public Task<List<Foo>> SequentialSourcesQuery()
{
    return AsyncEnumerableEx.Merge(Source1(), Source2()).ToListAsync().AsTask();
}

意识到这些库专注于提供丰富的功能集，而不是性能或效率。因此，如果一流的性能对您的用例很重要，请查看 niki.kante 的 solution很可能会胜过上述基于运算符的方法。