C#阻塞集合处理一个项目两次答案

【问题标题】：C# Blocking collection processing an item twiceC#阻塞集合处理一个项目两次
【发布时间】：2021-03-03 23:24:19
【问题描述】：

以下是实际生产环境的示例模拟。在某些情况下，在下面的示例中，从阻塞集合中检索到的项目会被不同的线程处理两次。

有人知道这个问题的原因吗？即使使用下面的代码，也很难模拟这个问题，这个问题只发生在生产环境中。

class Program
{
    static void Main(string[] args)
    {
        TestBlockingCollection obj1 = new TestBlockingCollection();

        obj1.InvokeThread();

        for(int i=0; i<=10; i++)
        {
            obj1.AddToLiveDataQueue($"Item-{i}");
        }
    }
}
class TestBlockingCollection
{
    BlockingCollection<string>Quueue { get; set; }
    public void AddToLiveDataQueue(string msg)
    {
        try
        {
            Console.WriteLine($"[{System.Threading.Thread.CurrentThread.ManagedThreadId}] Adding to live data queue");
            Quueue.TryAdd(msg);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
    }
    public void InvokeThread()
    {
        Quueue = new BlockingCollection<string>();
        ThreadStart threadObj = new ThreadStart(ConsumerThread);
        Thread thread = new Thread(threadObj);
        thread.Start();
    }
    private void ConsumerThread()
    {
        foreach (string item in Quueue.GetConsumingEnumerable())
        {
            try
            {
                Console.WriteLine($"[{System.Threading.Thread.CurrentThread.ManagedThreadId}] Started processing {item}");
                Thread.Sleep(10000); // does some operation
                Console.WriteLine($"[{System.Threading.Thread.CurrentThread.ManagedThreadId}] Finished processing {item}");

            }
            catch(Exception e)
            {
                Console.WriteLine(e.Message);
            }
        }
    }
}

【问题讨论】：

“被不同的线程处理了两次”。在您的示例代码中，只有一个线程在进行处理。
生产环境中会不会调用InvokeThread方法两次？
没有机会。它只调用一次，当 exe 启动时。
我有点紧张BlockingCollection 在InvokeThread 方法中被实例化。我希望它被声明为readonly，并在构造函数中实例化。我不认为这会解决你的问题。可能在程序的其他部分发生了其他事情。
我真的希望您发布实际存在问题的生产代码，而不是没有问题的演示代码。请问可以吗？

标签： c# .net multithreading c#-4.0

【解决方案1】：

我不是 100% 同意这一点，因为我看不到您的 BlockingCollection 的任何实现细节，但我的直觉是以下代码可能是罪魁祸首：

foreach (string item in Quueue.GetConsumingEnumerable())

您可能会将相同的项目复制到多个“可枚举”对象中，并且调用线程会处理整个列表。

可能值得尝试以下方式：

string item;
while ((item = Quueue.GetNext()) != null)
{
  try
  {
    Console.WriteLine($"[{System.Threading.Thread.CurrentThread.ManagedThreadId}] Started processing {item}");
    Thread.Sleep(10000); // does some operation
    Console.WriteLine($"[{System.Threading.Thread.CurrentThread.ManagedThreadId}] Finished processing {item}");

  }
  catch(Exception e)
  {
    Console.WriteLine(e.Message);
  }
}

GetNext() 每次调用时都会从 BlockingCollection 中显式删除一项且仅一项。

【讨论】：