.net - .NET 的 ConcurrentBag 之类的类如何？被执行？

Question

ConcurrentBag<T>我发现自己对即将到来的 .NET 4.0 框架中存在一个类非常感兴趣：

当订购无关紧要时，包对于存储物品很有用，而且与集合不同，包支持重复。

我的问题是：如何实现这个想法？我熟悉的大多数集合基本上相当于（在引擎盖下）某种形式的数组，其中的顺序可能并不“重要”，但有一个顺序（这就是为什么，即使它不需要，枚举也会几乎总是以相同的顺序遍历未更改的集合，无论是List、Queue、Stack等）。

如果我不得不猜测，我可能会建议在内部它可能是Dictionary<T, LinkedList<T>>; 但考虑到只使用任何类型T作为键是没有意义的，这实际上似乎很可疑。

我期待/希望的是，这实际上是一种已经在某处“弄清楚”的既定对象类型，知道这种既定类型的人可以告诉我。这对我来说太不寻常了——这些概念在现实生活中很容易理解，但作为开发人员很难转化为可用的类——这就是为什么我对这些可能性感到好奇的原因。

编辑：

一些响应者建议 aBag可能是内部哈希表的一种形式。这也是我最初的想法，但我预见到这个想法有两个问题：

1. 当您没有适合所讨论类型的哈希码函数时，哈希表并不是那么有用。
2. 简单地跟踪集合中对象的“计数”与存储对象不同。

正如 Meta-Knight 所建议的，也许举个例子可以更清楚地说明这一点：

public class ExpensiveObject() {
    private ExpensiveObject() {
        // very intense operations happening in here
    }

    public ExpensiveObject CreateExpensiveObject() {
        return new ExpensiveObject();
    }
}

static void Main() {
    var expensiveObjects = new ConcurrentBag<ExpensiveObject>();

    for (int i = 0; i < 5; i++) {
        expensiveObjects.Add(ExpensiveObject.CreateExpensiveObject());
    }

    // after this point in the code, I want to believe I have 5 new
    // expensive objects in my collection

    while (expensiveObjects.Count > 0) {
        ExpensiveObject expObj = null;
        bool objectTaken = expensiveObjects.TryTake(out expObj);
        if (objectTaken) {
            // here I THINK I am queueing a particular operation to be
            // executed on 5 separate threads for 5 separate objects,
            // but if ConcurrentBag is a hashtable then I've just received
            // the object 5 times and so I am working on the same object
            // from 5 threads at the same time!
            ThreadPool.QueueUserWorkItem(DoWorkOnExpensiveObject, expObj);
        } else {
            break;
        }
    }
}

static void DoWorkOnExpensiveObject(object obj) {
    ExpensiveObject expObj = obj as ExpensiveObject;
    if (expObj != null) {
        // some work to be done
    }
}

Answer 1

如果您查看 的详细信息ConcurrentBag<T>，您会发现它在内部基本上是一个自定义的链表。

由于 Bags 可以包含重复项，并且不能通过索引访问，因此双向链表是一个非常好的实现选择。这允许锁定非常细粒度以进行插入和删除（您不必锁定整个集合，只需锁定您插入/删除的节点）。由于您不担心重复，因此不涉及散列。这使得双链表变得完美。

Answer 2

这里有一些关于 ConcurrentBag 的好信息：http: //geekswithblogs.net/BlackRabbitCoder/archive/2011/03/03/c.net-little-wonders-concurrentbag-and-blockingcollection.aspx

ConcurrentBag 的工作方式是利用新的 ThreadLocal 类型（.NET 4.0 的 System.Threading 中的新类型），以便使用包的每个线程都有一个仅针对该线程的本地列表。

这意味着添加或删除线程本地列表需要非常低的同步。问题在于一个线程去消费一个项目，但它的本地列表是空的。在这种情况下，包执行“工作窃取”，它将从另一个线程中窃取一个项目，该线程在其列表中有项目。这需要更高级别的同步，这会为 take 操作增加一些开销。

Answer 3

