了解 BlockingCollection.TakeFromAny 的实现方法

怎么入门Golang编程？需要学习哪些知识点？这是新手们刚接触编程时常见的问题；下面golang学习网就来给大家整理分享一些知识点，希望能够给初学者一些帮助。本篇文章就来介绍《了解 BlockingCollection.TakeFromAny 的实现方法》，涉及到，有需要的可以收藏一下

我正在尝试实现一个异步方法，该方法采用 channelreaders 数组，并从任何具有可用项目的通道中获取值。它是与 blockingcollection.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 方法应异步读取项目，并返回消耗的项目以及 channelreaders 数组中关联通道的索引。如果所有通道都已完成（成功或有错误），或者在 await 期间全部完成，则该方法应异步抛出 channelclosedexception。

我的问题是：如何实现 takefromanyasync 方法？实现看起来相当棘手。显然，在任何情况下，该方法都不应从渠道消耗多个项目。此外，它不应该留下“一劳永逸”的任务，或者让可处置的资源得不到处理。该方法通常会在循环中调用，因此它也应该相当高效。它的复杂度应该不低于 o(n)，其中 n 表示通道数。

为了了解此方法的用处，您可以查看 go 语言的 select 语句。来自游览：

select 语句让 goroutine 等待多个通信操作。 select 会阻塞，直到其中一个案例可以运行，然后执行该案例。如果准备好多个，它会随机选择一个。

select {
case msg1 := <-c1:
    fmt.Println("received", msg1)
case msg2 := <-c2:
    fmt.Println("received", msg2)
}

在上面的示例中，要么从通道 c1 获取一个值并将其分配给变量 msg1，要么从通道 c2 获取一个值并将其分配给变量 msg2。 go select 语句不限于从通道读取。它可以包括多种异构情况，例如写入有界通道、等待计时器等。复制 go select 语句的完整功能超出了本问题的范围。

正确答案

我想出了这样的东西：

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);
    }
}

这是另一种方法。此实现在概念上与 alexm 的 implementation 相同，直到没有通道立即有可用项目为止。然后，它的不同之处在于避免了 task.whenany-in-a-loop 模式，而是为每个通道启动一个异步循环。所有循环都竞相更新共享的 ValueTuple consumed 变量，该变量在关键区域中更新，以防止消耗来自多个通道的元素。

/// <summary>
/// Takes an item asynchronously from any one of the specified channel readers.
/// </summary>
public static async Task<(T Item, int Index)> TakeFromAnyAsync<T>(
    ChannelReader<T>[] channelReaders,
    CancellationToken cancellationToken = default)
{
    ArgumentNullException.ThrowIfNull(channelReaders);
    if (channelReaders.Length == 0) throw new ArgumentException(
        $"The {nameof(channelReaders)} argument is a zero-length array.");
    foreach (var cr in channelReaders) if (cr is null) throw new ArgumentException(
        $"The {nameof(channelReaders)} argument contains at least one null element.");

    cancellationToken.ThrowIfCancellationRequested();

    // Fast path (at least one channel has an item available immediately)
    for (int i = 0; i < channelReaders.Length; i++)
        if (channelReaders[i].TryRead(out var item))
            return (item, i);

    // Slow path (all channels are currently empty)
    using var linkedCts = CancellationTokenSource
        .CreateLinkedTokenSource(cancellationToken);

    (T Item, int Index, bool HasValue) consumed = default;

    Task[] tasks = channelReaders.Select(async (channelReader, index) =>
    {
        while (true)
        {
            try
            {
                if (!await channelReader.WaitToReadAsync(linkedCts.Token)
                    .ConfigureAwait(false)) break;
            }
            // Only the exceptional cases below are normal.
            catch (OperationCanceledException)
                when (linkedCts.IsCancellationRequested) { break; }
            catch when (channelReader.Completion.IsCompleted
                && !channelReader.Completion.IsCompletedSuccessfully) { break; }

            // This channel has an item available now.
            lock (linkedCts)
            {
                if (consumed.HasValue)
                    return; // An item has already been consumed from another channel.

                if (!channelReader.TryRead(out var item))
                    continue; // We lost the race to consume the available item.

                consumed = (item, index, true); // We consumed an item successfully.
            }
            linkedCts.Cancel(); // Cancel the other tasks.
            return;
        }
    }).ToArray();

    // The tasks should never fail. If a task ever fails, we have a bug.
    try { foreach (var task in tasks) await task.ConfigureAwait(false); }
    catch (Exception ex) { Debug.Fail("Unexpected error", ex.ToString()); throw; }

    if (consumed.HasValue)
        return (consumed.Item, consumed.Index);
    cancellationToken.ThrowIfCancellationRequested();
    Debug.Assert(channelReaders.All(cr => cr.Completion.IsCompleted));
    throw new ChannelClosedException();
}

应该注意的是，这个解决方案以及 alexm 的解决方案都依赖于在元素被消耗时取消所有挂起的 WaitToReadAsync 操作。不幸的是，这会触发臭名昭著的 memory leak issue，它会影响具有空闲生产者的 .net 通道。当取消通道上的任何异步操作时，取消的操作将保留在内存中，附加到通道的内部结构，直到将元素写入通道。 microsoft 的这种行为 has been triaged 是设计的，尽管尚未排除对其进行改进的可能性。有趣的是，这种模糊性导致了这种效果 not eligible for documentation。因此，了解这一点的唯一方法是偶然，要么从非官方来源阅读它，要么陷入其中。

终于介绍完啦！小伙伴们，这篇关于《了解 BlockingCollection.TakeFromAny 的实现方法》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！