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

Posted 2023-02-15

【中文标题】C#阻塞集合处理一个项目两次

【英文标题】：C# Blocking collection processing an item twice

【发布时间】：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，并在构造函数中实例化。我不认为这会解决你的问题。可能在程序的其他部分发生了其他事情。

我真的希望您发布实际存在问题的生产代码，而不是没有问题的演示代码。请问可以吗？

【参考方案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 中显式删除一项且仅一项。