如何同时从 BlockingCollection 中获取项目？

Question

我有一个BlockingCollection 的整数。 下面是我正在尝试开发的代码，用于同时从 BlockingCollection 中删除项目。

static void Produce()
    {
        for (int i = 0; i < 100000; i++)
        {
            bc3.Add(i);
        }
        Run();
    }

    static void Run()
    {
        for (int i = 0; i < 5; i++)
        {
            Task.Factory.StartNew(Process, TaskCreationOptions.LongRunning);
        }
    }

    static void Process()
    {
        var stopWatch = new Stopwatch();
        stopWatch.Start();
        while (bc3.Count!= 0)
        {
            bc3.Take();
        }
        stopWatch.Stop();
        Console.WriteLine("Thread {0}",
        Thread.CurrentThread.ManagedThreadId);
        Console.WriteLine("Elapsed Time {0}", stopWatch.Elapsed.Seconds);

    }

这是通过创建 5 个任务以更快的方式删除项目的正确方法吗？

Answer 1

秒表问题

你的测量结果是错误的，因为你使用

stopWatch.Elapsed.Seconds

而不是

stopWatch.ElapsedMilliseconds

您只显示秒，而忽略分钟、小时等。

并发问题

不，这不是从 BlockingCollection 中删除项目的正确方法。不起作用的语句是

bc3.Count != 0

所有 5 个任务可能同时检查这个条件，发现还剩下 1 个项目。他们5个都去

bc3.Take();

一个任务能够移除该项目，其他 4 个任务正在等待。

解决这个问题的一种方法是添加

bc3.CompleteAdding();

在Produce().

垃圾回收问题

一旦第一个任务完成，Run() 方法就会完成，并且该方法中的所有任务都会超出范围并被垃圾回收。这可能会让您只看到 1 条而不是 5 条完成消息。

要解决此问题，请使用

    static void Run()
    {
        var tasks = new List<Task>();
        for (int i = 0; i < 5; i++)
        {
            tasks.Add(Task.Factory.StartNew(Process, TaskCreationOptions.LongRunning));
        }
        try
        {
            Task.WaitAll(tasks.ToArray());
        }
        catch (AggregateException)
        { }        
    }

同步成本

这是我的输出之一，包含 5 个任务（和 100000000 个项目）：

Thread 11
Thread 13
Thread 12
Thread 14
Thread 15
Elapsed Time 12878
Elapsed Time 13122
Elapsed Time 13128
Elapsed Time 13128
Elapsed Time 13128
Run: 13140

将此与单个任务进行比较：

Thread 12
Elapsed Time 10147
Run: 10149

这是因为只有一个 Take() 方法可以删除一个项目，并且同步需要额外的时间。