等待 BlockingCollection 完成 [关闭]

Question

对于多线程应用程序，我想 await 直到 BlockingCollection 完成并为空（IsCompleted = true）。我实现了以下内容，这似乎有效。

由于它是多线程，我什至不相信自己的影子。这会是一个健壮的实现吗？

public class BlockingCollectionEx<T> : BlockingCollection<T>
{
    public Task WaitCompleted => completedManualResetEvent.Task;
    private readonly TaskCompletionSource completedManualResetEvent = new();

    public new void CompleteAdding()
    {
        base.CompleteAdding();

        lock (completedManualResetEvent)
        {
            if (base.Count == 0 && !completedManualResetEvent.Task.IsCompleted)
                completedManualResetEvent.SetResult();
        }
    }

    public new IEnumerable<T> GetConsumingEnumerable()
    {
        foreach (var item in base.GetConsumingEnumerable())
            yield return item;

        lock (completedManualResetEvent) //if GetConsumingEnumerable is used by multiple threads, the 2nd one would throw an InvalidOperationException 
        {
            if (!completedManualResetEvent.Task.IsCompleted)
                completedManualResetEvent.SetResult();
        }
    }
    public new IEnumerable<T> GetConsumingEnumerable(CancellationToken cancellationToken) => throw new NotImplementedException();

    public new T Take() => throw new NotImplementedException();
    public new T Take(CancellationToken cancellationToken) => throw new NotImplementedException();

    public new bool TryTake([MaybeNullWhen(false)] out T item) => throw new NotImplementedException();
    public new bool TryTake([MaybeNullWhen(false)] out T item, int millisecondsTimeout) => throw new NotImplementedException();
    public new bool TryTake([MaybeNullWhen(false)] out T item, int millisecondsTimeout, CancellationToken cancellationToken) => throw new NotImplementedException();
    public new bool TryTake([MaybeNullWhen(false)] out T item, TimeSpan timeout) => throw new NotImplementedException();
}

用法：

var x = new BlockingCollectionEx<int> { 1, 2, 3 };
x.CompleteAdding();

Task.Run(() =>
{
    foreach (var item in x.GetConsumingEnumerable())
        // do stuff in Task 1
});
Task.Run(() =>
{
    foreach (var item in x.GetConsumingEnumerable())
        // do stuff in Task 2
});

await x.WaitCompleted;
Debug.Assert(x.IsCompleted);
// do stuff since the collection is emtpy

Answer 1

您的实现对于一般用途来说并不健壮，但对于遵守以下合同的应用程序来说可能已经足够了：

该集合必须仅由一位消费者使用，该消费者仅使用GetConsumingEnumerable 方法来使用它。

1. 如果没有消费者，集合为空，调用CompleteAdding方法，WaitCompleted任务永远不会完成。
2. 如果有两个或更多消费者，则枚举将失败，除一个消费者之外的所有消费者都将返回 InvalidOperationException。
3. 如果有一个消费者，但使用Take 或TryTake 方法消费集合，WaitCompleted 任务将永远无法完成。

在不了解您的具体用例的情况下，我无法说您是否有正当理由请求此功能。但总的来说，等待BlockingCollection<T> 变为空并完成的确切时刻通常并不重要。重要的是所有消费项目的处理完成的确切时刻，这发生在收集完成之后。

---

注意：此答案针对此问题的Revision 1 版本。