如何在沒有左偏的情況下合并多個異步序列？-有解無憂

我有幾個AsyncEnumerable<string>s 我想合并到一個 s 中AsyncEnumerable<string>，它應該包含從這些序列同時發出的所有元素。所以我使用了System.Interactive.Async包中的Merge運算子。問題是該運算子不會將所有序列視為平等。它更喜歡從引數串列左側的序列中發出元素，而忽略引數串列右側的序列。這是一個重現這種不良行為的最小示例：

var sequence_A = Enumerable.Range(1, 5).Select(i => $"A{i}").ToAsyncEnumerable();
var sequence_B = Enumerable.Range(1, 5).Select(i => $"B{i}").ToAsyncEnumerable();
var sequence_C = Enumerable.Range(1, 5).Select(i => $"C{i}").ToAsyncEnumerable();
var merged = AsyncEnumerableEx.Merge(sequence_A, sequence_B, sequence_C);
await foreach (var item in merged) Console.WriteLine(item);

此代碼片段還依賴于System.Linq.Async包。從sequence_A開始發出 5 個元素"A"，sequence_B從開始發出 5 個元素"B"，從開始sequence_C發出 5 個元素"C"。

輸出（不良）：

A1
A2
A3
A4
A5
B1
B2
B3
B4
B5
C1
C2
C3
C4
C5

在 Fiddle 上試試。

理想的輸出應如下所示：

A1
B1
C1
A2
B2
C2
A3
B3
C3
A4
B4
C4
A5
B5
C5

如果所有序列都有下一個元素可用，合并后的序列應該從每個序列中拉出一個元素，而不是從最左邊的序列中重復拉出元素。

如何確保我的序列與公平合并？我正在尋找具有理想行為的官方軟體包中的運算子組合，或者尋找Merge可以執行我想要的自定義運算子。

澄清：我對并發 Merge功能感興趣，其中同時觀察所有源序列，并且來自任何序列的任何發射都傳播到合并的序列。當多個序列可以立即發射一個元素時，公平的概念適用，在這種情況下，它們的發射應該是交錯的。在相反的情況下，當沒有立即可用的元素時，規則是“先到先到”。

更新：這是一個更真實的演示，包括生產者序列和消費列舉回圈中的延遲。它模擬了消耗由最左側序列生成的值比生成這些值所需的時間更長的情況。

var sequence_A = Produce("A", 200, 1, 2, 3, 4, 5);
var sequence_B = Produce("B", 150, 1, 2, 3, 4, 5);
var sequence_C = Produce("C", 100, 1, 2, 3, 4, 5);
var merged = AsyncEnumerableEx.Merge(sequence_A, sequence_B, sequence_C);
await foreach (var item in merged)
{
    Console.WriteLine(item);
    await Task.Delay(item.StartsWith("A") ? 300 : 50); // Latency
}

async IAsyncEnumerable<string> Produce(string prefix, int delay, params int[] values)
{
    foreach (var value in values)
    {
        var delayTask = Task.Delay(delay);
        yield return $"{prefix}{value}";
        await delayTask; // Latency
    }
}

The result is an undesirable bias for the values produced by the sequence_A:

A1
A2
A3
A4
A5
B1
B2
C1
B3
C2
B4
C3
C4
B5
C5

Try it on Fiddle.

uj5u.com熱心網友回復：

這是最終的代碼。該演算法已被修改以適應 OP。我在下面留下了原始代碼。

這使用了一個貪心演算法：回傳第一個可用值，并且不嘗試依次合并。每次任務完成時，同一個列舉器的下一個列舉器都會退到后面，以確保公平。

演算法如下：

該函式接受一個params源陣列。
如果沒有提供源可列舉，則提前紓困。
創建一個串列以將列舉器及其各自的任務作為元組保存。
獲取每個列舉器，呼叫MoveNextAsync并將該對存盤在串列中。
在一個回圈中，呼叫Task.WhenAny整個串列。
獲取結果Task并在串列中找到它的位置。
將元組保存在變數中并將其從串列中洗掉。
如果它回傳true，則yield該值并MoveNextAsync再次呼叫匹配的列舉器，將結果元組推到串列的后面。
如果回傳false，則回傳Dispose列舉數。
繼續回圈直到串列為空。
finally block 處理任何剩余的列舉數。
還有提供取消令牌的多載

