Запускаем Windows Azure Service Bus локально.

Всем привет! Совсем недавно в мире Windows Azure произошло очень важное событие, а именно был выпущен набор компонентов, позволяющих эмулировать Windows Azure Service Bus на локальной машине. Именно этой новой возможности и будет посвящен мой сегодняшний пост. Мы узнаем как установить эмулятор Service Bus себе на машину, как сконфигурировать его, а также напишем небольшое приложение для проверки его работоспособности.
Итак, согласно определению из MSDN, Windows Azure Service Bus for Windows Server – это набор инсталлируемых компонентов, предоставляющих возможность эмулирования соответствующей облачной технологии на ваших собственных серверах, с предустановленной ОС Windows Server. Программная модель абсолютна идентична облачной версии поэтому переключиться с локальной на облачную не составит большого труда и Вам не нужно будет переписывать приложение.
Основные возможности:

• Симметричная модель программирования. Написав код один раз для локальной версии его не надо будет менять при переключении на облачную.
• Отлаживать и тестировать приложения стало намного проще так как шина находится локально, что позволяет нам настраивать большинство ее параметров.
• Быть может сейчас Вы не готовы переходить в облако по каким-то причинам (безопасность, юридические аспекты и др.). Эмулятор позволяет начать писать приложение сразу, а перейти на облачную версию тогда, когда вы будете готовы.

Для установки эмулятора можем воспользоваться Web Platform Installer. Компонент носит название Service Bus 1.0 Beta. Кроме установки самих компонентов необходимо также проверить следующие характеристики вашей машины:

• У Вас есть доступ к инстансу SQL Server (Express версия также подходит)
• Сервис SQL Browser должен быть разрешен и запущен
• TCP/IP должен быть разрешен

Как только установка компонентов будет завершена необходимо настроить кластер. Здесь необходимо дать некоторые пояснения. Дело в том, что Service Bus for Windows Server имеет встроенную поддержку высокой доступности сервиса. Достигается это за счет использования кластеров вычислительных машин. Добавив как минимум 3 машины в кластер – Вы получите отказоустойчивое и высокодоступное решение. Схема такого решения представлена чуть ниже.


Следует отметить, что данная схема работает только для вычислительных инстансов. Для слоя хранения данных встроенной поддержки отказоустойчивости нет, необходимо использовать сторонние подходы, например, SQL Server Mirroring. Итак, первоначально нам необходимо создать новый кластер и добавить туда нашу девелоперскую машину. Делается это при помощи следующих PowerShell команд:

$mycert=ConvertTo-SecureString -string <password> -force -AsPlainText
New-SBFarm -FarmMgmtDBConnectionString "data source=localhost\sqlexpress;integrated security=true" –CertAutoGenerationKey $mycert
Add-SBHost -certautogenerationkey $mycert -FarmMgmtDBConnectionString "data source=localhost\sqlexpress; integrated security=true"

Здесь, в первой строке мы определяем пароль, который будет использоваться при генерации сертификата безопасности. Во второй строке мы создаем кластер/ферму, при этом необходимо указать правильный адрес SQL Server и криденшиалы для доступа к нему. Последней командой мы добавляем текущую машину в только что созданный кластер (здесь также следует обратить внимание на строку соединения). Как только все вышеуказанные команды выполняться успешно, на используемом экземпляре SQL Server мы должны увидеть 3 базы данных, которые буду служить как центром конфигурации кластера так и хранилищем данных для сообщений, курсирующих по шине.

Для проверки работоспособности кластера можно запустить специальную команду:

Get-SBFarmStatus

В результате выполнения команды мы должны получить следующую информацию:

Если какой либо из сервисов Service Bus не находится в состоянии Running, значит на предыдущем этапе возникли ошибки. В случае успешного запуска всех сервисов на следующем шаге нам необходимо создать неймспейс (как мы помним в облачной версии этот шаг выполняется на портале). Для этого необходимо выполнить следующую команду:

New-SBNamespace –Name TestSbNamespace –ManageUsers  yourDomain\yourUsername

В параметре –Name мы задаем название создаваемого неймспейса. В параметре –ManageUsers задается список пользователей, которые будут иметь права по управлению создаваемым неймспейсом. Как только команда выполнится успешно мы сможем перейти к написанию своего первого приложения, использующего локальную сервисную шину.

Реализация сценария Sender/Receiver на базе Windows Server Service Bus

