消息队列服务重新启动后服务没有收到消息

Question

我们有一个从n个消息队列接收消息的服务。但是，如果重新启动消息队列服务，则即使消息队列服务已成功重新启动，消息检索服务也会停止接收消息。

我试图专门捕获消息检索服务中引发的MessageQueueException，并再次调用队列的BeginReceive方法。但是，在2秒左右的时间内，消息队列服务才会重新启动，我得到大约1875个异常实例，然后在我们的StartListening方法中引发另一个MessageQueueException时服务停止运行。

有没有一个优雅的方式来从消息队列服务重新启动恢复？

private void OnReceiveCompleted(object sender, ReceiveCompletedEventArgs e) 
    { 
     MessageQueue queue = (MessageQueue)sender; 

     try 
     { 
      Message message = queue.EndReceive(e.AsyncResult); 

      this.StartListening(queue); 

      if (this.MessageReceived != null) 
       this.MessageReceived(this, new MessageReceivedEventArgs(message)); 
     } 
     catch (MessageQueueException) 
     { 
      LogUtility.LogError(String.Format(CultureInfo.InvariantCulture, StringResource.LogMessage_QueueManager_MessageQueueException, queue.MachineName, queue.QueueName, queue.Path)); 
      this.StartListening(queue); 
     }    
    } 

    public void StartListening(MessageQueue queue) 
    { 
     queue.BeginReceive(); 
    } 

我需要处理这个问题导致的无限循环问题，并清理它一下，但你明白了。

发生MessageQueueException时，调用RecoverQueue方法。

private void RecoverQueue(MessageQueue queue) 
    {    
     string queuePath  = queue.Path; 
     bool queueRecovered = false; 

     while (!queueRecovered) 
     { 
      try 
      { 
       this.StopListening(queue); 
       queue.Close(); 
       queue.Dispose(); 

       Thread.Sleep(2000); 

       MessageQueue newQueue = this.CreateQueue(queuePath); 

       newQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(this.OnReceiveCompleted); 

       this.StartListening(newQueue); 

       LogUtility.LogInformation(String.Format(CultureInfo.InvariantCulture, "Message queue {0} recovered successfully.", newQueue.QueueName)); 

       queueRecovered = true; 
      } 
      catch (Exception ex) 
      { 
       LogUtility.LogError(String.Format(CultureInfo.InvariantCulture, "The following error occurred while trying to recover queue: {0} error: {1}", queue.QueueName, ex.Message));     
      } 
     }   
    } 

    public void StopListening(MessageQueue queue) 
    { 
     queue.ReceiveCompleted -= new ReceiveCompletedEventHandler(this.OnReceiveCompleted);    
    } 

Answer 1

一旦接收到异常是服务重新启动的结果，你要释放旧MessageQueue，即您解除wire事件ReceiveCompleted，处置的MessageQueue等，然后创建MessageQueue和挂钩的新实例在新的MessageQueue实例上再次达到ReceiveCompleted事件。

或者，也可以使用在一定的时间间隔创建一个新的实例的轮询方法，调用MessageQueue.Receive(TimeSpan)，将等待一个传入消息或直到超时发生。在这种情况下，您处理消息并销毁实例并再次开始迭代。

通过每次重新创建MessageQueue，确保内置恢复。此外，创建MessageQueue的开销由于底层队列的内部缓存而很小。

伪代码...

while (!notDone)// or use a timer or periodic task of some sort... 
{ 
    try 
    { 
     using (MessageQueue queue = new MessageQueue(queuePath)) 
     { 
      Message message = queue.Receive(TimeSpan.FromMilliseconds(500)); 

      // process message 
     } 
    } 
    catch (MessageQueueException ex) 
    { 
     // handle exceptions 
    } 
}