在分配等方面有一些效率。我把它作為練習留給讀者。

 public static IAsyncEnumerable<T> Interleave<T>(params IAsyncEnumerable<T>[] sources) =>
     Interleave(default, sources);
 
 public static async IAsyncEnumerable<T> Interleave<T>([EnumeratorCancellation] CancellationToken token, IAsyncEnumerable<T>[] sources)
 {
     if(sources.Length == 0)
         yield break;
     var enumerators = new List<(IAsyncEnumerator<T> e, Task<bool> t)>(sources.Length);
     try
     {
         for(var i = 0; i < sources.Length; i  )
         {
             var e = sources[i].GetAsyncEnumerator(token);
             enumerators.Add((e, e.MoveNextAsync().AsTask()));
         }

         do
         {
             var taskResult = await Task.WhenAny(enumerators.Select(tuple => tuple.t));
             var ind = enumerators.FindIndex(tuple => tuple.t == taskResult);
             var tuple = enumerators[ind];
             enumerators.RemoveAt(ind);
             if(taskResult.Result)
             {
                 yield return tuple.e.Current;
                 enumerators.Add((tuple.e, tuple.e.MoveNextAsync().AsTask()));
             }
             else
             {
                 try
                 {
                     await tuple.e.DisposeAsync();
                 }
                 catch
                 { //
                 }
             }
         } while (enumerators.Count > 0);
     }
     finally
     {
         for(var i = 0; i < enumerators.Count; i  )
         {
             try
             {
                 await enumerators[i].e.DisposeAsync();
             }
             catch
             { //
             }
         }
     }
 }

dotnetfiddle

編輯以下內容并不是 OP 想要的，因為 OP 希望回傳任何結果，以先到者為準。我將把它留在這里，因為它很好地演示了這個演算法。

這是 asyncInterleave或Mergealgorithm的完整實作，在 SQL 術語中更常見的是Merge-Concatenation。

演算法如下：

該函式接受一個params源陣列。
如果沒有提供源可列舉，則提前紓困。
創建一個串列來保存列舉器。
獲取每個列舉器并將其存盤在串列中。
在一個回圈中，獲取每個列舉器和MoveNextAsync.
如果它回傳true，則yield該值并遞增回圈計數器。如果翻了，就回到開頭。
If it returns false, then Dispose it and remove from the list. Do not increment counter.
Continue looping until there are no more enumerators.
finally block disposes any remaining enumerators.
There is also an overload to provide a cancellation token

 public static IAsyncEnumerable<T> Interleave<T>(params IAsyncEnumerable<T>[] sources) =>
     Interleave(default, sources);
 
 public static async IAsyncEnumerable<T> Interleave<T>([EnumeratorCancellation] CancellationToken token, IAsyncEnumerable<T>[] sources)
 {
     if(sources.Length == 0)
         yield break;
     var enumerators = new List<IAsyncEnumerator<T>>(sources.Length);
     try
     {
         for(var i = 0; i < sources.Length; i  )
             enumerators.Add(sources[i].GetAsyncEnumerator(token));

         var j = 0;
         do
         {
             if(await enumerators[j].MoveNextAsync())
             {
                 yield return enumerators[j].Current;
                 j  ;
                 if(j >= enumerators.Count)
                     j = 0;
             }
             else
             {
                 try
                 {
                     await enumerators[j].DisposeAsync();
                 }
                 catch
                 { //
                 }
                 enumerators.RemoveAt(j);
             }
         } while (enumerators.Count > 0);
     }
     finally
     {
         for(var i = 0; i < enumerators.Count; i  )
         {
             try
             {
                 await enumerators[i].DisposeAsync();
             }
             catch
             { //
             }
         }
     }
 }

dotnetfiddle

This can obviously be significantly simplified if you only have a fixed number of source enumerables.

uj5u.com熱心網友回復：

該示例有點做作，因為所有結果都可以立即獲得。即使添加了一點延遲，結果也會好壞參半：

var sequence_A = AsyncEnumerable.Range(1, 5)
    .SelectAwait(async i =>{ await Task.Delay(i); return $"A{i}";});
var sequence_B = AsyncEnumerable.Range(1, 5)
    .SelectAwait(async i =>{ await Task.Delay(i); return $"B{i}";});
var sequence_C = AsyncEnumerable.Range(1, 5)
    .SelectAwait(async i =>{ await Task.Delay(i); return $"C{i}";});
var sequence_D = AsyncEnumerable.Range(1, 5)
    .SelectAwait(async i =>{ await Task.Delay(i); return $"D{i}";});

await foreach (var item in seq) Console.WriteLine(item);

每次都會產生不同的混合結果：

B1
A1
C1
D1
D2
A2
B2
C2
D3
A3
B3
C3
C4
A4
B4
D4
D5
A5
B5
C5

該方法的評論解釋說它被重新實作為更便宜和更公平：

//
// This new implementation of Merge differs from the original one in a few ways:
//
// - It's cheaper because:
//   - no conversion from ValueTask<bool> to Task<bool> takes place using AsTask,
//   - we don't instantiate Task.WhenAny tasks for each iteration.
// - It's fairer because:
//   - the MoveNextAsync tasks are awaited concurently, but completions are queued,
//     instead of awaiting a new WhenAny task where "left" sources have preferential
//     treatment over "right" sources.
//

轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/411871.html

標籤：

上一篇：等待任務的Task.Run()差異

下一篇：Firebase云功能在回傳前結束