由于排序无关紧要，因此 ConcurrentBag 可以在幕后使用哈希表来快速检索数据。但与Hashset不同的是，bag 接受重复项。也许每个项目都可以与 Count 属性配对，该属性在添加项目时设置为 1。如果您第二次添加相同的项目，您可以只增加该项目的 Count 属性。

然后，要删除计数大于 1 的项目，您只需减少该项目的计数即可。如果计数为 1，您将从哈希表中删除 Item-Count 对。

Answer 4

好吧，在 smalltalk 中（Bag 的概念来源于此），集合基本上与哈希相同，尽管它允许重复。但是，它不是存储重复的对象，而是维护一个“出现计数”，例如每个对象的引用计数。如果 ConcurrentBag 是一个忠实的实现，这应该给你一个起点。

Answer 5

我相信“包”的概念是“Multiset”的同义词。

如果您对它们的实现方式感兴趣，有许多“Bag”/“Multiset”实现（这些恰好是 java）是开源的。

这些实现表明，可以根据您的需要以多种方式实现“包”。有 TreeMultiset、HashMultiset、LinkedHashMultiset、ConcurrentHashMultiset 的例子。

Google Collections
Google 有许多“MultiSet”实现，其中一个是 ConcurrentHashMultiset。

Apache Commons
Apache 有许多“Bag”实现。

Answer 6

System.Collections.Concurrent 命名空间现在是开源的，现在可以在这里找到 ConcurrentBag 的实现：

https://github.com/dotnet/runtime/blob/main/src/libraries/System.Collections.Concurrent/src/System/Collections/Concurrent/ConcurrentBag.cs

以下是截至 2022 年 1 月 30 日的实施。它是 MIT 许可的。

    // Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.

using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;

namespace System.Collections.ObjectModel
{
    [Serializable]
    [DebuggerTypeProxy(typeof(CollectionDebugView<>))]
    [DebuggerDisplay("Count = {Count}")]
    [System.Runtime.CompilerServices.TypeForwardedFrom("mscorlib, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089")]
    public abstract class KeyedCollection<TKey, TItem> : Collection<TItem> where TKey: notnull
    {
        private const int DefaultThreshold = 0;

        private readonly IEqualityComparer<TKey> comparer; // Do not rename (binary serialization)
        private Dictionary<TKey, TItem>? dict; // Do not rename (binary serialization)
        private int keyCount; // Do not rename (binary serialization)
        private readonly int threshold; // Do not rename (binary serialization)

        protected KeyedCollection() : this(null, DefaultThreshold)
        {
        }

        protected KeyedCollection(IEqualityComparer<TKey>? comparer) : this(comparer, DefaultThreshold)
        {
        }

        protected KeyedCollection(IEqualityComparer<TKey>? comparer, int dictionaryCreationThreshold)
            : base(new List<TItem>()) // Be explicit about the use of List<T> so we can foreach over
                                      // Items internally without enumerator allocations.
        {
            if (dictionaryCreationThreshold < -1)
            {
                throw new ArgumentOutOfRangeException(nameof(dictionaryCreationThreshold), SR.ArgumentOutOfRange_InvalidThreshold);
            }

            this.comparer = comparer ?? EqualityComparer<TKey>.Default;
            threshold = dictionaryCreationThreshold == -1 ? int.MaxValue : dictionaryCreationThreshold;
        }

        /// <summary>
        /// Enables the use of foreach internally without allocations using <see cref="List{T}"/>'s struct enumerator.
        /// </summary>
        private new List<TItem> Items
        {
            get
            {
                Debug.Assert(base.Items is List<TItem>);
                return (List<TItem>)base.Items;
            }
        }

        public IEqualityComparer<TKey> Comparer => comparer;

        public TItem this[TKey key]
        {
            get
            {
                TItem item;
                if (TryGetValue(key, out item!))
                {
                    return item;
                }

                throw new KeyNotFoundException(SR.Format(SR.Arg_KeyNotFoundWithKey, key.ToString()));
            }
        }

        public bool Contains(TKey key)
        {
            if (key == null)
            {
                throw new ArgumentNullException(nameof(key));
            }

            if (dict != null)
            {
                return dict.ContainsKey(key);
            }

            foreach (TItem item in Items)
            {
                if (comparer.Equals(GetKeyForItem(item), key))
                {
                    return true;
                }
            }

            return false;
        }

