使用交换从 ConcurrentBag 中取出所有项目

Question

我正试图一口气从ConcurrentBag 获取所有项目。由于集合中没有像 TryEmpty 这样的东西，因此我采用了与此处所述相同的方式使用 Interlocked.Exchange：How to remove all Items from ConcurrentBag?

我的代码如下所示：

private ConcurrentBag<Foo> _allFoos; //Initialized in constructor.

public bool LotsOfThreadsAccessingThisMethod(Foo toInsert)
{
    this._allFoos.Add(toInsert);
    return true;
}

public void SingleThreadProcessingLoopAsALongRunningTask(object state)
{
    var token = (CancellationToken) state;
    var workingSet = new List<Foo>();

    while (!token.IsCancellationRequested)
    {
        if (!workingSet.Any())
        {
            workingSet = Interlocked.Exchange(ref this._allFoos, new ConcurrentBag<Foo>).ToList();
        }

        var processingCount = (int)Math.Min(workingSet.Count, TRANSACTION_LIMIT);

        if (processingCount > 0)
        {
            using (var ctx = new MyEntityFrameworkContext())
            {
                ctx.BulkInsert(workingSet.Take(processingCount));
            }
            workingSet.RemoveRange(0, processingCount);
        }
    }
}

问题是这有时会遗漏添加到列表中的项目。我编写了一个测试应用程序，它将数据提供给我的ConcurrentBag.Add 方法并验证它正在发送所有数据。当我在Add 调用上设置断点并检查ConcurrentBag 之后的计数时，它为零。该项目只是没有被添加。

我相当肯定这是因为 Interlocked.Exchange 调用没有使用 ConcurrentBag 的内部锁定机制，所以它在交换的某个地方丢失了数据，但我不知道实际发生了什么。

如何在不借助自己的锁定机制的情况下一次从ConcurrentBag 中取出所有项目？还有为什么Add 会忽略这个项目？

Answer 1

我认为不需要从ConcurentBag 获取所有项目。您可以通过如下更改处理逻辑（无需自己的同步或互锁交换）来实现您尝试实现的完全相同的行为：

public void SingleThreadProcessingLoopAsALongRunningTask(object state)
{
    var token = (CancellationToken)state;
    var buffer = new List<Foo>(TRANSACTION_LIMIT);
    while (!token.IsCancellationRequested)
    {
        Foo item;
        if (!this._allFoos.TryTake(out item))
        {
            if (buffer.Count == 0) continue;
        }
        else
        {
            buffer.Add(item);
            if (buffer.Count < TRANSACTION_LIMIT) continue;
        }
        using (var ctx = new MyEntityFrameworkContext())
        {
            ctx.BulkInsert(buffer);
        }
        buffer.Clear();
    }
}