Для тестирования напишем простое решение, позволяющее отправлять сообщения и принимать их посредством запущенной локально сервисной шины. Структура проекта относительно проста.


Это 2 консольных приложения.Одно из них используется для отправки сообщений, другое для приема. Давайте посмотрим на исходный код отправителя (Sender.cs).

namespace Microsoft.Samples.MessagingWithQueues
{
    using System;
    using System.Collections.Generic;
    using Microsoft.ServiceBus;
    using Microsoft.ServiceBus.Messaging;
    using System.Net.NetworkInformation;
    using System.Collections.Generic;

    public class Sender
    {
        const string QueueName = "MessagesQueue";
        static string ServiceNamespace;
        static string ServerDomainName = "localhost";
        static int HttpPort = 4446;
        static int TcpPort = 9354; 

        static void Main(string[] args)
        {
            GetUserCredentials();    
            
            Uri rootAddressManagement = ServiceBusEnvironment.CreatePathBasedServiceUri("sb", ServiceNamespace, string.Format("{0}:{1}", ServerDomainName, HttpPort));
            Uri rootAddressRuntime = ServiceBusEnvironment.CreatePathBasedServiceUri("sb", ServiceNamespace, string.Format("{0}:{1}", ServerDomainName, TcpPort));


            MessagingFactorySettings mfSettings = new MessagingFactorySettings();
            mfSettings.TokenProvider = TokenProvider.CreateWindowsTokenProvider(
                           new List<Uri>() { rootAddressManagement });           

            MessagingFactory factory = MessagingFactory.Create(rootAddressRuntime, mfSettings); 


            try
            {     
                NamespaceManagerSettings nmSettings = new NamespaceManagerSettings();               
                nmSettings.TokenProvider = TokenProvider.CreateWindowsTokenProvider(new List<Uri>() { rootAddressManagement });
                
                NamespaceManager namespaceManager = new NamespaceManager(rootAddressManagement, nmSettings);
                Console.WriteLine("\nCreating Queue '{0}'...", Sender.QueueName);

                // Delete if exists
                if (namespaceManager.QueueExists(Sender.QueueName))
                {
                    namespaceManager.DeleteQueue(Sender.QueueName);
                }
                namespaceManager.CreateQueue(Sender.QueueName);
                QueueClient myQueueClient = factory.CreateQueueClient(Sender.QueueName);
                string message = String.Empty;
                do
                {
                    message = Console.ReadLine();
                    myQueueClient.Send(CreateMessage(Guid.NewGuid().ToString(), message));
                } while (!String.IsNullOrEmpty(message));
            }
            catch (Exception e)
            {
                Console.WriteLine("Unexpected exception {0}", e.ToString());
                throw;
            }
            finally
            {
                // Closing factory close all entities created from the factory.
                if(factory != null)
                    factory.Close();
            }
            
        }

        static void GetUserCredentials()
        {
            Console.Write("Please provide the server Fully Quaified Domain Name to use (blank for localhost): ");
            ServerDomainName = Console.ReadLine();
            if (ServerDomainName.Equals("localhost") || String.IsNullOrEmpty(ServerDomainName))
            {
                var ipProperties = IPGlobalProperties.GetIPGlobalProperties();
                ServerDomainName = string.Format("{0}.{1}", ipProperties.HostName, ipProperties.DomainName);
            }
            
            Console.Write("Please provide the service namespace to use: ");
            ServiceNamespace = Console.ReadLine();
        }

        private static BrokeredMessage CreateMessage(string messageId, string messageBody)
        {            
            BrokeredMessage message = new BrokeredMessage(messageBody);
            message.MessageId = messageId;
            return message;
        }
    }
}

В первую очередь для начала отправки сообщений необходимо собрать информацию о месторасположении сервера и неймспейсе. Это происходит в методе GetUserCredentials, где мы запрашиваем DNS-имя сервера (в случае запуска шины локально можно ничего не вводить) и название неймспейса (в нашем случае это будет TestSbNamespace). Как облачная так и серверная версия Service Bus используют access-токены для предоставления доступа к своим внутренним элементам (subsriptions, topics, queues, entities ...). Но в отличие от серверной версии, облачная тесно интегрирована с Windows Azure Access Control Service поэтому для локальной версии была разработана своя подсистема, называемая Service Bus Security Token Service (SBST). Эта подсистема интегрирована с моделью безопасности Windows и позволяет генерировать токены, базируясь на Windows аккаунтах (поддерживается как локальное хранилище так и Active Directory). Для получения токена от локальной SBST мы используем метод TokenProvider.CreateWindowsTokenProvider, принимающий список защищаемых неймспейсов. В нашем случае нам необходимо сгенерировать токен дважды: для объекта класса MessagingFactory и для NamespaceManagerSettings. Для отправки сообщения в очередь необходимо создать объект класса BrokeredMessage и передать его методу Send, который доступен у объекта класса QueueClient.
Если в процессе написания кода вы столкнетесь в проблемой отсутствующих сборок, то скорее всего Вам необходимо будет инсталлировать пакет под названием Service Bus 1.0 Beta (for Windows Server) – делается это из окна пакетного менеджера Nuget

