读写作为并行任务答案

【问题标题】：Read and Write as Parallel Tasks读写作为并行任务
【发布时间】：2013-10-27 12:58:13
【问题描述】：

读写2个并行任务如下图：

Task[] tasks = new Task[2];
var entityCollection = new BlockingCollection<Dictionary<String, object>>();

tasks[0] = Task.Factory.StartNew(() => ReadData(entityCollection), TaskCreationOptions.LongRunning);
tasks[1] = Task.Factory.StartNew(() => WriteJsontoFile(JSONFileName, entityCollection), TaskCreationOptions.LongRunning);

Task.WaitAll(tasks);

读取任务：

private void ReadData(BlockingCollection<Dictionary<String, object>> collection)
{
    do
    {
        //continuously data is being read in to entities, this part is working fine and then adding it to collection of BlockingCollection type to be consumed in Write task
        entitites.ToList().ForEach(e => collection.Add(e));
    } while (true);
    collection.CompleteAdding();
}

写任务：

private void WriteJsontoFile(String JsonFileName, BlockingCollection<Dictionary<String, object>> source)
{
    using (StreamWriter sw = new StreamWriter(JsonFileName, true))
    {
        Parallel.ForEach(source.GetConsumingPartitioner(), (line) => ser.Serialize(sw, line));
    }
}

GetConsumingPartitioner()相关代码：

public static class BlockingCollection
{
    public static Partitioner<T> GetConsumingPartitioner<T>(
    this BlockingCollection<T> collection)
    {
        return new BlockingCollectionPartitioner<T>(collection);
    }
}

class BlockingCollectionPartitioner<T> : Partitioner<T>
{
    private BlockingCollection<T> _collection;

    internal BlockingCollectionPartitioner(BlockingCollection<T> collection)
    {
        if (collection == null)
            throw new ArgumentNullException("collection");
        _collection = collection;
    }

    public override bool SupportsDynamicPartitions
    {
        get { return true; }
    }

    public override IList<IEnumerator<T>> GetPartitions(int partitionCount)
    {
        if (partitionCount < 1)
            throw new ArgumentOutOfRangeException("partitionCount");
        var dynamicPartitioner = GetDynamicPartitions();
        return Enumerable.Range(0, partitionCount).Select(_ =>
            dynamicPartitioner.GetEnumerator()).ToArray();
    }

    public override IEnumerable<T> GetDynamicPartitions()
    {
        return _collection.GetConsumingEnumerable();
    }
}

我在写入任务中得到以下异常：

Count 不能小于零。\r\n参数名称：count

【问题讨论】：

那么这个异常到底发生在哪里呢？它有什么类型，什么堆栈？
出于某种奇怪的原因，VS 显示此消息而没有更多信息“{无法评估表达式，因为线程在无法进行垃圾收集的点停止，可能是因为代码已优化。}”。无论如何，布拉姆的回答奏效了。
它适用于您非常狭窄的示例代码，但您不能停止使用该代码（请不要使用超时或轮询循环......）。另外，你为什么使用分区器并且现在很高兴没有它们？这令人惊讶，因为我认为您这样做是有原因的。
这里是关于分区器的一些事情：blogs.msdn.com/b/pfxteam/archive/2010/04/06/9990420.aspx我仍然有分区器在工作代码中

标签： c# .net c#-4.0 task-parallel-library

【解决方案1】：

这不是消费的标准语法

BlockingCollection Class

// Consume consume the BlockingCollection 
while (true) Console.WriteLine(bc.Take());

【讨论】：

其实我正在使用 foreach (var item in bc.GetConsumingEnumerable()) { //code } 它正在工作