ConcurrentBag 和多线程应用程序的潜在竞争条件

Question

在过去的几个月里，我一直在思考如何改进我使用 DispatcherTimer 定期检查资源以查看是否需要更新/处理的流程。更新资源（“产品”）后，将产品移至流程中的下一步等。资源可能立即可用，也可能不会立即可用。

我一直在挣扎的原因有两个。一个原因是我想异步实现这个过程，因为它目前只是同步的。第二个原因是我已经确定了我的实现被卡住的区域，这似乎不是一个不常见的设计模式，但我不知道如何简洁地描述它，所以我不知道如何从谷歌获得有用的答案.

一个相当重要的注意事项是，我通过直接 USB 连接访问这些产品，因此我使用 LibUsbDotNet 与设备交互。我已将 USB 连接设为异步，因此我可以同时连接到多个产品并一次处理任意数量。

public Class Product
{
 public bool IsSoftwareUpdated = false;
 public bool IsProductInformationCorrect = false;
 public bool IsEOLProcessingCompleted = false;

 public Product(){}
 ~Product()
}

public class ProcessProduct
{
 List<Product> bagOfProducts                   = new List<Product>(new Product[10]);

 ConcurrentBag<Product> UnprocessedUnits       = new ConcurrentBag<Product>();
 ConcurrentBag<Product> CurrentlyUpdating      = new ConcurrentBag<Product>();
 ConcurrentBag<Product> CurrentlyVerifyingInfo = new ConcurrentBag<Product>();
 ConcurrentBag<Product> FinishedProcessing     = new ConcurrentBag<Product>();

 DispatcherTimer _timer = new DispatcherTimer();

 public ProcessProduct()
 {
     _timer.Tick += Timer_Tick;                            //Every 1 second, call Timer_Tick
     _timer.Interval = new TimeSpan(0,0,1);                //1 Second timer
     
     bagOfProducts.ForEach(o => UnprocessedUnits.Add(o));  //Fill the UnprocessedUnits with all products
 
     StartProcessing();
 }
 private void StartProcessing()
 {
     _timer.Start();
 }

 private void Timer_Tick(object sender, EventArgs e)
 {
     ProductOrganizationHandler();

     foreach(Product prod in CurrentlyUpdating.ToList())
     {
         UpdateProcessHandler(prod);  //Async function that uses await
     }
     foreach(Product prod in CurrentlyVerifyingInfo.ToList())
     {
         VerifyingInfoHandler(prod);  //Async function that uses Await
     }
     if(FinishedProcessing.Count == bagOfProducts.Count)
     {
         _timer.Stop();  //If all items have finished processing, then stop the process
     }
 }
 
 private void ProductOrganizationHandler()
 {
     //Take(read REMOVE) Product from each ConcurrentBag  1 by 1 and moves that item to the bag that it needs to go
     //depending on which process step is finished
     //(or puts it back in the same bag if that step was not finished).
     //E.G, all items are moved from UnprocessUnits to CurrentlyUpdating or CurrentlyVerifying etc.
     //If a product is finished updating, it is moved from CurrentlyUpdating to CurrentlyVerifying or FinishedProcessing
 }
 private async void UpdateProcessHandler(Product prod)
 {
     await Task.Delay(1000).ConfigureAwait(false);
     //Does some actual work validating USB communication and then running through the USB update
 }
 private async void VerifyingInfoHandler(Product prod)
 {
     await Task.Delay(1000).ConfigureAwait(false);
     //Does actual work here and communicates with the product via USB
 }
}

可通过my code on Pastebin 获得完整的可编译代码示例。

所以，我的问题真的是：这段代码中是否有任何有意义的竞争条件？具体来说，使用 ProductOrganizationHandler() 代码和循环通过 Timer_Tick() 中的 ConcurrentBags（因为每次都会发生对 Timer_Tick() 的新调用第二）。我确信这段代码大部分时间都可以工作，但我担心以后会因为罕见的竞争条件而发生难以跟踪的错误，例如 ProductOrganizationHandler() 需要> 出于某种愚蠢的原因运行 1 秒。

作为次要说明：这甚至是此类流程的最佳设计模式吗？ C# 是我的第一门 OOP 语言，并且在工作中都是自学的（我几乎所有的工作都是嵌入式 C），所以我没有任何 OOP 设计模式的正式经验。

我的主要目标是在每个设备通过 USB 可用时异步更新/验证/通信。一旦列表中的所有产品都完成（或超时），则该过程完成。该项目在 .NET 5 中。

编辑：对于后来提出相同问题的任何人，这就是我所做的。

我不明白 DispatcherTimer 将 Ticks 添加到 Dispatcher queue。这意味着只有在没有另一个 Tick 实例正在运行时才会运行一个刻度，或者换句话说，Timer_Tick 将在下一个 Timer_Tick 实例运行之前运行完成。

因此，我所关心的线程/并发问题中的大多数（全部？）都是没有根据的，我可以将 Timer_Tick 视为单线程非并发函数（确实如此）。

另外，为了防止 Ticks 堆积，我在Timer_Tick 的开头运行了_timer.Stop()，并在Timer_Tick 的末尾重新启动了计时器。

Answer 1

首先，您使用的是DispatchTimer，这将在 UI 线程上引发滴答声。据我所知，示例中没有多线程。还有其他计时器，如System.Timers.Timer，如果这是意图，则会在后台线程上引发事件。但是，如果您只是想经常检查和更新状态，并且不运行任何阻塞的代码，那么只需使用 UI 线程就可以了，并且会大大简化事情。

即使我们假设ProductOrganizationHandler 确实在工作线程上运行，从一个并发集合中删除项目并将它们放入另一个集合中通常仍然是安全的。但它不能保证项目以任何特定的顺序处理，也不保证任何特定的项目都由给定的计时器滴答声处理。但是由于计时器会定期滴答，所有的项目最终都应该被处理。请记住，大多数计时器都需要被释放，因此您需要以某种方式处理它，包括处理过早停止的情况。

请记住，async 并不意味着并发，因此除非您的 USB 库提供异步方法，否则我不会使用它。即使那样我也会避免async void，因为这会促进捕获的同步上下文的异常，可能会导致应用程序崩溃，因此它应该主要用于最外层，如按钮事件处理程序或计时器，然后您可能应该以某种方式处理异常.

至于最好的方法，我会看看DataFlow library。