问题标签 [blockingcollection]

我尝试写一些东西来散列数字，并根据列表检查它们是否存在匹配的散列。

我使用 for 循环让这个工作正常，然后我决定尝试使用 Parallel.For 来加快速度 - 不幸的是，这会导致 ArgumentOutOfRangeException，我无法调试。

调用 encrypter.EncryptCardNumber 时随机发生错误（似乎在 returnValue.ToString() 上）

我有两个问题：

1. 为什么我得到一个例外？
2. 我是否正确使用 BlockingCollection 来实现我想要实现的目标？

原来的

如何利用 TPL 和/或并发集合在 C# 中创建生产者-消费者队列？我正在使用 .NET 4.5+。

这是我的第一次尝试：

2014 年 9 月 5 日编辑：基于反馈的版本 2。

编辑 2014 年 9 月 7 日：问题已解决。

我发现了Microsoft Task Parallel Library Dataflow已经封装了我想要的东西的类以及更多。NuGet 包：Install-Package Microsoft.Tpl.Dataflow

为了使社区受益，我将分享一些测试代码。（也在https://dotnetfiddle.net/WbwUqz）

我必须分批执行 1000 条消息，例如操作 A、B、C。我可以并行执行这些操作。我为他们创建了群组。为了增加并行性，我在每个组中创建了子组。子组中的任务需要串行执行。但是两个子组可以并行执行。一批 1000 完成后，我必须做一些处理，即保存在 db 中。但我无法理解，如何等待所有任务完成（我对在 1000 次任务结束时等待中间不感兴趣）。欢迎任何建议。

所以基本上我会为每个子组创建 OrderlyThreadPool。我是来自说源的接收消息，如果没有消息可用，它会阻塞。所以我的代码，看起来像

问题：有多个线程访问一个资源。我需要将它们的数量限制为一个常数MaxThreads。无法进入线程池的线程应该会收到错误消息。

解决方案：我开始BlockingCollection<string> pool在下面的算法中使用 a ，但我看到这BlockingCollection需要调用CompleteAdding，我不能这样做，因为我总是得到传入线程（为了调试目的，我在下面的示例中硬编码为 10 ），想想网络请求.

关于如何更好地实现这一目标的任何想法？

谢谢！

PS这里是多线程新手，如果您对阅读材料有任何建议，我将不胜感激。

LE：根据我得到的答案，我能够使用以下算法实现所需的行为：

我有一个使用ConcurrentDictionary.

在这个类中，有三个函数对 this 执行一些操作ConcurrentDictionnary。

每个函数由不同的线程调用。

• 第一个函数操作：dictionnary.Where、dictionnary.TryRemove
• 第二个函数操作：dictionnary.Where
• 第三个函数操作：dictionnary.TryAdd

最后一个功能在不同的时间阻塞。我必须在不阻塞的情况下将我的元素添加到字典中，但我该怎么做？

我正在尝试通过使用 BlockingCollection 来实现数据包生产者-消费者行为，但我无法理解其中的文档。据我所知，由于显然不能保证我可以删除任何东西，因此系统太慢而无法用于接收和处理数据包等性能关键的事情。我需要这个的原因是因为接收数据包的线程与处理它们的线程不同。我错过了什么，还是我需要采取另一种方法？

示例代码：

在我的应用程序中，我有几个线程用于处理 TCP 连接（一个用于读取，一个用于发送，一个用于处理新的传入连接）。每个线程为所有客户端处理给定类型的操作，因此假设它向TcpClient不同 IP 上的 5 个实例发送数据。我使用 aBlockingCollection作为缓冲区，因为我从发送线程访问它，但也从另一个生成要发送的数据的线程访问它。我在发送线程中运行的函数如下所示：

注意：BlockingCollection使用的是 type <object[]>。我的问题是，在某个流量点，缓冲区开始填满。我在缓冲区中设置了最多 500 条消息的限制，它很容易溢出。现在，如果我理解正确（不确定），TryTake它会尝试删除该项目，如果此时正在使用该集合，它会等待并重试。（注意：我也尝试将超时设置为 50 毫秒）。如果这是真的（如果不是，请有人纠正我并提出不同的原因），问题可能是集合在大多数时间TryTake被调用时都很忙。会是这样吗？如果是，如何解决？

至于集合的使用，通过生成数据的线程，在迭代 1-80 个项目的 foreach 中，大约每 2 秒访问一次集合。缓冲区开始出现大约 20 多个项目的问题，直到那时，它都很好。发件人线程现在只发送给一个客户端，以后会达到 15 个。所以在高峰期，这将是 80 个项目 x 15 个用户 = 大约每 2 秒 1200 次访问。任何建议都非常感谢，谢谢。

我有一个关于使用 BlockingCollection 和 Dictionary 编写代码的问题。

我的目标是读取一堆文本文件并以并行方式处理它们。处理后的数据将存储在 BlockingCollection 实例中，以便可以将这些处理后的数据写入文件。

我想使用 BlockingCollection 的原因是......

(1) 在 GenerateDataFiles() 做 CPU 密集型工作时节省时间，同时消费者 Task 可以做 IO 相关工作，并且

(2) 与我在将所有已处理数据写入文件之前将所有已处理数据存储在列表中的情况相比，减少内存使用量。

对于（2），如果我在将所有数据写入文件之前存储它们，内存消耗超过我的桌面可以承受的（因为它读取超过 30GB 的数据），所以我必须使用这种生产者 - 消费者的方法。

此外，我在 BlockingCollection 实例（或字典）中插入键值对时遇到问题。请指出进行数据插入的正确方法。

以下代码是我尝试解决问题的方法。因为我是 BlockingCollection 的新手，所以我可能在这方面犯了一些错误。请提出一些更改（和修改的代码），以便我解决问题。

I have a program in which i am getting bitmap images from Camera loading them in blocking collection and processing in a thread. I am calling SetImage from UI thread. It works for few seconds then i run into out of memory exception. Please advise

我最近需要一个IProducerConsumerCollection<T>实现，但我希望它在TryAdd达到一定容量时阻止并TryTake在它为空时阻止。我确信这BlockingCollection实际上是一个实现，IProducerConsumerCollection<T>但意识到事实并非如此。这是为什么？

哪个属性BlockingCollection不适合实现IProducerConsumerCollection接口？

我知道这BlockingCollection是一个包装器，IProducerConsumerCollection但无论如何我认为它本身应该实现相同的接口。