标签: concurrent-collections

我试图使用Dictionary重新编写一些代码来使用ConcurrentDictionary.我已经回顾了一些示例,但我仍然无法实现AddOrUpdate函数.这是原始代码:

    dynamic a = HttpContext;
    Dictionary<int, string> userDic = this.HttpContext.Application["UserSessionList"] as Dictionary<int, String>;

   if (userDic != null)
   {
      if (useDic.ContainsKey(authUser.UserId))
      {
        userDic.Remove(authUser.UserId);
      }
   }
  else
  {
     userDic = new Dictionary<int,string>();
  }
  userDic.Add(authUser.UserId, a.Session.SessionID.ToString());
  this.HttpContext.Application["UserDic"] = userDic;

我不知道要为更新部分添加什么:

userDic.AddOrUpdate(authUser.UserId,
                    a.Session.SessionID.ToString(),
                    /*** what to add here? ***/);

任何指针将不胜感激.

除了Concurrent Collections是命名空间还是一个类之外,命名空间中SynchronizedCollection<T>的并发集合如何System.Collections.Concurrent相互不同SynchronizedCollection<T>？

SynchronizedCollection<T>并发集合中的所有类都提供了线程安全的集合.我如何决定何时使用其中一个,为什么？

在我开始一个项目之前,我编写了一个简单的测试来比较来自(System.Collections.Concurrent)的ConcurrentBag相对于锁定和列表的性能.我非常惊讶ConcurrentBag比使用简单的List锁定慢10倍.据我所知,当读写器是同一个线程时,ConcurrentBag效果最好.但是,我没想到它的性能会比传统的锁更糟糕.

我已经运行了一个测试,其中有两个Parallel for循环写入和读取列表/包.然而,写入本身显示了巨大的差异:

private static void ConcurrentBagTest()
   {
        int collSize = 10000000;
        Stopwatch stopWatch = new Stopwatch();
        ConcurrentBag<int> bag1 = new ConcurrentBag<int>();

        stopWatch.Start();


        Parallel.For(0, collSize, delegate(int i)
        {
            bag1.Add(i);
        });


        stopWatch.Stop();
        Console.WriteLine("Elapsed Time = {0}", 
                          stopWatch.Elapsed.TotalSeconds);
 }

在我的盒子上,这需要3-4秒才能运行,相比之下这段代码的0.5-0.9秒:

       private static void LockCollTest()
       {
        int collSize = 10000000;
        object list1_lock=new object();
        List<int> lst1 = new List<int>(collSize);

        Stopwatch stopWatch = new Stopwatch();
        stopWatch.Start();


        Parallel.For(0, collSize, delegate(int i)
            {
                lock(list1_lock)
                {
                    lst1.Add(i);
                }
            });

        stopWatch.Stop();
        Console.WriteLine("Elapsed = {0}", 
                          stopWatch.Elapsed.TotalSeconds);
       }

正如我所提到的,进行并发读写并不能帮助并发包测试.我做错了什么还是这个数据结构真的很慢？

[编辑] - …

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

ConcurrentBag<T> objectList = new ConcurrentBag<T>();

timeChunks.ForEach(timeChunk =>
{
    List<T> newList = Foo.SomeMethod<T>(x => x.SomeReadTime > timeChunk.StartTime);
    objectList.Concat<T>(newList);
});

这段代码曾经在Parallel.ForEach()中,但我把它改成了上面所以我可以解决它.变量newList确实有对象,但是在objectList.Concat <>行之后,objectList总是有0个对象.Concat <>不能那样工作吗？我是否需要使用Add()方法一次一个地向ConcurrentBag添加项目？

正如MSDN所说

ConcurrentDictionary<TKey, TValue> Class表示可以由多个线程同时访问的键值对的线程安全集合.

但据我所知,System.Collections.Concurrent课程是为PLINQ设计的.

我有Dictionary<Key,Value>保持服务器中的在线客户端,并且当我有权访问它时通过锁定对象使其成为线程安全的.

我可以放心地取代Dictionary<TKey,TValue>通过ConcurrentDictionary<TKey,TValue>我的情况？更换后性能会提高吗？

这里在第5约瑟夫阿尔巴哈利提到,它专为并行编程

