为什么我永远无法清空 BlockingCollection？答案

【问题标题】：Why can I never empty the BlockingCollection?为什么我永远无法清空 BlockingCollection？
【发布时间】：2017-02-05 20:02:16
【问题描述】：

我定期下载一些 JSON，比如每 10 秒...当数据到达时，会触发一个事件。触发的事件只是将 JSON 添加到 BlockingCollection<string>（待处理）。

我正在尝试尽快处理 JSON（一旦到达...）：

public class Engine
{
    private BlockingCollection<string> Queue = new BlockingCollection<string>();
    private DataDownloader DataDownloader;

    public void Init(string url, int interaval)
    {
        dataDownloader = new DataDownloader(url, interaval);
        dataDownloader .StartCollecting();
        dataDownloader .DataReceivedEvent += DataArrived;

        //Kick off a new task to process the incomming JSON
        Task.Factory.StartNew(Process, TaskCreationOptions.LongRunning);
    }

    /// <summary>
    /// Processes the JSON in parallel
    /// </summary>
    private void Process()
    {
       Parallel.ForEach(Queue.GetConsumingEnumerable(), ProcessJson);
    }

    /// <summary>
    /// Deserializes JSON and adds result to database
    /// </summary>
    /// <param name="json"></param>
    private void ProcessJson(string json)
    {
        using (var db = new MyDataContext())
        {
            var items= Extensions.DeserializeData(json);
            foreach (var item in items)
            {
                db.Items.Add(item);
                db.SaveChanges();
            }
        }
    }

    private void DataArrived(object sender, string json)
    {
        Queue.Add(json);
        Console.WriteLine("Queue length: " + Queue.Count);
    }
}

当我运行该程序时，它可以工作并且数据被添加到数据库中，但是如果我查看来自Console.WriteLine("Queue length: " + Queue.Count); 的控制台消息，我会得到如下信息：

我尝试将我的Process 修改为如下所示：

/// <summary>
/// Processes the JSON in parallel
/// </summary>
private void Process()
{

    foreach (var json in Queue.GetConsumingEnumerable())
    {
        ProcessJson(json);
    }
 }

然后我添加了多个Task.Factory.StartNew(Process, TaskCreationOptions.LongRunning);，但我遇到了同样的问题......

有人知道这里出了什么问题吗？

【问题讨论】：

处理的长度不是取决于你输入的json数据吗？如果前几个很简单，那么它们会很快执行，而如果你的第 8 个很大，它会堆积起来。在添加每个项目后，您是否还必须保存对 db 的更改？（我没有使用 DataContext 的经验，但我想那一定很昂贵）
它已经尽可能快了，是操作系统的职责。如果您的代码生成速度比消耗速度快，这很可能是因为处理 json 并不便宜，那么除了增加集合之外别无选择。这就是为什么 BlockingCollection 中的“阻塞”很重要，它确保您的程序不会因 OOM 而崩溃。
@CrudaLilium Nah，JSON 的大小通常相同，所以不是这样......但很好的想法。
@HansPassant 使用 WebClient 每 10 秒下载一次数据。添加更多消费者肯定会减少队列中的项目数量吗？我希望大多数时候队列中有 0 个项目（我尝试了最多 10 个消费者任务）。
谷歌“阿姆达尔定律”，使用并发分析器。为您提供事实而非猜测。

标签： c# concurrency producer-consumer blockingcollection

【解决方案1】：

队列将在处理开始之前被填满。可能是因为需要加载实体框架的东西并且必须建立数据库连接，这需要一段时间。

然后GetConsumingEnumerable()开始赶上下载进程，在下载过程中耗尽队列。集合为空，MoveNext() 返回 false，Parallel.ForEach() 退出，Process() 结束。

然后您会看到队列开始填满，因为它不再被消耗。

您需要继续尝试从 BlockingCollection 中读取，直到下载过程完成。

【讨论】：

即使我将foreach 包装在while(true) 中并添加多个任务，这种情况仍然会发生。