2008R2上消息队列的正确使用

Question

我不是程序员，但我试图通过给他们一些指导来帮助他们。我们不再拥有任何有关 msmq 的内部专业知识。我们正在尝试使用它来将一些功能与调度应用程序集成。

调度应用程序通过使用自定义构建的 dll 进行网络调用来启动作业。 dll 调用 weburl。 Web 应用程序将运行其任务并向网站发送有关其执行的任务的更新。网站将消息写入队列。调用该站点的 dll 正在监视队列中是否有带有分配给该作业的标签的消息。当它收到最终状态消息时，它会关闭。

我们每隔几个小时就会收到以下消息。我们每小时运行近 100 个使用此方法的作业。在底部列出的代码中，jobid 对应于消息队列中消息的标签。每个作业在开始时都会发出一个 jobid，并将使用它作为发送到该作业的 msmq 的每条消息的标签。

 System.Messaging.MessageQueueException (0x80004005): Message that the cursor is currently pointing to has been removed from the queue by another process or by another call to Receive without the use of this cursor.
  at System.Messaging.MessageQueue.ReceiveCurrent(TimeSpan timeout, Int32 action, CursorHandle cursor, MessagePropertyFilter filter, MessageQueueTransaction internalTransaction, MessageQueueTransactionType transactionType)
  at System.Messaging.MessageEnumerator.get_Current() 

这是它的代码。

  while ( running )
        {
            // System.Console.WriteLine( "Begin Peek" );
            messageQueue.Peek();
            //System.Console.WriteLine( "End Peek" );
            messageQueue.MessageReadPropertyFilter.SetAll();

            using ( MessageEnumerator enumerator = messageQueue.GetMessageEnumerator2() )
            {
                enumerator.Reset();

                while ( enumerator.MoveNext() )
                {
                    Message msg = enumerator.Current;

                    if ( msg.Label.Equals( this.jobid ) )
                    {
                        StringBuilder sb = new StringBuilder();
                        /*
                        try
                        {
                            sb.Append( "Message Source: " );
                            //sb.Append( msg.SourceMachine );
                            sb.Append( " Sent: " );
                            sb.Append( msg.SentTime );
                            sb.Append( " Label " );
                            sb.Append( msg.Label );
                            sb.Append( " ID: " );
                            sb.Append( msg.Id );
                            sb.Append( " CorrelationID: " );
                            sb.Append( msg.CorrelationId );
                            sb.Append( " Body Type: " );
                            sb.Append( msg.BodyType );
                        }
                        catch ( Exception )
                        {
                            throw;
                        }
                        finally
                        {
                            System.Console.WriteLine( sb.ToString() );
                        }
                        */
                        //System.Console.WriteLine( "Receiving Message started" );
                        using ( Message message = messageQueue.ReceiveById( msg.Id ) )
                        {
                            //System.Console.WriteLine( "Receiving Message Complete" );
                            //sb = new StringBuilder();
                            string bodyText = string.Empty;

                            try
                            {
                                System.IO.StringWriter sw = new System.IO.StringWriter( sb );
                                System.IO.StreamReader sr = new System.IO.StreamReader( message.BodyStream );

                                while ( !sr.EndOfStream )
                                {
                                    sw.WriteLine( sr.ReadLine() );
                                }
                                sr.Close();
                                sw.Close();
                                bodyText = ( string ) FromXml( sb.ToString(), typeof( string ) );
                                int indx = bodyText.IndexOf( ',' );
                                string tokens = bodyText.Substring( indx + 1 );
                                indx = tokens.IndexOf( ',' );
                                string command = tokens.Substring( 0, indx );
                                tokens = tokens.Substring( indx + 1 );
                                if ( command.Equals( COMMAND_STARTED ) )
                                {
                                    System.Console.WriteLine( "STARTED " + tokens );
                                }
                                else if ( command.Equals( COMMAND_UPDATE ) )
                                {
                                    System.Console.WriteLine( tokens );
                                }
                                else if ( command.Equals( COMMAND_ENDED_OK ) )
                                {
                                    System.Console.WriteLine( tokens );
                                    System.Console.WriteLine( "WEBJOB: Success" );
                                    finalResults = new FinalResults( 0, 0, "Success" );
                                    running = false;
                                }
                                else if ( command.Equals( COMMAND_ENDED_WARNING ) )
                                {
                                    System.Console.WriteLine( tokens );
                                    System.Console.WriteLine( "WEBJOB: Warning Issued" );
                                    finalResults = new FinalResults( 1, 1, "Warning" );
                                    running = false;
                                }
                                else if ( command.Equals( COMMAND_ENDED_FAIL ) )
                                {
                                    System.Console.WriteLine( tokens );
                                    System.Console.WriteLine( "WEBJOB: Failure" );
                                    finalResults = new FinalResults( 2, 16, "Failure" );
                                    running = false;
                                }
                            }
                            catch ( Exception )
                            {
                                throw;
                            }
                            finally
                            {
                                //System.Console.WriteLine( "Body: " + bodyText );
                            }
                        }
                    }
                }
            }
        }

        return finalResults;
    }

    MessageQueue messageQueue = null;
    string webServiceURL = "";
    Dictionary<string, string> parms = new Dictionary<string, string>();
    string jobid = "NONE";

