c# - 与 ConcurrentDictionary 一起使用时 C# LINQ OrderBy 线程安全吗？

Question

我的工作假设是，当与 System.Collections.Concurrent 集合(包括 ConcurrentDictionary)一起使用时，LINQ 是线程安全的。

(其他 Overflow 帖子似乎同意:link)

但是，对 LINQ OrderBy 扩展方法的实现的检查表明，对于实现 ICollection 的并发集合子集，它似乎不是线程安全的(例如 < strong>ConcurrentDictionary).

OrderedEnumerable GetEnumerator ( source here ) 构造一个 Buffer 结构 ( source here ) 的实例，它试图转换集合到 ICollection(ConcurrentDictionary 实现)，然后执行 collection.CopyTo 并使用一个初始化为集合大小的数组。

因此，如果 ConcurrentDictionary(在本例中为具体的 ICollection)在 OrderBy 操作期间增加了大小，在初始化数组和复制到数组之间，此操作会抛出。

下面的测试代码显示了这个异常:

(注意:我很欣赏在线程安全的集合上执行 OrderBy，它在你下面发生变化并没有那么有意义，但我认为它不应该抛出)

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace Program
{
    class Program
    {
        static void Main(string[] args)
        {
            try
            {
                int loop = 0;
                while (true) //Run many loops until exception thrown
                {
                    Console.WriteLine($"Loop: {++loop}");

                    _DoConcurrentDictionaryWork().Wait();
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine(ex);
            }
        }

        private static async Task _DoConcurrentDictionaryWork()
        {
            var concurrentDictionary = new ConcurrentDictionary<int, object>();
            var keyGenerator = new Random();
            var tokenSource = new CancellationTokenSource();

            var orderByTaskLoop = Task.Run(() =>
            {
                var token = tokenSource.Token;
                while (token.IsCancellationRequested == false)
                {
                    //Keep ordering concurrent dictionary on a loop
                    var orderedPairs = concurrentDictionary.OrderBy(x => x.Key).ToArray(); //THROWS EXCEPTION HERE

                    //...do some more work with ordered snapshot...
                }
            });

            var updateDictTaskLoop = Task.Run(() =>
            {
                var token = tokenSource.Token;
                while (token.IsCancellationRequested == false)
                {
                    //keep mutating dictionary on a loop
                    var key = keyGenerator.Next(0, 1000);
                    concurrentDictionary[key] = new object();
                }
            });

            //Wait for 1 second
            await Task.Delay(TimeSpan.FromSeconds(1));

            //Cancel and dispose token
            tokenSource.Cancel();
            tokenSource.Dispose();

            //Wait for orderBy and update loops to finish (now token cancelled)
            await Task.WhenAll(orderByTaskLoop, updateDictTaskLoop);
        }
    }
}

OrderBy 抛出异常导致了几个可能的结论之一:

1) 我关于 LINQ 对并发集合线程安全的假设是不正确的，只有在 LINQ 查询期间不发生变化的集合(无论是否并发)上执行 LINQ 才是安全的

2) LINQ OrderBy 的实现存在一个错误，实现尝试将源集合转换为 ICollection 并尝试执行集合复制是不正确的(并且它应该只需通过迭代 IEnumerable 的默认行为即可)。

3) 我误解了这里发生的事情......

非常感谢您的想法!

Answer 1

在任何地方都没有说明 OrderBy(或其他 LINQ 方法)应该始终使用源 IEnumerable 的 GetEnumerator 或者它应该是线程安全的在并发集合上。 promise 的就是这个方法

Sorts the elements of a sequence in ascending order according to a key.

ConcurrentDictionary 在某种全局意义上也不是线程安全的。对于对其执行的其他操作，它是线程安全的。更重要的是，文档说

All public and protected members of ConcurrentDictionary are thread-safe and may be used concurrently from multiple threads. However, members accessed through one of the interfaces the ConcurrentDictionary implements, including extension methods, are not guaranteed to be thread safe and may need to be synchronized by the caller.

因此，您的理解是正确的(OrderBy 会看到您传递给它的 IEnumerable 实际上是 ICollection，然后会得到该集合的长度，分配该大小的缓冲区，然后将调用 ICollection.CopyTo，这当然对任何类型的集合都不是线程安全的)，但这不是 OrderBy 中的错误因为 OrderBy 和 ConcurrentDictionary 都没有 promise 你的假设。

如果您想在 ConcurrentDictionary 上以线程安全的方式执行 OrderBy，您需要依赖 promise 线程安全的方法。例如:

// note: this is NOT IEnumerable.ToArray()
// but public ToArray() method of ConcurrentDictionary itself
// it is guaranteed to be thread safe with respect to other operations
// on this dictionary
var snapshot = concurrentDictionary.ToArray();
// we are working on snapshot so no one other thread can modify it
// of course at this point real contents of dictionary might not be
// the same as our snapshot
var sorted = snapshot.OrderBy(c => c.Key);

如果您不想分配额外的数组(使用ToArray)，您可以使用Select(c => c)，在这种情况下它会起作用，但随后我们再次处于没有实际意义的领域，并依靠某些东西在未 promise 的情况下安全使用(Select 也不会总是枚举您的集合。如果集合是数组或列表 - 它会快捷方式并改用索引器)。所以你可以像这样创建扩展方法:

public static class Extensions {
    public static IEnumerable<T> ForceEnumerate<T>(this ICollection<T> collection) {
        foreach (var item in collection)
            yield return item;
    }
}

如果你想安全并且不想分配数组，可以像这样使用它:

concurrentDictionary.ForceEnumerate().OrderBy(c => c.Key).ToArray();

在这种情况下，我们强制枚举 ConcurrentDictionary(我们知道它在文档中是安全的)，然后将其传递给 OrderBy，因为知道它不会对此造成任何伤害纯 IEnumerable。请注意，正如 mjwills 在评论中正确指出的那样，这与 ToArray 并不完全相同，因为 ToArray 会生成快照(锁定集合以防止在构建数组时进行修改)和 Select\yield 不获取任何锁(因此可能会在枚举进行时添加\删除项目)。尽管我怀疑在执行问题中描述的事情时它是否重要 - 在这两种情况下 OrderBy 完成后 - 你不知道你的订购结果是否反射(reflect)了当前的收集状态。