流式数据BlockingCollection

Question

在Stephen的书第88页

http://www.microsoft.com/download/en/details.aspx?id=19222

这是密码

private BlockingCollection<T> _streamingData = new BlockingCollection<T>();
// Parallel.ForEach
Parallel.ForEach(_streamingData.GetConsumingEnumerable(),
item => Process(item));
// PLINQ
var q = from item in _streamingData.GetConsumingEnumerable().AsParallel()
...
select item;

斯蒂芬接着提到

“当将调用GetConsumingEnumerable作为数据源的结果传递给Parallel.ForEach时，循环使用的线程有可能在集合变为空时阻塞。而Parallel.ForEach可能不会将阻塞的线程释放回ThreadPool以供退休或其他用途。因此，有了上面所示的代码，如果有一段时间集合为空，则进程中的线程计数可能会稳步增长；”

我不明白为什么线程数会增加？

如果集合是空的，那么块集合不会请求任何更多的线程吗？

因此，您不需要执行WithDegreeOfParallelism来限制在BlockingCollection上使用的线程数。

Answer 1

线程池有一个爬山算法，它使用它来估计适当的线程数。只要增加线程的吞吐量，线程池就会创建更多的线程。它将假设发生了一些阻塞或IO，并试图通过检查系统中的处理器数来使CPU饱和。

这就是为什么在线程池线程上执行IO和阻塞内容可能是危险的。

下面是上述行为的一个完全有效的示例：

        BlockingCollection<string> _streamingData = new BlockingCollection<string>();

        Task.Factory.StartNew(() =>
            {
                for (int i = 0; i < 100; i++)
                {
                    _streamingData.Add(i.ToString());
                    Thread.Sleep(100);
                }
            });

        new Thread(() =>
            {
                while (true)
                {
                    Thread.Sleep(1000);
                    Console.WriteLine("Thread count: " + Process.GetCurrentProcess().Threads.Count);
                }
            }).Start();

        Parallel.ForEach(_streamingData.GetConsumingEnumerable(), item =>
            {
            });

我不知道为什么线程数不断上升，尽管它并没有增加吞吐量。根据我解释的模型，它不会生长。但我不知道我的模型是否真的正确。

也许线程池有一个额外的启发，如果它没有看到任何进展(以每秒完成的任务来衡量)，它就会生成线程。这将是有意义的，因为这可能会防止应用程序中出现大量死锁。如果重要任务由于等待现有任务退出并使线程可用而无法运行，则可能会发生死锁。这是一个众所周知的线程池问题.