• 并发集合针对并行编程进行了调整.除了高度并发的场景外,传统的集合在所有集合中都优于它们.
• 线程安全的集合不保证使用它的代码是线程安全的.
• 如果在另一个线程正在修改它时枚举并发集合,则不会抛出任何异常.相反,你会得到新旧内容的混合.
• List没有并发版本.
• 并发堆栈,队列和包类在内部使用链接列表实现.这使得它们比非并发Stack和Queue类的内存效率更低,但对并发访问更好,因为链表有利于无锁或低锁实现.(这是因为将节点插入链表需要更新几个引用,而将元素插入类似List的结构可能需要移动数千个现有元素.)

基本上,如果我想做以下事情:

public class SomeClass
{
    private static ConcurrentDictionary<..., ...> Cache { get; set; }
}

这是否让我避免lock在整个地方使用s？

非常简单:除了ConcurrentDictionary(我将使用它,但它不是真正正确的概念),是否有任何Concurrent集合(IProducerConsumer实现)支持基于项或谓词的简单相等删除特定项定义删除条件？

说明:我有一个多线程,多阶段的工作流算法,它从数据库中提取对象并将它们放在"起始"队列中.从那里他们被下一阶段抓住,进一步努力,并塞进其他队列.这个过程将持续几个阶段.同时,第一个阶段由其主管再次调用,并将对象拉出数据库,这些对象可以包括仍处于进行中的对象(因为它们尚未完成处理,因此没有使用标记集重新保留他们已经完成了).

我正在设计的解决方案是一个主要的"工作"集合; 当第一阶段检索到对象进行处理时,对象进入该队列,并且在完成必要处理的工作流程的任何阶段"处理"后,对象被重新保存到数据库后被删除.当对象在该列表中时,如果第一阶段重新检索它,它将被忽略.

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

我想弄清楚使用队列的最佳方法是什么.我有一个返回DataTable的进程.反过来,每个DataTable都与之前的DataTable合并.有一个问题,在最终的BulkCopy(OutOfMemory)之前要保留的记录太多.

所以,我已经确定我应该立即处理每个传入的DataTable.考虑一下ConcurrentQueue<T>......但我不知道该WriteQueuedData()方法如何知道将一个表出列并将其写入数据库.

例如:

public class TableTransporter
{
    private ConcurrentQueue<DataTable> tableQueue = new ConcurrentQueue<DataTable>();

    public TableTransporter()
    {
        tableQueue.OnItemQueued += new EventHandler(WriteQueuedData);   // no events available
    }

    public void ExtractData()
    {
        DataTable table;

        // perform data extraction
        tableQueue.Enqueue(table);
    }

    private void WriteQueuedData(object sender, EventArgs e)
    {
        BulkCopy(e.Table);
    }
}

我的第一个问题是,除了我实际上没有订阅任何事件的事实,如果我ExtractData()异步调用这将是我需要的全部内容吗？第二,我是否缺少关于ConcurrentQueue<T>函数的方式以及需要某种形式的触发器与排队对象异步工作的东西？

更新 我刚刚派生出一个ConcurrentQueue<T>具有OnItemQueued事件处理程序的类.然后:

new public void Enqueue (DataTable Table)
{
    base.Enqueue(Table);
    OnTableQueued(new TableQueuedEventArgs(Table));
}

public void OnTableQueued(TableQueuedEventArgs table)
{
    EventHandler<TableQueuedEventArgs> handler = TableQueued; …

我正在制作一个应用程序,它需要一堆日记帐分录并计算总和.

当有多个线程调用该addToSum()方法时,下面的方法是线程/并发安全.我想确保每次通话都能正确更新总数.

如果不安全,请说明我必须做些什么来确保螺纹安全.

我需要synchronize获取/放置还是有更好的方法？

private ConcurrentHashMap<String, BigDecimal> sumByAccount;

public void addToSum(String account, BigDecimal amount){
    BigDecimal newSum = sumByAccount.get(account).add(amount);
    sumByAccount.put(account, newSum);
}

非常感谢!

更新:

谢谢大家的答案,我已经知道上面的代码不是线程安全的.

感谢Vint建议AtomicReference作为替代品synchronize.之前我AtomicInteger用来保存整数和,我想知道BigDecimal是否有类似的东西.

关于两者的赞成和反对,这是一个明确的结论吗？

我试图概述C#中集合背后的线程安全理论.

为什么Java中没有并发集合？(java docs).一些集合看起来是线程安全的,但我不清楚这个位置是什么,例如关于:

• 复合作业,
• 使用迭代器的安全性,
• 写操作

我不想重新发明轮子!(我不是一个多线程大师,绝对不会低估这有多难).

我希望社区可以提供帮助.