Usługa nie otrzymuje wiadomości po ponownym uruchomieniu Usługi kolejkowania wiadomości.

Question

Mamy usługę, która odbiera wiadomości od n kolejek wiadomości. Jednak jeśli usługa kolejkowania wiadomości zostanie ponownie uruchomiona, usługa pobierania wiadomości przestanie odbierać wiadomości nawet po pomyślnym ponownym uruchomieniu usługi kolejkowania wiadomości.

Próbowałem specjalnie złapać wyjątek MessageQueueException, który jest zgłaszany w usłudze pobierania wiadomości i ponownie wywołać metodę BeginReceive kolejki. Jednak w ciągu 2 sekund, aby usługa kolejkowania wiadomości została uruchomiona ponownie, otrzymuję około 1875 wystąpień wyjątku, a następnie usługa przestaje działać, gdy inny obiekt MessageQueueException zostanie zgłoszony w naszej metodzie StartListening.

Czy istnieje elegancki sposób na odzyskanie po ponownym uruchomieniu usługi kolejkowania wiadomości?

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(); 
    } 

Muszę poradzić sobie z kwestią nieskończonej pętli, która to powoduje i trochę ją wyczyść, ale masz pomysł.

Po wystąpieniu wyjątku MessageQueueExue wywołaj metodę 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

Po otrzymaniu wyjątek, który jest wynikiem ponownego uruchomienia usługi, trzeba zwolnić stare MessageQueue, tj unwiring wydarzenie ReceiveCompleted, umieszczając MessageQueue, itd. Następnie należy utworzyć nową instancję MessageQueue i hakiem do zdarzenia ReceiveCompleted ponownie w nowej instancji MessageQueue.

Alternatywnie można użyć metody odpytywania, która tworzy nowe wystąpienie w określonym przedziale czasu, wywołania MessageQueue.Receive(TimeSpan), będą czekać na wiadomość przychodzącą lub do czasu upłynięcia limitu czasu. W takim przypadku zajmiemy się tą wiadomością i zniszczymy instancję MessageQueue i ponownie uruchomimy iterację.

Po odtworzeniu MessageQueue za każdym razem zapewnia wbudowane odzyskiwanie. Ponadto narzut tworzenia MessageQueue jest minimalny z powodu wewnętrznego buforowania podstawowej kolejki.

Pseudo-kod ...

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 
    } 
}