问题标签 [blockingcollection]

我想将文本打印到文件中，但我想确保不会通过写入磁盘来阻止主线程。

我使用 BlockingCollection 创建了以下方案。

我有一个包含 60 秒 TryTake 的无限循环。

谁能告诉我他们是否发现这种方法的效率有任何问题？我相信只要将新的 txt 添加到集合中，它就会写入磁盘，否则等待 60 秒。所以它在等待输入时没有旋转。

以前，我使用了非阻塞收集方法和 2 个队列，每隔几秒切换一次。一个队列有新的 txt，而我将另一个写入磁盘。这意味着在写入磁盘之前会有一些延迟，如果发生崩溃，我可能会丢失最后的数据。

利用：

我已经阅读了一些其他类似但不相同的链接，试图找到一些答案： 如何批量使用 BlockingCollection<T>

但是，（在上面的链接中）不使用 GetConsumingEnumerable 似乎很可疑。

当消费者（应该是单数）清空集合时，有效阻止生产者的正确方法是什么？

[我们想要进行批处理，因为每个批处理都会执行一个 Web 服务调用，如果每条消息/项目都需要自己的调用，这将是一个瓶颈。批处理消息/项目是解决此瓶颈的方法。]

理想情况下：

1) 接收消息

2) 新的生产者任务推入收集

3）当集合“满”（任意限制）时，阻塞所有生产者，新的消费者任务消耗所有集合，然后为生产者解除阻塞。

换句话说; 我希望（并行生产者）xor（单一消费者）随时对集合起作用。

似乎以前应该这样做过，但我似乎找不到专门以这种方式运行的代码片段。

谢谢你的帮助。

我在涉足多线程时遇到了一个绊脚石。我想我知道问题是什么，但无法确定如何解决它。但我可能错了。

总之，我有生产者和消费者线程。生产者线程将来自外部源的数据收集到数据表中，然后将它们放入集合中。消费者然后从集合中获取数据表。我将 BlockingCollection 用作公共静态集合，以便两个线程都可以访问它，这两个线程存在于两个不同的类中。我现在将展示代码的主要部分，然后解释什么是有效的，什么是无效的。

生产者线程：

消费者线程

所以我的测试表明，当生产者线程创建数据表时，它成功地将它们添加到集合中。我可以通过在 add 方法之前和之后使用 count 来看到这一点。计算每个表中的行数，我还可以确认添加的表与创建的表相同。此外，take 方法还成功删除了一个表，并且该表与输入的表相匹配。我通过计算集合中的表数和计算“采取”数据表中的行数都知道这一点。

我的问题是当我尝试运行 foreach 循环以打印出结果时。最初它工作并开始将数据打印到屏幕上，但随后抛出此错误：

我相信这似乎发生在将新数据表添加到集合中时。我相信这得到了错误的支持，该错误表明 .movenext() 遇到了错误，这意味着它正试图移动到集合中的下一个值？

但我真的不知道该怎么做，即使我是对的。我试过复制数据表，但仍然遇到和错误。我原以为一旦它创建了一个数据表并使用集合中的 take 方法复制了数据表，我就可以随意迭代它。我的直觉是数据表仍然指向集合中的数据表，当新数据表进入时它会被撞倒，它会引发错误。但这是完全的猜测，可能完全错误。任何人都可以帮忙吗？

如果您需要更多信息，我很乐意发布。

在四核系统中运行此程序将消耗 25% 的 CPU 功率。所以基本上有些东西正在全速运行。我将其缩小到消费者，但是负载不会在按下“x”时停止，这应该终止我的消费者。

我的代码

我的问题是：
1 、为什么CPU负载这么高？GetConsumingEnumerable() 不应该阻塞，因此根本不使用 CPU 时间吗？
2.为什么在 cts.Cancel() 上没有停止？

我有一个 tcp 监听器，它监听和写入来自服务器的数据。我用 aBlockingCollection来存储数据。这里我不知道文件什么时候结束。所以，我的文件流总是打开的。

