问题标签 [concurrent-collections]

我有一个集合，它绑定到 WPF UI 中的数据网格。

我的要求就像我必须为集合中的每个项目每秒更新 10 次属性的值。

所以我采取了 ConcurrentBag 类型的集合。更新每个项目的值后。我正在明确触发 RaisePropertyChange。但用户界面没有改变。

可能重复：
ConcurrentBag 中可能存在内存泄漏？

我的应用程序中有史诗般的内存泄漏。我以一种方法添加到本地 concurrentBag 集合中的所有数据从未被收集。

这个简单的代码演示了我如何使用它：

我所期望的：将在“Func”方法中创建并处理包。

我所看到的：包永远不会处理，保存在 Parallel.ForEach 中创建的所有线程，保存我添加的所有数据。=(

好的，当我将它添加到列表中时，我可以使用“TryTake”从包中删除它。但空包仍然记忆犹新。

现在我通过使用 List 而不是 ConcurrentBag 来解决问题。但是我睡不好觉，因为我在我的记忆档案中看到了这个。对不起我的英语 =)

更新

我改变了我的方法“Func”：

然后我在 VS 中创建项目，编译并运行我的程序。这个实例图是由“.Net Memory Profiler 4.0”从内存快照中创建的，我在程序完成所有工作后的 10 分钟内收集了它：

http://xmages.net/storage/10/1/0/f/d/upload/4c67f305.jpg （抱歉，无法发布图片）

有没有办法一次将多个项目添加到 ConcurrentBag，而不是一次一个？我在 ConcurrentBag 上没有看到 AddRange() 方法，但有一个 Concat()。但是，这对我不起作用：

此代码曾经位于 Parallel.ForEach() 中，但我将其更改为上述代码，以便对其进行故障排除。变量 newList 确实有对象，但是在 objectList.Concat<> 行之后，objectList 中总是有 0 个对象。Concat<> 不是这样工作的吗？我是否需要使用 Add() 方法一次向 ConcurrentBag 添加一个项目？

Microsoft 的并行编程白皮书描述了在各种 FLOPS 阈值下最佳的情况，并且 FLOPS 率是何时应使用特定实现的决策点。

如何在我的应用程序中测量 FLOPS？

如果有任何框架实现具有以下行为的集合，我很感兴趣。

假设它最初包含： [1, 2, 3]

• 我迭代它（使用迭代器）并到达元素 2，现在我将 4 添加到末尾（集合现在将是 [1, 2, 3, 4]）。
• 现在我创建一个新的迭代器并迭代集合，得到 [1, 2, 3, 4]
• 我继续使用第一个迭代器进行迭代，它只会给我 3 并返回
• 现在重置第一个迭代器会给我 [1, 2, 3, 4] （类似于创建一个新的）。

同样应该适用于删除元素。如果我删除 3 而不是添加，第二个迭代器应该给我 [1, 2] 而第一个迭代器仍然会给我 3 和结尾。

所以当我得到和迭代器时，我希望它给我创建迭代器时拥有的集合（即使我稍后迭代它，我会迭代一点并稍后继续），当我重置迭代器时，它会被垃圾收集将更新到最新版本，我应该能够在不同时间创建多个迭代器实例，这些实例将根据创建迭代器时数组的内容提供不同的版本。

我需要它与多个线程一起工作，并且最好有一个有效的实现。

有谁知道这样一个集合的任何实现，还是我必须自己实现它？

我有以下代码：

我的问题是，在设置新任务时是否需要使用 Take 或 Try Take 方法，或者上述方法是否可行？我的意思是，我不需要从收藏中安全地删除我的项目，然后在任务安全完成后重新添加它。我可以为任何想回答我的问题的人提供的唯一保证是，没有两个任务将使用“Cache.TreeNodeItemsCollection”中的同一个节点。

我还没有添加一些代码来启动任务，但是一旦我知道我当前的设置是否正常，就会添加这些代码？

我有一个托管在 Windows 服务中的 WCF 服务。

WCF 服务被注入一个ConcurrentQueue<SomeClass>.

Windows 服务（托管 WCF 服务）也注入了ConcurrentQueue<SomeClass>.

Windows 服务启动 WCF 服务，并产生一个工作线程。

工作线程将 ( Enqueue) 对象推SomeClass送到ConccurrentQueue.

我希望以某种方式在对象入队时触发 WCF 服务，ConcurrentQueue以便它可以将其（以及可能在队列中的任何其他对象）出列，并向所有连接的客户端发送消息。

.

两个问题：

1. 如何使 WCF 服务“挂钩”注入的 ConccurentQueue 的“入队”方法，以便在将对象推入队列时 - WCF 服务会对此做些什么？

2. 如果我设法以某种方式挂钩“入队”事件，并且现在 Windows 服务中的工作线程已将两个对象入队 - 这意味着它还会“触发”两次 WCF 服务的“入队挂钩” - 那些两个触发事件发生在不同的线程中？我需要以某种方式确保 WCF 服务按顺序从队列中提取对象。我不想突然看到对于每个 Enqueue 操作 - WCF 服务在单独的线程中触发并将对象拉出......

