.net源码分析 - ConcurrentDictionary泛型

迷南。 2021-06-26 16:06 339阅读 0赞

继上篇`Dictionary`源码分析，上篇讲过的在这里不会再重复

`ConcurrentDictionary`源码地址：

[https://github.com/dotnet/corefx/blob/master/src/System.Collections.Concurrent/src/System/Collections/Concurrent/ConcurrentDictionary.cs][https_github.com_dotnet_corefx_blob_master_src_System.Collections.Concurrent_src_System_Collections_Concurrent_ConcurrentDictionary.cs]

## 前言 ##

`ConcurrentDictionary`一大特点是线程安全，在没有`ConcurrentDictionary`之前在多线程下用`Dictionary`，不管读写都要加个锁，不但麻烦，性能上也不是很好，因为在上篇分析中我们知道`Dictionary`内部是由多个`bucket`组成，不同`bucket`的操作即使在多线程下也可以不互相影响，如果一个锁把整个`Dictionary`都锁住实在有点浪费。

不过凡事都有两面性，给每个`Bucket`都加一个锁也不可取，`Bucket`的数量和`Dictionary`元素数量是一样的，而`Bucket`可能会有一部分是空的，而且访问`Dictionary`的线程如果数量不是太多也根本用上不这么多锁，想想即使有10个线程在不停的操作这个`Dictionary`，同时操作的最多也就10个，即使两两冲突访问同一个`Bucket`，5个锁就够了，当然这是最好的情况，最坏情况是这5个`bucket`用同一个锁。所以，要得到最好的结果需要尝试取一个最优解，而影响因素则是`bucket`数量和线程数量。我们想要的结果是锁够用但又不浪费。

微软得出的结果是默认的锁的数量是CPU核的个数，这个线程池默认的线程数量一样。随着`Dictionary`的扩容，锁的个数也可以跟着增加，这个可以在构造函数中自己指定。

下面看看`ConcurrentDictionary`里元素是做了怎样的封装。

private volatile Tables _tables; // 这不同于Dictionary的bucket 数组，而是整个封装起来，而且用volatile来保证读写时的原子性
     
    private sealed class Tables
    { 
        internal readonly Node[] _buckets; // bucket成了这样，也就是ConcurrentDictionary可以认为是一个bucket数组，每个Bucket里又由next来形成链表
        internal readonly object[] _locks; // 这个就是锁的数组了
        internal volatile int[] _countPerLock; // 这个是每个锁罩的元素个数
    
        internal Tables(Node[] buckets, object[] locks, int[] countPerLock)
        { 
            _buckets = buckets;
            _locks = locks;
            _countPerLock = countPerLock;
        }
    }
    
    //由Dictionary里的Entry改成Node，并且把next放到Node里
    private sealed class Node
    { 
        internal readonly TKey _key;
        internal TValue _value;
        internal volatile Node _next; //next由volatile修饰，确保不被优化且读写原子性
        internal readonly int _hashcode;
    
        internal Node(TKey key, TValue value, int hashcode, Node next)
        { 
            _key = key;
            _value = value;
            _next = next;
            _hashcode = hashcode;
        }
    }

## 里面的一些变量： ##

private readonly bool _growLockArray; // 是否在Dictionary扩容时也增加锁的数量
    
    private int _budget; // 单个锁罩的元素的最大个数
    
    private const int DefaultCapacity = 31;  //ConcurrentDictionary默认大小，和List,Dictionary不一样
    
    private const int MaxLockNumber = 1024;  //最大锁的个数，不过也可以在构造函数中弄个更大的，不般没必要

## 看看构造函数初始化做了些啥 ##

internal ConcurrentDictionary(int concurrencyLevel, int capacity, bool growLockArray, IEqualityComparer<TKey> comparer)
    { 
        if (concurrencyLevel < 1)
        { 
            throw new ArgumentOutOfRangeException(nameof(concurrencyLevel), SR.ConcurrentDictionary_ConcurrencyLevelMustBePositive);
        }
        if (capacity < 0)
        { 
            throw new ArgumentOutOfRangeException(nameof(capacity), SR.ConcurrentDictionary_CapacityMustNotBeNegative);
        }
        if (comparer == null) throw new ArgumentNullException(nameof(comparer));
    
        // The capacity should be at least as large as the concurrency level. Otherwise, we would have locks that don't guard
        // any buckets.
        if (capacity < concurrencyLevel) //concurrencyLevel就是锁的个数，容量小于锁的个数时，一部分锁就真是完全没用了
        { 
            capacity = concurrencyLevel; //所以容量至少要和锁的个数一样
        }
    
        object[] locks = new object[concurrencyLevel]; //初始化锁数组
        for (int i = 0; i < locks.Length; i++)
        { 
            locks[i] = new object();
        }
    
        int[] countPerLock = new int[locks.Length]; //初始化锁罩元素个数的数组
        Node[] buckets = new Node[capacity]; //初始化Node
        _tables = new Tables(buckets, locks, countPerLock); //初始化table
    
        _comparer = comparer;
        _growLockArray = growLockArray;  //这就是指定锁是否增长
        _budget = buckets.Length / locks.Length;  //锁最大罩【容量/锁个数】这么多元素（毕竟不是扛把子，罩不了太多），多了怎么办，扩地盘。。
    }

## 通过常用的函数看看是怎么做到多线程安全的 ##

TryGet

public bool TryGetValue(TKey key, out TValue value)
    { 
        if (key == null) ThrowKeyNullException();
        return TryGetValueInternal(key, _comparer.GetHashCode(key), out value);
    }
    
    private bool TryGetValueInternal(TKey key, int hashcode, out TValue value)
    { 
        Debug.Assert(_comparer.GetHashCode(key) == hashcode);
                
        //先用本地变量存一下，免得在另外一个线程扩容时变了
        Tables tables = _tables;
    
        //又是hashcode取余哈，不多说
        //int bucketNo = (hashcode & 0x7fffffff) % bucketCount;
        int bucketNo = GetBucket(hashcode, tables._buckets.Length);
    
        //这个用Valatile确保读了最新的Node
        Node n = Volatile.Read<Node>(ref tables._buckets[bucketNo]);
        //遍历bucket，真怀疑这些代码是几个人写的，风格都不一样
        while (n != null)
        { 
            //找到了
            if (hashcode == n._hashcode && _comparer.Equals(n._key, key))
            { 
                //返回true和value
                value = n._value;
                return true;
            }
            n = n._next;
        }
    
        value = default(TValue);
        return false;
    }

GetOrAdd

public TValue GetOrAdd(TKey key, Func<TKey, TValue> valueFactory)
    { 
        if (key == null) ThrowKeyNullException();
        if (valueFactory == null) throw new ArgumentNullException(nameof(valueFactory));
    
        int hashcode = _comparer.GetHashCode(key);
    
        TValue resultingValue;
        //先TryGet，没有的再TryAdd
        if (!TryGetValueInternal(key, hashcode, out resultingValue))
        { 
            TryAddInternal(key, hashcode, valueFactory(key), false, true, out resultingValue);
        }
        return resultingValue;
    }
    
    private bool TryAddInternal(TKey key, int hashcode, TValue value, bool updateIfExists, bool acquireLock, out TValue resultingValue)
    { 
        Debug.Assert(_comparer.GetHashCode(key) == hashcode);
    
        while (true)
        { 
            int bucketNo, lockNo;
    
            Tables tables = _tables;
            //GetBucketAndLockNo函数里面就是下面两句
            //bucketNo = (hashcode & 0x7fffffff) % bucketCount; 取余得bucket No.，和Dictionary一样
            //lockNo = bucketNo % lockCount; 也是取余得锁No. 也就是一个锁也是可能给多个Bucket用的
            GetBucketAndLockNo(hashcode, out bucketNo, out lockNo, tables._buckets.Length, tables._locks.Length);
    
            bool resizeDesired = false;
            bool lockTaken = false;
            try
            { 
                if (acquireLock) //参数指定需要锁的话就锁上这个bucket的锁，也就在构造函数初始化时不需要锁
                    Monitor.Enter(tables._locks[lockNo], ref lockTaken);
    
                //这里是做个校验，判断tables是否在这边取完锁后其他线程把元素给扩容了，扩容会生成一个新的tables，tables变了的话上面的锁就没意义了，需要重来，所以这整个是在while(true)里面
                if (tables != _tables)
                { 
                    continue;
                }
    
                Node prev = null;
                //这里就遍历bucket里的链表了，和Dictionary差不多
                for (Node node = tables._buckets[bucketNo]; node != null; node = node._next)
                { 
                    Debug.Assert((prev == null && node == tables._buckets[bucketNo]) || prev._next == node);
                    if (hashcode == node._hashcode && _comparer.Equals(node._key, key))//看是否找到
                    { 
                        //看是否需要更新node
                        if (updateIfExists)
                        { 
                            if (s_isValueWriteAtomic) //这个是判断是否是支持原子操作的值类型，比如32位上byte,int,byte,short都是原子的，而long,double就不是了，支持原子操作的直接赋值就可以了，得注意是值类型，引用类型可不能这么搞
                            { 
                                node._value = value;
                            }
                            else //不是原子操作的值类型就new一个node
                            { 
                                Node newNode = new Node(node._key, value, hashcode, node._next);
                                if (prev == null)
                                { 
                                    tables._buckets[bucketNo] = newNode;
                                }
                                else
                                { 
                                    prev._next = newNode;
                                }
                            }
                            resultingValue = value;
                        }
                        else//不更新就直接取值
                        { 
                            resultingValue = node._value;
                        }
                        return false;  //找到了返回false，表示不用Add就Get了
                    }
                    prev = node;
                }
    
                // 找了一圈没找着，就Add吧，new一个node用Volatile的写操作写到bucket里
                Volatile.Write<Node>(ref tables._buckets[bucketNo], new Node(key, value, hashcode, tables._buckets[bucketNo]));
                checked//这里如果超出int大小，抛overflow exception, 能进这里表示一个锁罩int.MaxValue大小的Node，真成扛把子了，极端情况下只有一个锁而且Node的大小已经是Int.MaxValue才可能会出现（还要看budget同不同意）
                { 
                    tables._countPerLock[lockNo]++;
                }
    
                //如果锁罩的Node个数大于budget就表示差不多需要扩容了，黑社会表示地盘不够用了
                if (tables._countPerLock[lockNo] > _budget)
                { 
                    resizeDesired = true;
                }
            }
            finally
            { 
                if (lockTaken) //出现异常要把锁释放掉
                    Monitor.Exit(tables._locks[lockNo]);
            }
    
            if (resizeDesired)
            { 
                GrowTable(tables); //扩容
            }
    
            resultingValue = value; //result值
            return true;
        }
    }

## 再来看看是怎么扩容的 ##

private void GrowTable(Tables tables)
    { 
        const int MaxArrayLength = 0X7FEFFFFF;
        int locksAcquired = 0;
        try
        { 
            // 先把第一个锁锁住，免得其他线程也要扩容走进来
            AcquireLocks(0, 1, ref locksAcquired);
    
            //如果table已经变了，也就是那些等着上面锁的线程进来发现已经扩容完了直接返回就好了
            if (tables != _tables)
            { 
                return;
            }
    
            // 计算每个锁罩的元素的个数总和，也就是当前元素的个数
            long approxCount = 0;
            for (int i = 0; i < tables._countPerLock.Length; i++)
            { 
                approxCount += tables._countPerLock[i];
            }
    
            //如果元素总和不到Bucket大小的1/4，说明扩容扩得不是时候，归根结底是budget小了
            if (approxCount < tables._buckets.Length / 4)
            { 
                _budget = 2 * _budget;//2倍增加budget
                if (_budget < 0) //小于0说明overflow了，看看，前面用check，这里又用小于0。。
                { 
                    _budget = int.MaxValue; //直接最大值吧
                }
                return;
            }
    
            int newLength = 0;
            bool maximizeTableSize = false;
            try
            { 
                checked
                { 
                    //2倍+1取得一个奇数作了新的容量
                    newLength = tables._buckets.Length * 2 + 1;
    
                    //看是否能整除3/5/7，能就+2，直到不能整除为止，也挺奇怪这算法，List是2倍，Dictionary是比2倍大的一个质数，这里又是另外一种，只能说各人有各人的算法
                    while (newLength % 3 == 0 || newLength % 5 == 0 || newLength % 7 == 0)
                    { 
                        newLength += 2;
                    }
    
                    Debug.Assert(newLength % 2 != 0);
    
                    if (newLength > MaxArrayLength)
                    { 
                        maximizeTableSize = true;
                    }
                }
            }
            catch (OverflowException)
            { 
                maximizeTableSize = true;
            }
    
            if (maximizeTableSize)//进这里表示溢出了
            { 
                newLength = MaxArrayLength; //直接给最大值
    
                _budget = int.MaxValue; //budget也给最大值，因为没法再扩容了，给小了进来也没意义
            }
    
            //扩容之后又是熟悉的重新分配元素，和Dictionary基本一致，这里要先把所有锁锁住，前面已经锁了第一个，这里锁其他的
            AcquireLocks(1, tables._locks.Length, ref locksAcquired);
    
            object[] newLocks = tables._locks;
    
            //如果允许增加锁并则锁的个数还不到1024，就增加锁
            if (_growLockArray && tables._locks.Length < MaxLockNumber)
            { 
                newLocks = new object[tables._locks.Length * 2]; //也是2倍增加
                Array.Copy(tables._locks, 0, newLocks, 0, tables._locks.Length); //旧锁复制到新数组里
                for (int i = tables._locks.Length; i < newLocks.Length; i++) //再初始化增的锁
                { 
                    newLocks[i] = new object();
                }
            }
    
            //新的Node数组
            Node[] newBuckets = new Node[newLength];
            int[] newCountPerLock = new int[newLocks.Length];
    
            //遍历bucket
            for (int i = 0; i < tables._buckets.Length; i++)
            { 
                Node current = tables._buckets[i];//当前node
                while (current != null)
                { 
                    Node next = current._next;
                    int newBucketNo, newLockNo;
                    //算新的bucket No.和lock No.
                    GetBucketAndLockNo(current._hashcode, out newBucketNo, out newLockNo, newBuckets.Length, newLocks.Length);
                    //重建个新的node，注意next指到了上一个node，和Dictionary里一样
                    newBuckets[newBucketNo] = new Node(current._key, current._value, current._hashcode, newBuckets[newBucketNo]);
    
                    checked
                    { 
                        newCountPerLock[newLockNo]++; //这个锁又罩了一个小弟，加一个
                    }
    
                    current = next;
                }
            }
    
            //调整下budget
            _budget = Math.Max(1, newBuckets.Length / newLocks.Length); 
    
            //得到新的table
            _tables = new Tables(newBuckets, newLocks, newCountPerLock);
        }
        finally
        { 
            // 释放锁
            ReleaseLocks(0, locksAcquired);
        }
    }

通过这几个函数差不多也就清楚了`ConcurrentDictionary`整个的原理，其他函数有兴趣的可以去看看，都差不多这个意思。  
总结

说完了，总结下，`ConcurrentDictionary`可以说是为了避免一个大锁锁住整个`Dictionary`带来的性能损失而出来的，当然也是采用空间换时间，不过这空间换得还是很值得的，一些`object`而已。

原理在于`Dictionary`本质是是一个链表数组，只有在多线程同时操作到数组里同一个链表时才需要锁，所以就用到一个锁数组，每个锁罩着几个小弟(`bucket`及`bucket`内的链表元素)，这样多线程读写不同锁罩的区域的时候可以同时进行而不会等待，进而提高多线程性能。

不过也凡事无绝对，不同业务场景的需求不一样，可能`Dictionary`配合`ReaderWriterLockSlim`在某些场景（比如读的机会远大于写的）可能会有更好的表现。

[https_github.com_dotnet_corefx_blob_master_src_System.Collections.Concurrent_src_System_Collections_Concurrent_ConcurrentDictionary.cs]: https://github.com/dotnet/corefx/blob/master/src/System.Collections.Concurrent/src/System/Collections/Concurrent/ConcurrentDictionary.cs