我的部分代码是：

问题是我不能在循环中处理文件流，否则我必须为每一行创建文件流，循环可能永远不会结束。

我正在使用 a Blockingcollectionas aFIFO queue但我正在对文件进行很多操作，其中consumer可能很容易遇到文件锁，所以我所做的是创建一个简单的 try catch，其中消费者重新排队本身，但在很长一段时间FIFO queue内队列中的许多其他项目，这足以暂停，但在一个空的或非常短的情况下FIFO queue，这意味着consumer永久地锤击队列，重复出现可能仍将被文件锁定的自身。

IE

consumer busy-> 重新排队 -> consumer busy -> 重新排队（无限期）

BlockingCollection如果新消费者不到 10 秒，有没有办法让它不尝试运行它？即可能在队列中获取网络并继续，如果它的 createdDateTime 为空（第一次尝试的默认值）或> 10秒，则只接受下一个消费者？

我有一个棘手的情况，想知道是否有人可以对此事有所了解：

我有一个像这样调用的阻塞收集操作工作者

现在“自我”的东西是伪代码，这就是为什么，这就是（worker容器类在哪里）runQueueBlockingCollection

现在我想要发生的是 Action (() => {} 第一个片段中的匿名方法在队列完成后执行并且能够包含 MD5 总和。我知道这里存在线程安全问题，所以我知道我必须invoke回到textBox1所在的父线程，但目前我只是迷失了如何做到这一点（甚至可能吗？）

编辑

我通常不会“编辑”答案，但如果其他人遇到这个问题，这是我最终使用的代码，这要归功于usr. 注意BeginInvoke

另外，如果您想知道我为什么这样做，那只是为了让我可以“监视”文件夹中的任何文件更改，然后存储元数据更改，因为可以将整个文件夹拖入/拖出，因此需要排队机制在 FIFO 队列中，父文件夹也被散列，因此任何文件更改都需要冒泡，最后可能是文件在尝试散列时被锁定，在这种情况下，FIFO 队列可以等待

我有一个问题是从电话BlockingCollection<MyItem>中醒来需要太多时间。.Take()

场景是这样的：我有一个线程可以BlockingCollection非常快地推送数据（实际上在 XUnit 中我做了一个for循环。我有一个 3Task只是坐在.Take()通话中等待添加项目。从输出中我可以看到将近 200 个项目（最多 1 秒或更长时间）在第一次Task醒来并实际从BlockingCollection.

我有多个基于以管道方式组织的“缓冲区”，并且所有这些缓冲区都因动作BlockingCollection唤醒时间过长而受到影响。.Take()

我试过了.TryTake()，.GetConsumingEnumerable()结果相同。

在这个管道的末端，我有一个单线程的慢速函数，可以一个一个地处理项目，并且处理单个项目可能需要未知的时间。我只需要确保“项目”非常快地从“缓冲区”传递到“缓冲区”（一旦它被插入第一个“缓冲区”）

我只需要确保启动时间（.Take()或.TryTake()等）将在项目添加到集合时接近时间发生

我现在得到的是一个每 5000 毫秒触发一次的计时器：

着火时，它会设置处理文件的队列：

阻塞队列本身：

我所拥有的工作方式是，启动计时器->停止计时器->触发收集文件夹中所有文件的过程->处理所有文件->重新启动计时器。

我不禁认为有更好的方法来做我正在做的事情，特别是当将为帐户创建的队列数量为 200 - 400 时。

谢谢

这可能有一个明显的答案，但无论如何我都会问。

是否有等效于 C# 的阻塞集合的变量？我想要的是让我的所有线程都能够访问共享变量。它不是一个集合，只是一个共享变量，在每个线程使用它时都会对其值进行调整。我喜欢blockingcollection 的地方在于ques 和locks 是由C# 管理的，并且希望只有一个变量有类似的东西？

我可以使用公共静态变量并自己创建锁，但我认为我应该检查一下。

编辑：联锁是否有可能。