等待 BlockingCollection 完成

问题描述

对于多线程应用程序，我想 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

解决方法

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

该集合只能由一个消费者使用，该消费者仅使用 GetConsumingEnumerable 方法使用它。

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

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

---

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