Блог

Я использую служебную шину 1.1 локально, и я создаю WindowsService, который должен просто прослушивать определенную очередь, запущенную на другом компьютере.

Код WindowsService для подключения к SB выглядит следующим образом:

 MessagingFactory _receiverFactory = MessagingFactory.CreateFromConnectionString(sbConnectionString);
MessageReceiver _messageReceiver = _receiverFactory.CreateMessageReceiver(listeningQueue);

BrokeredMessage brokeredMessage = _messageReceiver.Receive(TimeSpan.FromSeconds(60));
  

Все работает нормально, пока я не перезапущу серверы: я заметил, что если WindowsService запускается ДО того, как будут доступны все службы служебной шины, мое приложение вообще никогда не подключится, пока я вручную не перезапущу свой WindowsService.

На самом деле, если SB не работает полностью во время создания MessagingFactory, инструкция приема всегда будет завершаться ошибкой, даже если службы SB возвращаются во второй раз…

Я пытался перехватить исключения (MessagingCommunicationException) и перестроить MessagingFactory, но просто не работает, единственный способ повторного подключения — перезапустить WindowsService.

Есть предложения?

1. Настройте ферму служб высокой доступности («HA»).

https://blogs.technet.microsoft.com/meamcs/2013/12/08/recommended-practices-service-bus-for-windows-server/

http://www.planetgeek.ch/2014/12/10/service-bus-for-windows-server-high-availability/

Но даже если вы делаете или не делаете # 1 «ХА», вы должны сделать # 2.

2. Используйте «Polly» для перехвата и автоматического повторения ОПРЕДЕЛЕННЫХ исключений

    using System;
   using System.Collections.Generic;
   using System.Linq;
   using System.Threading;
   
   using Microsoft.ServiceBus;
   using Microsoft.ServiceBus.Messaging;
   
   using Polly;
   
   private static void KeepReadingTheQueue(QueueClient qc)
   {
       int retries = 0;
       int eventualFailures = 0;
   
       Policy pol = Policy.Handle<System.OperationCanceledException>(ex => null != ex)
           .Or<UnauthorizedAccessException>(ex => null != ex)
           .Or<Microsoft.ServiceBus.Messaging.MessagingCommunicationException>()
           .Or<Microsoft.ServiceBus.Messaging.MessageLockLostException>()
           .WaitAndRetryForever(
           sleepDurationProvider: attempt => TimeSpan.FromMilliseconds(500), // Wait 500ms between each try.
           onRetry: (exception, calculatedWaitDuration, context) => // Capture some info for logging!
           {
   
               ReportDuplicatesOrMissing("Policy.OnRetry", batchNumbers);
   
               // This is your new exception handler! 
               // Tell the user what they've won!
               Console.WriteLine("Log, then retry: "   exception.Message, ConsoleColor.Yellow);
               retries  ;
           });
   
       int i = 0;
   
       int receiveBatchSize = 1; /* i have this in a custom settings class, hard coded here */
   
       while (true)
       {
           i  ;
   
           ReportDuplicatesOrMissing("while.true", batchNumbers);
   
           try
           {
               // Retry the following call according to the policy - 15 times.
               pol.Execute(() =>
               {
                   IEnumerable<BrokeredMessage> messages = null;
                   if (receiveBatchSize <= 1)
                   {
                       // Receive the message from the queue
                       BrokeredMessage receivedMessage = qc.Receive();
                       List<BrokeredMessage> listMsgs = new List<BrokeredMessage>();
                       listMsgs.Add(receivedMessage);
                       messages = listMsgs as IEnumerable<BrokeredMessage>;
                   }
                   else
                   {
                       messages = qc.ReceiveBatch(receiveBatchSize);
                       int count = messages.Count();
                       if (count > 0)
                       {
                           Console.WriteLine("ReceiveBatch.Count='{0}'", count);
                       }
                   }
   
                   foreach (BrokeredMessage receivedMessage in messages)
                   {
                       if (receivedMessage != null)
                       {
                           MyPayLoadObject concreteMsgObject = receivedMessage.GetBody<MyPayLoadObject>();
                           receivedMessage.Complete();
                           Console.WriteLine(@"Message received: {0}", concreteMsgObject);
                       }
                       else
                       {
                           Console.WriteLine(@"No new messages in the queue");
                           Thread.Sleep(1000);
                       }
   
                       Thread.Sleep(100);
                   }
   
               });
           }
           catch (Exception e)
           {
               Console.WriteLine("Request "   i   " eventually failed with: "   e.Message, ConsoleColor.Red);
               eventualFailures  ;
           }
       }
   }
     

packages.config

 <?xml version="1.0" encoding="utf-8"?>
<packages>
  <package id="Microsoft.WindowsAzure.ConfigurationManager" version="2.0.1.0" targetFramework="net45" />
  <package id="Polly" version="4.3.0" targetFramework="net45" />
  <package id="WindowsAzure.ServiceBus" version="2.1.4.0" targetFramework="net45" />
</packages>