标签: concurrent-collections

System.Collections.Queueclass具有Queue.Synchronized返回线程安全的Queue实现的方法.

但是通用的,System.Collections.Generic.Queue 没有Synchronized方法.在这一点上,我有两个问题:

1. 为什么通用的没有这种方法？它是一个框架API设计决策吗？
2. 返回的队列 Queue.Synchronized与ConcurrentQueue<T>类不同？

谢谢.

前言：我问这个只是因为我没有环境（数据集足够大+计算能力）以可靠的方式对其进行测试。

问题：给定一个并发包，加载了数十亿个项目，由单个线程访问/使用，它的性能是否类似于List？换句话说，对 a 的枚举是否比对 aConcurrent Bag更具有表演性List<T>？

我正在创建一个同时执行方法的接口,同时抽象出同步细节(在需要时交换分布式实现).我已经创建了一个单独的jvm实现,它允许将字符串作为互斥体存储在映射中,以确保使用一个引用,即使传入不同引用的字符串.并发似乎工作正常,但我是惊讶地看到测试表明参考计数从未减少.我假设使用WeakValues()就足以防止内存泄漏,但似乎并非如此.任何人都可以指出可能导致这种泄漏的原因吗？

public class SynchronousMethodExecutorSynchronizedImpl implements ISynchronousMethodExecutor {

// mutex map to provide string references
final Map<String, String> mutexMap = new MapMaker()
    .weakValues()
    .makeComputingMap(
        new Function<String, String>() {
        @Override
        public String apply(String id) {
            return id;
        }
    });

@Override
public Object doSynchronousMethod(String domain, String id, ISynchronousMethod synchronousMethod) {
    synchronized(mutexMap.get(domain + "." + id))
    {
        return synchronousMethod.execute();
    } 
}

}

这是在最后一个断言失败的测试:

public class SynchronousMethodExecutorSynchronizedImplTest extends TestCase {
int counter;
SynchronousMethodExecutorSynchronizedImpl methodExecutor;

@Override
public void before() throws Exception {
    super.before();

    methodExecutor = new …

如果有任何框架实现了具有以下行为的集合,我感兴趣.

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

• 我迭代它(使用迭代器)并到达元素2,现在我将4添加到结尾(集合现在将是[1,2,3,4]).
• 现在我创建一个新的迭代器并迭代集合,产生[1,2,3,4]
• 我继续迭代第一个迭代器,它只给我3并返回
• 现在重置第一个迭代器会给我[1,2,3,4](类似于创建一个新的迭代器).

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

所以,当我得到迭代器时,我想让它给我一个我创建迭代器时的集合(即使我稍后迭代它,我迭代一点然后继续),当我重置迭代器时,它会被垃圾收集它将更新到最新版本,我应该能够在不同的时间创建多个迭代器实例,根据创建迭代器时的数组内容,将提供不同的版本.

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

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

可能重复:
ConcurrentHashMap和Collections.synchronizedMap(Map)之间有什么区别？

我正在阅读HashMap,Collenctions.synchonizedMap和ConcurrentHashMap之间的差异.我的理解是Collections.synchronizedMap对整个集合应用了锁,因此性能开销.但ConcurrentHashMap不使用同步.它使用段来实现结果,因此它提供了与HashMap类似的性能.

如果我的理解是正确的,请建议.如果是这种情况,即使可能没有多个线程访问它,我可以在任何地方使用ConcurrentHashMap吗？

最近阅读了关于不可变集合的信息。当读取操作的执行频率高于写入操作时，建议将它们用作线程安全的读取操作。

然后我想测试读取性能ImmutableDictionary与ConcurrentDictionary. 这是这个非常简单的测试（在 .NET Core 2.1 中）：

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Diagnostics;
using System.Linq;
using System.Threading.Tasks;

namespace ImmutableSpeedTests
{
    class Program
    {
        public class ConcurrentVsImmutable
        {
            public int ValuesCount;
            public int ThreadsCount;

            private ImmutableDictionary<int, int> immutable = ImmutableDictionary<int, int>.Empty;
            private ConcurrentDictionary<int, int> concurrent = new ConcurrentDictionary<int, int>();

            public ConcurrentVsImmutable(int valuesCount, int threadsCount)
            {
                ValuesCount = valuesCount;
                ThreadsCount = threadsCount;
            }

            public void Setup()
            {
                // fill both collections. I don't measure …

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

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

Windows 服务（承载 WCF 服务）也注入了ConcurrentQueue<SomeClass>.

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

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

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

.

两个问题：

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

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

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

.

[更新]

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

因此,以下代码将每秒转储整个列表.

var list = new List<object>();

for (int i = 0; i < 100; i++)
{
    list.Add(new { A = i.ToString(), B = new Random().Next() });
    list.Dump(); // How to DumpLatest()?
    Thread.Sleep(1000);
}

但是,如何在不添加新转储输出的情况下更新转储输出呢？

这里有一个相关的问答,但它对我不起作用.

可能重复:
ConcurrentBag中可能的内存泄漏？

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

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

void Main()
{
    var l = new List<int>(){1,2,3,4};
    Func(l);
    l.Clear();
    l=null;
}

void Func(List<int> list)
{
    var bag = new ConcurrentBag<int>();
    Parallel.ForEach(list, k=> bag.Add(++k));

    list.Clear();

    foreach(int i in bag) //I know, I doing it wrong.
    {
        list.Add(i);
    }
}

我的期望:将创建包并在方法"Func"处置.

我所看到的:bag永远不会被释放,保存在Parallel.ForEach中创建的所有线程,保存我添加的所有数据.=(

好的,当我将它添加到列表中时,我可以使用"TryTake"从包中删除项目.但空袋仍然留在记忆中.

现在我通过使用List而不是ConcurrentBag来解决问题.但是因为我在记忆中看到这个,所以我睡不好觉.抱歉我的eng =)

UPDATE

我改变了我的方法"Func":

void Func(List<int> list)
{
    var bag = new ConcurrentBag<int>();
    Parallel.ForEach(list, k=> bag.Add(++k));

    list.Clear();

    int i;
    while(bag.TryTake(out i))
    {
        list.Add(i);
    }

    bag = null;
    GC.Collect();
    GC.WaitForPendingFinalizers();
}

然后我在VS中创建项目,编译并运行我的程序.这个实例图是由内存快照中的".Net Memory Profiler 4.0"创建的,我在程序完成所有工作后的10分钟内收集: …

当我打电话IProducerConsumerCollection<T>.TryAdd(<T>)或者IProducerConsumerCollection<T>.TryTake(out <T>)这些因为另一个线程正在使用该集合而失败？

或者,即使在另一个线程完成了集合之后,如果有空间要添加或者要采取的东西,它将始终返回true？

我在这里看不到任何东西:http://msdn.microsoft.com/en-us/library/dd287147.aspx