c# - 如何同时压缩两个 IAsyncEnumerable？

Question

我有两个异步序列，我想将它们成对“压缩”，为此我使用了 Zip来自 System.Linq.Async 的运算符包裹。不过，这个运算符的行为方式并不理想，至少对于我来说是这样。它不是同时枚举两个序列，而是按顺序枚举它们，结果是将延迟相加。我的每个序列平均每秒发出一个元素，并且我预计组合序列也会每秒发出压缩对，但实际上我每 2 秒发出一对。下面是演示此行为的最小示例:

static async IAsyncEnumerable<int> First()
{
    for (int i = 1; i <= 5; i++) { await Task.Delay(1000); yield return i; }
}

static async IAsyncEnumerable<int> Second()
{
    for (int i = 1; i <= 5; i++) { await Task.Delay(1000); yield return i; }
}

var stopwatch = Stopwatch.StartNew();
await foreach (var pair in First().Zip(Second()))
    Console.WriteLine(pair);
Console.WriteLine($"Duration: {stopwatch.ElapsedMilliseconds:#,0} msec");

输出:

(1, 1)
(2, 2)
(3, 3)
(4, 4)
(5, 5)
Duration: 10,155 msec

Try it on Fiddle .

有什么办法可以让程序在 5 秒而不是 10 秒内完成Zip这两个序列吗？我对具有理想行为的自定义运算符或官方包中的运算符组合感兴趣。

Answer 1

类似这样的东西似乎有效:

public static async IAsyncEnumerable<(TFirst, TSecond)> Zip<TFirst, TSecond>(this IAsyncEnumerable<TFirst> first, IAsyncEnumerable<TSecond> second)
{
    await using var e1 = first.GetAsyncEnumerator();    
    await using var e2 = second.GetAsyncEnumerator();
    
    while (true)
    {
        var t1 = e1.MoveNextAsync().AsTask();
        var t2 = e2.MoveNextAsync().AsTask();
        await Task.WhenAll(t1, t2);
        
        if (!t1.Result || !t2.Result)
            yield break;
        
        yield return (e1.Current, e2.Current);
    }
}

查看 dotnetfiddle.net .

当然，这会遗漏空检查之类的东西，因此可以进行一些改进:这留给读者作为练习。

我也不相信 Task.WhenAll 比 bool r1 = wait t1; 更好。 bool r2 = 等待 t2; if (!r1 || !r2) Yield Break; 此处。