        public bool TryGetValue(TKey key, [MaybeNullWhen(false)] out TItem item)
        {
            if (key == null)
            {
                throw new ArgumentNullException(nameof(key));
            }

            if (dict != null)
            {
                return dict.TryGetValue(key, out item!);
            }

            foreach (TItem itemInItems in Items)
            {
                TKey keyInItems = GetKeyForItem(itemInItems);
                if (keyInItems != null && comparer.Equals(key, keyInItems))
                {
                    item = itemInItems;
                    return true;
                }
            }

            item = default;
            return false;
        }

        private bool ContainsItem(TItem item)
        {
            TKey key;
            if ((dict == null) || ((key = GetKeyForItem(item)) == null))
            {
                return Items.Contains(item);
            }

            TItem itemInDict;
            if (dict.TryGetValue(key, out itemInDict!))
            {
                return EqualityComparer<TItem>.Default.Equals(itemInDict, item);
            }

            return false;
        }

        public bool Remove(TKey key)
        {
            if (key == null)
            {
                throw new ArgumentNullException(nameof(key));
            }

            if (dict != null)
            {
                TItem item;
                return dict.TryGetValue(key, out item!) && Remove(item);
            }

            for (int i = 0; i < Items.Count; i++)
            {
                if (comparer.Equals(GetKeyForItem(Items[i]), key))
                {
                    RemoveItem(i);
                    return true;
                }
            }

            return false;
        }

        protected IDictionary<TKey, TItem>? Dictionary => dict;

        protected void ChangeItemKey(TItem item, TKey newKey)
        {
            if (!ContainsItem(item))
            {
                throw new ArgumentException(SR.Argument_ItemNotExist, nameof(item));
            }

            TKey oldKey = GetKeyForItem(item);
            if (!comparer.Equals(oldKey, newKey))
            {
                if (newKey != null)
                {
                    AddKey(newKey, item);
                }
                if (oldKey != null)
                {
                    RemoveKey(oldKey);
                }
            }
        }

        protected override void ClearItems()
        {
            base.ClearItems();
            dict?.Clear();
            keyCount = 0;
        }

        protected abstract TKey GetKeyForItem(TItem item);

        protected override void InsertItem(int index, TItem item)
        {
            TKey key = GetKeyForItem(item);
            if (key != null)
            {
                AddKey(key, item);
            }

            base.InsertItem(index, item);
        }

        protected override void RemoveItem(int index)
        {
            TKey key = GetKeyForItem(Items[index]);
            if (key != null)
            {
                RemoveKey(key);
            }

            base.RemoveItem(index);
        }

        protected override void SetItem(int index, TItem item)
        {
            TKey newKey = GetKeyForItem(item);
            TKey oldKey = GetKeyForItem(Items[index]);

            if (comparer.Equals(oldKey, newKey))
            {
                if (newKey != null && dict != null)
                {
                    dict[newKey] = item;
                }
            }
            else
            {
                if (newKey != null)
                {
                    AddKey(newKey, item);
                }

                if (oldKey != null)
                {
                    RemoveKey(oldKey);
                }
            }

            base.SetItem(index, item);
        }

        private void AddKey(TKey key, TItem item)
        {
            if (dict != null)
            {
                dict.Add(key, item);
            }
            else if (keyCount == threshold)
            {
                CreateDictionary();
                dict!.Add(key, item);
            }
            else
            {
                if (Contains(key))
                {
                    throw new ArgumentException(SR.Format(SR.Argument_AddingDuplicate, key), nameof(key));
                }

                keyCount++;
            }
        }

        private void CreateDictionary()
        {
            dict = new Dictionary<TKey, TItem>(comparer);
            foreach (TItem item in Items)
            {
                TKey key = GetKeyForItem(item);
                if (key != null)
                {
                    dict.Add(key, item);
                }
            }
        }

        private void RemoveKey(TKey key)
        {
            Debug.Assert(key != null, "key shouldn't be null!");
            if (dict != null)
            {
                dict.Remove(key);
            }
            else
            {
                keyCount--;
            }
        }
    }
}