Код обработчика сообщений очень похож на код отправителя  за исключением того, что здесь мы входим в бесконечный цикл ожидания сообщений и для их получения из очереди используем метод Receive. Полностью код обработчика сообщений представлен ниже:

namespace Microsoft.Samples.MessagingWithQueues
{
    using System;
    using Microsoft.ServiceBus;
    using Microsoft.ServiceBus.Messaging;
    using System.Net.NetworkInformation;
    using System.Collections.Generic;
    using System.Threading;

    public class Receiver
    {
        const string QueueName = "MessagesQueue";
        static string ServiceNamespace;
        static string ServerDomainName = "localhost";
        static int HttpPort = 4446;
        static int TcpPort = 9354; 

        static void Main(string[] args)
        {
            GetUserCredentials();

            Uri rootAddressManagement = ServiceBusEnvironment.CreatePathBasedServiceUri("sb", ServiceNamespace, string.Format("{0}:{1}", ServerDomainName, HttpPort));
            Uri rootAddressRuntime = ServiceBusEnvironment.CreatePathBasedServiceUri("sb", ServiceNamespace, string.Format("{0}:{1}", ServerDomainName, TcpPort));


            MessagingFactorySettings mfSettings = new MessagingFactorySettings();
            mfSettings.TokenProvider = TokenProvider.CreateWindowsTokenProvider(
                           new List<Uri>() { rootAddressManagement });

            MessagingFactory factory = MessagingFactory.Create(rootAddressRuntime, mfSettings); 

            try
            { 
                QueueClient myQueueClient = factory.CreateQueueClient(Receiver.QueueName, ReceiveMode.PeekLock);
                Console.WriteLine("\nReceiving messages from Queue '{0}'...", Receiver.QueueName);

                while (true)
                {
                    var message = myQueueClient.Receive();
                    if (message != null)
                    {
                        Console.WriteLine(String.Format("Message received: Id = {0}, Body = {1}", message.MessageId, message.GetBody<string>()));
                        message.Complete();
                    }
                }
            }
            catch (Exception e)
            {
                Console.WriteLine("Unexpected exception {0}", e.ToString());
                throw;
            }
            finally
            {
                // Closing factory close all entities created from the factory.
                if (factory != null)
                    factory.Close();
            }
        }

        static void GetUserCredentials()
        {
            Console.Write("Please provide the server FQDN to use (blank for localhost): ");
            ServerDomainName = Console.ReadLine();
            if ((string.Compare(ServerDomainName, "localhost", true) == 0) || string.IsNullOrEmpty(ServerDomainName))
            {
                var ipProperties = IPGlobalProperties.GetIPGlobalProperties();
                ServerDomainName = string.Format("{0}.{1}", ipProperties.HostName, ipProperties.DomainName);
            }

            Console.Write("Please provide the service namespace to use: ");
            ServiceNamespace = Console.ReadLine();
        }
    }
}

Метод Complete необходим для оповещения сервисной шины о том, что сообщение успешно обработано. После вызова этого метода сообщение физически удаляется из очереди. Если в обработчике произойдет сбой во время обработки сообщения, то оно снова станет доступным по истечение определенного промежутка времени. Это один из вариантов функционирования очереди, называемый PeekLock (Все варианты получения сообщений описаны на сайте Windows Azure). Сам процесс взаимодействия отправителя и получателя хорошо продемонстрирован на видео:
Как видим разработанное приложение на базе Service Bus for Windows Server работает исправно. Надеюсь информация была представлена в максимально удобном и понятном виде. Если будут вопросы или предложения, пишите в комментариях, будем обсуждать :) Спасибо за внимание! Ссылка на скачивание готового решения: