18

在尝试创建一个简单的管道时BlockingCollection<T>，我使用 Task Parallel Library 遇到了一个可量化和可重复的问题ConcurrentQueue<T>。GetConsumingEnumerable

简而言之，从一个线程将条目添加到默认值BlockingCollection<T>（在引擎盖下依赖于 a ConcurrentQueue<T>）并不能保证它们会BlockingCollection<T>从另一个调用该GetConsumingEnumerable()方法的线程中弹出。

我创建了一个非常简单的 Winforms 应用程序来重现/模拟它，它只是将整数打印到屏幕上。

  • Timer1负责排队工作项......它使用一个并发字典_tracker，以便它知道它已经添加到阻塞集合中的内容。
  • Timer2只是记录BlockingCollection&的计数状态_tracker
  • START 按钮启动 a Paralell.ForEach，它简单地遍历阻塞集合GetConsumingEnumerable()并开始将它们打印到第二个列表框。
  • STOP 按钮停止Timer1阻止更多条目被添加到阻塞集合中。
public partial class Form1 : Form
{
    private int Counter = 0;
    private BlockingCollection<int> _entries;
    private ConcurrentDictionary<int, int> _tracker;
    private CancellationTokenSource _tokenSource;
    private TaskFactory _factory;

    public Form1()
    {
        _entries = new BlockingCollection<int>();
        _tracker = new ConcurrentDictionary<int, int>();
        _tokenSource = new CancellationTokenSource();
        _factory = new TaskFactory(); 
        InitializeComponent();
    }

    private void timer1_Tick(object sender, EventArgs e)
    { //ADDING TIMER -> LISTBOX 1
        for(var i = 0; i < 3; i++,Counter++)
        {
            if (_tracker.TryAdd(Counter, Counter))
            _entries.Add(Counter);
            listBox1.Items.Add(string.Format("Adding {0}", Counter));
        }
    }

    private void timer2_Tick_1(object sender, EventArgs e)
    { //LOGGING TIMER -> LIST BOX 3
        listBox3.Items.Add(string.Format("Tracker Count : {0} / Entries Count : {1}", _tracker.Count, _entries.Count));
    }

    private void button1_Click(object sender, EventArgs e)
    { //START BUTTON -> LOGS TO LIST BOX 2

        var options = new ParallelOptions {
                                CancellationToken = _tokenSource.Token,
                                MaxDegreeOfParallelism = 1
                            };

        _factory.StartNew(() => { Parallel.ForEach(_entries.GetConsumingEnumerable(), options, DoWork); });

        timer1.Enabled = timer2.Enabled = true;
        timer1.Start();
        timer2.Start();
    }

    private void DoWork(int entry)
    {
        Thread.Sleep(1000); //Sleep for 1 second to simulate work being done.
        Invoke((MethodInvoker)(() => listBox2.Items.Add(string.Format("Processed {0}", entry))));
        int oldEntry;
        _tracker.TryRemove(entry, out oldEntry);
    }

    private void button2_Click(object sender, EventArgs e)
    { //STOP BUTTON
        timer1.Stop();
        timer1.Enabled = false;
    }

以下是事件的顺序：

  • 按开始
  • Timer1 滴答和 ListBox1 立即更新 3 条消息（添加 0、1、2）
  • ListBox2 随后更新了 3 条消息，间隔 1 秒
    • 处理 0
    • 处理 1
    • 处理 2
  • Timer1 滴答和 ListBox1 立即更新 3 条消息（添加 3、4、5）
  • ListBox2 用 2 条消息更新，间隔 1 秒
    • 处理 3
    • 处理 4
    • 处理 5未打印...似乎已经“丢失”
  • 按 STOP 以防止计时器 1 添加更多消息
  • 等等...“处理中 5”仍然没有出现

缺少条目

您可以看到并发字典仍在跟踪 1 项尚未处理并随后从中删除_tracker

如果我再次按开始，则 timer1 开始添加更多 3 个条目，并行循环恢复活力，打印 5、6、7 和 8。

在随后的项目被推到它后面后返回条目

我完全不知道为什么会发生这种情况。再次调用 start 显然调用了一个 newtask，它调用了一个 Paralell foreach，并重新执行 GetConsumingEnumerable()，它神奇地找到了丢失的条目......我

为什么BlockingCollection.GetConsumingEnumerable()不保证迭代添加到集合中的每个项目。

为什么添加更多条目随后会导致它“解开”并继续处理？

4

4 回答 4

20

你不能GetConsumingEnumerable()在Parallel.ForEach().

使用GetConsumingPartitioner来自TPL 的附加功能

在博文中，您还将得到一个解释为什么不能使用GetConsumingEnumerable()

Parallel.ForEach 和 PLINQ 默认使用的分区算法使用分块来最小化同步成本：不是每个元素获取一次锁，而是获取锁，获取一组元素（一个块），然后释放锁。

即 Parallel.ForEach 等到它收到一组工作项后再继续。正是你的实验显示的。

于 2012-04-18T12:19:58.087 回答
7

从 .net 4.5 开始，您可以创建一个分区器，它一次只需要一个项目：

var partitioner = Partitioner.Create(jobsBatchesQ.queue.GetConsumingEnumerable(), EnumerablePartitionerOptions.NoBuffering);
Parallel.ForEach(partitioner, new ParallelOptions { MaxDegreeOfParallelism = (currentTask.ParallelLevel > 0 ? currentTask.ParallelLevel : 1) }, (batch, state) => {//do stuff}

https://msdn.microsoft.com/en-us/library/system.collections.concurrent.enumerablepartitioneroptions(v=vs.110).aspx

于 2015-11-24T18:50:03.387 回答
2

我无法使用简单的控制台应用程序复制您的行为，而这些应用程序基本上做同样的事情（在 .Net 4.5 beta 上运行，这可能会有所作为）。但我认为发生这种情况的原因是Parallel.ForEach()试图通过将输入集合分成块来优化执行。并且使用您的可枚举，在您向集合中添加更多项目之前，无法创建一个块。有关详细信息，请参阅MSDN 上的 PLINQ 和 TPL 的自定义分区器。

要解决此问题，请不要使用Parallel.ForEach(). 如果您仍想并行处理这些项目，则可以Task在每次迭代中启动 a。

于 2012-04-18T12:10:50.570 回答
0

为了清楚起见，我觉得我应该注意的是，如果您能够在执行 Parallel.foreach 之前调用 BlockingCollection 的 .CompleteAdding() 方法，那么您在上面描述的问题将不成问题。我已经多次将这两个对象一起使用，效果很好。

此外，您可以在调用 CompleteAdding() 后随时重新设置 BlockingCollection，以便在需要时添加更多项目 (_entries = new BlockingCollection();)

如果您多次单击开始和停止按钮，如下更改上面的点击事件代码将解决您缺少条目的问题并使其按预期工作：

private void button2_Click(object sender, EventArgs e)
{ //STOP BUTTON
    timer1.Stop();
    timer1.Enabled = false;
>>>>_entries.CompleteAdding();
>>>>_entries = new BlockingCollection<int>();
}
于 2012-11-04T06:00:42.910 回答