ConcurrentBag - 添加多个项目？答案

【问题标题】：ConcurrentBag - Add Multiple Items?ConcurrentBag - 添加多个项目？
【发布时间】：2012-04-27 23:57:05
【问题描述】：

有没有办法一次将多个项目添加到 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 添加一个项目？

【问题讨论】：

标签： c# concurrency concurrent-collections

【解决方案1】：

是的:)

Concat 可能是Enumerable 扩展之一。它不会向ConcurrentBag 添加任何内容，它只是返回一些时髦的对象，其中包含原始包以及您尝试添加的任何内容。

注意Concat 的结果不再是ConcurrentBag，所以你不想使用它。它是通用 LINQ 框架的一部分，可以组合不可变序列。当然，该框架不会尝试将操作数的并发属性扩展到结果，因此生成的对象不太适合多线程访问。

（基本上，Concat 适用于ConcurrentBag，因为它暴露了IEnumerable<T> 接口。）

【讨论】：

这个“答案”显然没有回答问题。不知道为什么它被标记为答案。
“是”哪个 OP 问题？

【解决方案2】：

Concat 是 LINQ 提供的扩展方法。这是一个不可变的操作，它返回另一个IEnumerable，它可以枚举源集合，紧随其后的是指定的集合。它不会以任何方式更改源集合。

您需要一次将您的项目添加到ConcurrentBag。

【讨论】：

【解决方案3】：

我遇到了类似的问题，尝试并行处理较小的数据块，因为一大块正在使我用来在发送端访问我的数据的 Web 服务超时，但我不希望事情运行得更慢依次处理每个块。逐条处理数据记录甚至更慢 - 因为我调用的服务可以处理批量请求，所以最好尽可能多地提交而不超时。

就像 Vlad 所说，将并发包连接到对象类型的列表不会返回并发包，因此 concat 不起作用！（我花了一段时间才意识到我做不到。）

试试这个 - 创建一个List<T>，然后创建一个ConcurrentBag<List<T>>。在每次并行迭代中，它将向并发包中添加一个新列表。并行循环完成后，循环遍历 ConcurrentBag 和 concat（或联合，如果您想消除可能的重复项）到您创建的第一个 List<T>，以便将所有内容“展平”到一个列表中。

【讨论】：

最后使用 SelectMany 来扁平化。

【解决方案4】：

（我知道这是一篇旧帖子，想补充一点）。

就像其他人说的：是的，你需要一一添加。在我的例子中，我添加了一个小的扩展方法来让事情变得更简洁，但在引擎盖下它做同样的事情：

    public static void AddRange<T>(this ConcurrentBag<T> @this, IEnumerable<T> toAdd)
    {
        foreach (var element in toAdd)
        {
            @this.Add(element);
        }
    }

然后：

    ConcurrentBag<int> ccBag = new ConcurrentBag<int>();
    var listOfThings = new List<int>() { 1, 2, 4, 5, 6, 7, 8, 9 };
    ccBag.AddRange(listOfThings);

我还研究了使用 AsParallel 在扩展方法中添加，但是在运行了一些添加各种大小的字符串列表的测试之后，使用 AsParallel（如此处所示）相对于传统的 for 循环来说始终慢.

    public static void AddRange<T>(this ConcurrentBag<T> @this, IEnumerable<T> toAdd)
    {
        toAdd.AsParallel().ForAll(t => @this.Add(t));
    }

【讨论】：

这应该是问题的答案。
这是正确答案。还运行测试尝试使用 Parallel (AsParallel().ForAll()) 并且速度较慢。

【解决方案5】：

Concat 方法是包含在公共 Enumerable 静态类中的一种方法，支持 System.Linq 库（内部 .NET System.Core 组装）。

但是，Concat 包含向ConcurrentBag<T> 对象提供“添加范围”要求的限制，其行为如下图所示（在 Visual Studio 中的第 47 行）：

要Concat方法匹配“添加范围”要求，需要更新当前ConcurrentBag<T>对象实例；如果程序需要添加多个范围，则需要从当前的ConcurrentBag<T>（递归地）为每个范围创建自动引用实例。

然后，我不使用Concat 方法，如果我可以推荐，我不推荐。我遵循Eric 的答案中的一个类似示例，其中我从ConcurrentBag<T> 开发了一个派生类，它为我提供了使用ConcurrentBag<T> 基本方法将IEnumerable<T> 项目添加到派生实例的AddRange 方法，如下所示：

public class ConcurrentBagCompleted<T> : ConcurrentBag<T>
{
    public ConcurrentBagCompleted() : base() { }

    public ConcurrentBagCompleted(IEnumerable<T> collection):base(collection)
    {
    }

    public void AddRange(IEnumerable<T> collection)
    {
        Parallel.ForEach(collection, item =>
        {
            base.Add(item);
        });
    }
}

【讨论】：

【解决方案6】：

ConcurrentBag<T> 的第二个构造函数将 IEnumerable 作为参数并从中构造包。对于这种情况，这是我发现的最快、最安全的方法。请参考ConcurrentBag<T>(IEnumerable<T>) 因此，在问题的情况下，解决方案是：

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

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

【讨论】：