在2008R2上正確使用Message Queue

Question

我不是程序員，但我試圖通過給他們一些指導來幫助他們。我們在msmq上不再有內部專業知識。我們正在嘗試使用它來將一些功能與調度應用程序集成。

調度應用程序通過使用自定義內置dll進行網絡調用來關閉工作。該dll調用weburl。網絡應用程序將運行其任務並將更新發送到網站，以瞭解其執行的任務。網站將消息寫入隊列。調用該站點的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

這通常意味着接收操作可以完成之前，正在接收的消息正被其他東西刪除。另一個應用程序或另一個線程與您的代碼使用不同的隊列引用在同一個進程中。

是否有可能有兩個處理器代碼實例（我猜這是一個控制檯應用程序）同時運行？在相同或不同的機器上？或者某些其他應用程序或工具從隊列中刪除消息？

曾經有一個.NET 2.0的預發佈版本中存在一個錯誤，它會在某些壓力條件下導致此問題，但據我記憶，它們在發貨前已經修復。

Answer 3

由於MessageQueue的內部方法ReceiveCurrent中存在併發問題，因此失敗。 異常堆棧跟蹤顯示發起於枚舉器的調用。當前行和異常發生在ReceiveCurrent。 Enumerator.Current使用「peek」選項調用ReceiveCurrent。你可以問，當我遇到同樣的問題時，我也遇到了這種情況，如何通過「消息已收到」錯誤消失？它只是試圖偷看還沒有收到的下一封郵件？ 那麼答案就在於ReceiveCurrent代碼，這是avaliable審查這裏： 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的消息在拋出異常並且方法失敗後仍然存在於隊列中。

可以做些什麼是防止enumerator.Current調用try catch，如果這個特定的異常被捕獲，只需繼續枚舉與隊列中的下一個可用消息。

我曾經使用過Cursor對象而不是枚舉器，但它遇到了同樣的問題。但使用Cursor的用法還有另外一種方法來降低這種情況發生的風險，即在掃描/窺視消息時關閉當前Queue對象的MessagePropertyFilter的所有不需要的屬性，尤其是Body屬性。因爲在窺視過程中通常不需要接收主體，但是大多數情況下，消息的主體會導致內存重新分配，並且需要在ReceiveCurrent中調用第二個StaleSafeReceive調用。 對於這種異常，仍然需要使用直接使用Cursor來調用這個異常。