我正在编写一个 Windows 服务来使用 MSMQ 消息。该服务将有高活跃期（80k 消息很快收到）和长时间不活跃（可能几天没有新消息）。
处理消息是非常受网络限制的，所以我从并行性中获得了很大的好处。但是在不活动期间，我不想占用一堆线程等待不会很快到来的消息。
MSMQ 接口似乎非常专注于同步工作流 - 获取一条消息、处理它、获取另一条消息等。我应该如何构建我的代码，以便我可以在高活动期间利用并行性而不是平局在无活动期间启动一堆线程？使用 TPL 的奖励积分。伪代码将不胜感激。
【问题讨论】：
标签： .net-4.0 parallel-processing msmq task-parallel-library
多年来，我已经完成了 MSMQ（包括移动实现）的分配，您对“同步工作流”的描述是正确的。并不是说您不能通过 TPL 获取各种消息信封并在不同的内核上处理它们……限制因素是读/写队列……本质上是串行操作。例如，您不能一次发送 8 条消息（一台具有 8 个内核的计算机）。
我也有类似的需求（不使用 System.Messaging 命名空间），并在我读过 Campbell 和 Johnson 的“Microsoft.NET 并行编程”一书中的帮助下解决了这个问题。
查看他们的“并行任务”一章，特别是使用全局队列的部分，该队列与每线程本地队列协作进行工作处理（即 TPL），使用“工作窃取”算法来执行负载平衡。我在一定程度上模仿了他们的解决方案。我的系统的最终版本在性能上有很大的不同（从每秒 23 条消息到超过 200 条消息）。
根据您的系统从 0 到 80,000 所需的时间，您需要采用相同的设计并将其分布在多个服务器（每个服务器具有多个处理器和多个内核）上。从理论上讲，我的设置需要不到 7 分钟的时间来磨光所有 80K，因此通过添加第二台计算机，它将减少到大约 3 分 20 秒等，等等等等。诀窍是工作窃取逻辑。
深思……
快速编辑：顺便说一句，这台计算机是戴尔 T7500 工作站，配备双四核 Xeons @ 3GHz、24 GB RAM、Windows 7 Ultimate 64 位版本。
【讨论】：
这是我最终所做的简化版本：
while(true) {
int msgCount = 0;
Parallel.ForEach(Enumerable.Range(0,20), (i) => {
MessageQueue queue = new MessageQueue(_queuePath);
try {
msg = queue.Receive(TimeSpan.Zero);
// do work
Interlocked.Increment(ref msgCount);
catch(MessageQueueException mqex) {
if (mqex.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout) {
return; // nothing in queue
}
else throw;
}
}
if (msgCount < 20)
Thread.Sleep(1000); // nothing more to do, take a break
}
所以我尝试一次接收 20 条消息，并计算我收到的消息。对于那 20 个人，我让 TPL 进城。最后，如果我处理的消息少于 20 条，则队列为空，我将线程休眠一秒钟，然后重试。
【讨论】：
MessageQueue 对象，因为我不相信同时使用来自多个线程的单个对象是安全的。我没有分析所有这些对象创建的成本，但一个简单的优化是每个线程创建一个，而不是每次迭代创建一个。 Parallel.ForEach 对此有一个重载。
Only the GetAllMessages method is thread safe.。你的代码是如何解决这个问题的？
NServiceBus 对这个问题有一个很好的概念。它被称为Distributor。这个想法是分发器可以转发要完成的工作并将其分发到任意数量的运行子节点。取决于正在完成的工作类型，例如繁重的计算与磁盘写入相比，您可以将其分布在多个进程甚至多台机器上。
【讨论】：
解决方案还部分取决于消息的处理方式。
我使用了托管在 Windows Server AppFabric 中的 WorkflowService，并带有 Net.Msmq 绑定和事务队列。需要事务性 net.msmq 绑定来处理乱序消息处理。工作流是一个 .Net 4.0.1 状态机，消息从不同系统进入同一个队列。例如，可以让一个系统在另一个系统发送消息以对其进行实例化之前将更新发送到状态机实例。为了启用乱序消息处理，工作流服务主机使用 BufferedReceive 来锁定消息，并反复尝试从锁定子队列中获取它们。 BufferedReceive 将最大待处理消息设置为可能的最大批处理长度，因为锁定队列中的消息返回到重试队列在前面。
WF 还有一些限制设置。我的最大可能批处理长度约为 20000。我已将 MaxConcurrentCalls 设置为 64，将 MaxConcurrentInstances 设置为 20000。这导致 IIS/WAS 处理 64 个并发调用。
但是，事情就是这样，因为工作流中的接收是单向的，这并不意味着一旦接收完成，衍生的 WF 进程就会终止。在我的场景中接下来发生的事情是，在消息出列并调用 WF 实例（这是 64 个调用之一）之后，工作流引擎会安排许多后续步骤，其中一个是数据库操作。
问题在于，最多 64 次调用可能是最大的，但如果消息消耗率高于异步进程完成率，则在处理传入的一批消息时，执行线程数会更高（在我的案例WF实例）。这可能会导致意外的事情发生，例如 ADO.NET 连接池的最大连接数默认为 100。这将导致进程超时等待来自耗尽池的连接。对于这个特殊问题，您可以提高 MaxPoolSize 值，也可以使用 Service Broker 异步处理数据库操作（这意味着工作流程更加复杂）。
希望这对某人有所帮助。
【讨论】：
我是这样做的（动态更改代码，以免出现拼写错误）：
for (int i = 0; i < numberOfSimultaneousRequests; i++)
priorityQueue.BeginReceive(TimeSpan.FromDays(30), state, callback);
回调看起来像这样：
private void ProcessMessage(IAsyncResult asyncResult)
{
try
{
Message msg = priorityQueue.EndReceive(asyncResult);
//Do something with the message
}
finally
{
priorityQueue.BeginDequeue(null, ProcessMessage);//start processing another one
}
【讨论】：
只是尝试以某种方式相似，tpl 似乎能够在遇到物理问题时抛出某种线程安全异常，例如尝试在 tpl foreach 之外创建一个 sqlconnection 并在循环体中使用它 - 它抛出了一个异常为了我。我在进入正文之前新建了一个队列，枚举了一个字符串列表，看起来还不错，我的代码在 i7 2500 8gb 和本地 msmq 上使用 1way 消息传递始终在 500 毫秒以下处理 10000 个项目
【讨论】：
我在一个名为 CodeRonin 的博客上找到了一个完整的解决方案。在我看来，这是整个互联网上唯一完整的例子。谢谢你，CodeRonin！
http://code-ronin.blogspot.de/2008/09/msmq-transactional-message-processing.html
【讨论】：