我希望我的问题足够清楚......

.

[更新]

与 Chris 聊天后，我得出的结论是，最好的方法是将工作线程与 WCF 服务分开，并让工作线程作为客户端调用 WCF 服务。这意味着我不必摆弄队列。

很简单：除了 ConcurrentDictionary （如果我必须使用它，但它不是真正正确的概念）之外，是否有任何 Concurrent 集合（IProducerConsumer 实现）支持基于项目或谓词的简单相等来删除特定项目定义移除条件？

说明：我有一个多线程、多阶段的工作流算法，它从数据库中提取对象并将它们放在“开始”队列中。从那里他们被下一阶段抓住，进一步处理，并塞进其他队列。这个过程会持续几个阶段。同时，第一阶段由其主管再次调用并将对象从数据库中拉出，这些对象可能包括仍在处理中的对象（因为它们尚未完成处理，因此尚未使用标志集重新持久化）他们完成了）。

我正在设计的解决方案是“工作中”的大师系列；对象在第一阶段检索以进行处理时进入该队列，并在工作流的任何阶段完成必要处理后将它们重新保存到数据库作为“已处理”后将其删除。当对象在该列表中时，如果它被第一阶段重新检索，它将被忽略。

我曾计划使用 ConcurrentBag，但唯一的删除方法 (TryTake) 从包中删除任意项目，而不是指定的项目（并且 ConcurrentBag在 .NET 4 中很慢）。ConcurrentQueue 和 ConcurrentStack 也不允许删除它会给你的下一个项目以外的项目，留下 ConcurrentDictionary，它可以工作但超出我的需要（我真正需要的是存储正在处理的记录的 Id；它们在工作流程中不会改变）。

我目前正在编写一个 IRC 机器人。我想避免过多的洪水，所以我决定创建一个消息队列，每隔 X 毫秒发送下一条消息，但我的尝试失败了。第 43 行：

引发 OutOfMemory 异常。我完全不知道我做错了什么。

也许我还应该解释这种（可能很复杂）排队方式背后的一般想法。

首先，主要Hashtable queueht存储ConcurrentQueue<string>类型，其中消息的目标用作键。我希望机器人遍历哈希表，从每个队列发送一条消息（如果队列被清空，则删除密钥）。我想不出一个合适的方法来处理哈希表本身，所以我决定创建另一个队列，ConcurrentQueue<string> queue它会在清空队列时存储键和它们的使用顺序。

假设一个队列中有数百个项目的假设情况（这是可能的），任何新请求都会被天知道延迟多长时间（消息之间的内置延迟加上延迟），所以我有方法 Add() rebuild queue。我创建了一个深层副本queueht（或者我希望如此）并queue基于这个一次性副本生成一个新副本，并在此过程中摆脱它。

我认为我的思路和/或代码是非常错误的，因为我几乎没有线程经验，集合比简单数组和 OOP 习惯/约定更复杂。我非常感谢通过解释解决我的问题。提前致谢！

编辑：发布整个班级。

EDIT2：我知道这while (ht.Count > 0)将无限期执行。它只是以前版本的一部分，看起来像这样：

但是集合在评估时不能被修改（我发现很难），所以它不再那样了。我只是忘了更改条件while。

我冒昧地尝试了 TheThing 的解决方案。虽然它似乎实现了它的目的，但它并没有发送任何消息......这是它的最终形式：

我尝试切换到ConcurrentQueue，它应该也可以工作（尽管以更线程安全的方式，并不是说我对线程安全一无所知）。我也尝试切换到System.Threading.Timer，但这也无济于事。我很久以前就没有想法了。

编辑：作为一个彻头彻尾的白痴，我没有设置 Timer 启动的时间。将 bool 部分更改为更改计时器的到期时间和间隔的 Start() 方法使其工作。问题解决了。

我有一张经常阅读但很少写入的地图。有些操作（可以是读或写）涉及多个需要原子操作的对象，所以我使用了 ReadWriteLock 来提高性能。

现在我可以选择将并发映射降级为普通哈希映射，但我担心一些缓慢的迭代代码。

如果我降级地图，长迭代器必须持有读锁以避免并发访问异常。我认为这会阻塞写入线程太久。

由于一些迭代器对不一致的数据不敏感，我可以保留并发映射，以便迭代器可以与并发写入一起使用。但是，这会为正确使用锁的操作增加不必要的开销（来自并发映射）。

或者，我可以实现类似 Read-on-write 映射，其中克隆整个（非并发）映射以进行写入操作，以便现有迭代器继续在读锁之外工作。

显然，所有这些方法都是有效的，性能取决于实际的代码和设置。但是，我想知道是否对此有任何研究（所以我不必自己做实验）？