如何解决等待 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
方法使用它。
- 如果没有消费者,集合为空,并且调用了
CompleteAdding
方法,那么WaitCompleted
任务将永远不会完成。 - 如果有两个或更多消费者,除了一个消费者之外,其他所有消费者的枚举都将失败并返回
InvalidOperationException
。 - 如果有一个消费者,但使用
Take
或TryTake
方法消费集合,则WaitCompleted
任务永远不会完成。
在不知道您的具体用例的情况下,我无法确定您请求此功能是否有正当理由。但一般来说,等待 BlockingCollection<T>
变空并完成的确切时刻通常并不重要。重要的是处理所有消费项目完成的确切时刻,这发生在收集完成之后。
注意:此答案针对此问题的 Revision 1 版本。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。