我正在嘗試實作一個異步方法，該方法采用一個ChannelReader<T>s 陣列，并從任何具有可用專案的通道中獲取一個值。它是一種與該方法具有類似功能的BlockingCollection<T>.TakeFromAny方法，具有以下簽名：
public static int TakeFromAny(BlockingCollection<T>[] collections, out T item,
CancellationToken cancellationToken);
此方法回傳collections陣列中從中洗掉專案的索引。async方法不能有引數out，所以我要實作的 API 是這樣的：
public static Task<(T Item, int Index)> TakeFromAnyAsync<T>(
ChannelReader<T>[] channelReaders,
CancellationToken cancellationToken = default);
該方法應該異步讀取一個專案，并回傳消耗的專案以及陣列TakeFromAnyAsync<T>中關聯通道的索引。channelReaders如果所有通道都已完成（成功或有錯誤），或者在 期間全部完成await，則該方法應異步拋出ChannelClosedException.
我的問題是：如何實作該TakeFromAnyAsync<T>方法？實作看起來相當棘手。很明顯，在任何情況下，該方法都不應從渠道中消耗超過一項。此外，它不應該留下一勞永逸的任務，或者讓一次性資源閑置。該方法通常會在回圈中呼叫，因此它也應該相當有效。它的復雜度不應該比 O(n) 差，其中n通道的數量。
要了解此方法在哪里有用，您可以查看Go語言的select宣告。從旅游：
該
select陳述句讓 goroutine 等待多個通信操作。A
select阻塞，直到它的一個案例可以運行，然后它執行那個案例。如果多個都準備好了，它會隨機選擇一個。
select {
case msg1 := <-c1:
fmt.Println("received", msg1)
case msg2 := <-c2:
fmt.Println("received", msg2)
}
在上面的例子中，要么從通道中獲取一個值c1并分配給變數msg1，要么從通道中獲取一個值c2并分配給變數msg2。Goselect陳述句不限于從通道中讀取。它可以包括多種異構情況，例如寫入有界通道、等待計時器等。復制 Goselect陳述句的全部功能超出了這個問題的范圍。
uj5u.com熱心網友回復：
我想出了這樣的事情：
public static async Task<(T Item, int Index)> TakeFromAnyAsync<T>(
ChannelReader<T>[] channelReaders,
CancellationToken cancellationToken = default)
{
if (channelReaders == null)
{
throw new ArgumentNullException(nameof(channelReaders));
}
if (channelReaders.Length == 0)
{
throw new ArgumentException("The list cannot be empty.", nameof(channelReaders));
}
if (channelReaders.Length == 1)
{
return (await channelReaders[0].ReadAsync(cancellationToken), 0);
}
// First attempt to read an item synchronosuly
for (int i = 0; i < channelReaders.Length; i)
{
if (channelReaders[i].TryRead(out var item))
{
return (item, i);
}
}
using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken))
{
var waitToReadTasks = channelReaders
.Select(it => it.WaitToReadAsync(cts.Token).AsTask())
.ToArray();
var pendingTasks = new List<Task<bool>>(waitToReadTasks);
while (pendingTasks.Count > 1)
{
var t = await Task.WhenAny(pendingTasks);
if (t.IsCompletedSuccessfully && t.Result)
{
int index = Array.IndexOf(waitToReadTasks, t);
var reader = channelReaders[index];
// Attempt to read an item synchronosly
if (reader.TryRead(out var item))
{
if (pendingTasks.Count > 1)
{
// Cancel pending "wait to read" on the remaining readers
// then wait for the completion
try
{
cts.Cancel();
await Task.WhenAll((IEnumerable<Task>)pendingTasks);
}
catch { }
}
return (item, index);
}
// Due to the race condition item is no longer available
if (!reader.Completion.IsCompleted)
{
// .. but the channel appears to be still open, so we retry
var waitToReadTask = reader.WaitToReadAsync(cts.Token).AsTask();
waitToReadTasks[index] = waitToReadTask;
pendingTasks.Add(waitToReadTask);
}
}
// Remove all completed tasks that could not yield
pendingTasks.RemoveAll(tt => tt == t ||
tt.IsCompletedSuccessfully && !tt.Result ||
tt.IsFaulted || tt.IsCanceled);
}
int lastIndex = 0;
if (pendingTasks.Count > 0)
{
lastIndex = Array.IndexOf(waitToReadTasks, pendingTasks[0]);
await pendingTasks[0];
}
var lastItem = await channelReaders[lastIndex].ReadAsync(cancellationToken);
return (lastItem, lastIndex);
}
}
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/494486.html
標籤：C＃ 去 异步 阻塞收集 system.threading.channels
上一篇：切換到C中的新呼叫堆疊？