实时批量数据处理答案

【问题标题】：Batch data processing in real time实时批量数据处理
【发布时间】：2011-08-03 08:31:58
【问题描述】：

我的任务是优化线性数据处理例程的性能。以下是对已有内容的概述：

数据来自 UDP 端口，我们有多个侦听器侦听不同的端口并将原始数据写入 SQL Server 数据库（我们将表称为 RawData）。然后我们有一个单线程线性应用程序的多个实例，从 RawData 表中获取原始数据并处理单个数据行。什么处理手段是将原始数据与给定实体的先前接收的数据进行比较，进行计算以计算不同读数的数量，然后为每个单独的数据行调用几个 Web 服务，最后为每个数据添加一条新记录ProcessedData 表中的行。相应的实体记录也在其他表中更新。

我看到问题的方式，它可以分解成更小的部分，我可以利用生产者/消费者模式进行数据处理：生产者的一个线程填充一个共享（阻塞）队列，多个消费者从队列中获取数据行并对它们进行并行处理。消费者完成后，他们将处理后的数据放入另一个共享队列，然后另一个消费者线程（单个）将访问该队列，该线程将执行 SqlBulkCopy 以插入新记录。在此过程中，将有其他共享队列存储实体信息以进行更新，而另一个消费者将获取实体的更新信息并执行更新。

问题是，尽管它看起来很简单，但在我看来它是一种繁琐的方法。我确实觉得有更好的方法来做我正在寻找的事情。关于实现上述生产者/消费者模式的任何建议？或者我应该为我的问题寻找不同的设计模式？

提前致谢

【问题讨论】：

当您说“共享查询”时，您的意思是“共享队列”吗？
是的，我的错。感谢您注意到它。已更正

标签： .net sql-server multithreading design-patterns batch-processing

【解决方案1】：

您提出的解决方案听起来很合理，我一点也不认为它很麻烦。它易于理解、易于实施、有效且高效。它还允许您调整生产者和消费者的数量以实现最佳性能。分解成更小的部分，各部分之间的交流有限。

因此，您拥有的是多个线程（生产者）从 UDP 读取数据并将这些项目存储在共享队列中。称之为RawData 队列。多个消费者从该队列中读取数据、处理项目并将结果放入另一个共享队列。称之为ProcessedData 队列。最后，您有一个读取ProcessedData 队列并将项目存储在数据库中的线程。

.NET BlockingCollection 非常适合。

这可能会有所帮助：Question on C# threading with RFID

【讨论】：

感谢吉姆的回复。是的，我对哪种类型的队列最适合我的需求存在分歧。我正在考虑使用循环队列或双缓冲区，以尽量减少锁定时间。我知道目前我们没有那么多数据要处理，但我一直在考虑可扩展性。当流量增加时，我不想回去重写应用程序。
我对 BlockingCollection 有一点问题。尽管它实现 ConcurrentQueue 作为其集合基础，但我注意到，当我为消费者添加多个线程时，它们不能保证按顺序从队列中获取项目。我尝试了 Parallel.Foreach 和 Parallel.For。他们似乎都不尊重物品的顺序。然后我尝试了Task.StartNew，它似乎尊重这个序列。此外，我需要将消费者的输出存储到另一个需要订购的队列。我可以使用相同的 BlockingCollection 并在其上运行 .OrderBy 吗？还是最好使用 SortedList 并手动处理锁定？
@Dimitri：你在那里触及了很多问题。最好将其作为另一个问题发布。