Answer 1

kprobst 的解释很可能就是正在发生的事情。即使您看到队列中存在此特定消息，如果不同的应用程序（或同一应用程序的不同实例）从此队列中选取一条（任何）消息，也会使光标无效。

从本质上讲，如果多个进程从同一队列中获取数据，则此代码无法正常工作。

Answer 2

这通常意味着您正在 Receiving() 的消息在接收操作完成之前已被其他内容删除。另一个应用程序或与您的代码位于同一进程中的另一个线程使用不同的队列引用。

您是否有可能同时运行两个处理器代码实例（我猜这是一个控制台应用程序）？在相同或不同的机器上？或者其他一些应用程序或工具从队列中删除消息？

.NET 2.0 的预发行版本之一曾经存在一个错误，在某些压力条件下会导致此问题，但据我记得它在发布之前已修复。

Answer 3

由于 MessageQueue 的内部方法 ReceiveCurrent 中存在并发问题，此操作失败。
异常堆栈跟踪显示调用源自 enumerator.Current 行，异常发生在 ReceiveCurrent。 Enumerator.Current 使用“peek”选项调用 ReceiveCurrent。您可以问，当我遇到同样的问题时，我也遇到过，查看如何失败并出现“消息已收到”错误？它只是想查看下一条尚未收到的消息？
答案就在 ReceiveCurrent 代码中，可以在此处查看：
https://referencesource.microsoft.com/# System.Messaging/System/Messaging/MessageQueue.cs,02c33cc512659fd7,references

ReceiveCurrent 首先进行 StaleSafeReceive 调用来查看下一条消息。但是，如果此调用返回它需要更多内存来接收整个消息（带有
“while (MessageQueue.IsMemoryError(status)”在其源代码中)，它分配所需的内存并进行另一个 StaleSafeReceive 调用来获取消息。
这是非常经典的 Win32 API 使用模式，因为它最终是基于 C 的。

这里的问题是，如果在 ReceiveCurrent 内对 StaleSafeReceive 的第一次和第二次调用之间另一个进程或线程“接收”，即从队列中删除该消息，则第二次调用会抛出这个确切的异常。这就是“查看”操作失败的原因。
请注意，它可能是导致异常的枚举器正在扫描的任何消息，而不是正在查找的消息。这就解释了为什么在抛出异常并且方法失败后，具有该作业 ID 的消息仍然存在于队列中。

可以做的是使用 try catch 来保护枚举器。当前调用，如果捕获到此特定异常，则只需使用队列中的下一个可用消息继续枚举即可。

我使用了 Cursor 对象而不是枚举器，但它遇到了同样的问题。但是使用 Cursor 有另一种方法可以降低发生这种情况的风险，即在扫描/查看消息时关闭当前 Queue 对象的 MessagePropertyFilter 的所有不需要的属性，尤其是 Body 属性。因为在查看期间，通常不需要接收正文，但大多数情况下，消息正文会导致重新分配内存，并需要在 ReceiveCurrent 中进行第二个 StaleSafeReceive 调用。
在 peek 调用中直接使用 Cursor 时，仍然需要尝试捕获此异常。