Все, что нужно знать об обмене сообщениями через службу шины Azure (часть 1)

Те зависимости, которые нам нужно реализовать, следующие:

• Пространство имен служебной шины (используется Uri служебной шины)
• Поставщик токенов
• Сервис Автобус Сервис Uri

Пространство имен служебной шины — это именно то, как оно звучит, и в конечном итоге определяет наше личное пространство имен в рамках нашей службы служебной шины. Эта первая зависимость должна быть создана на портале Azure в разделе «Шина обслуживания».

Вам нужно будет указать пространство имен (запомните это), в котором будет известна ваша конечная точка обслуживания служебной шины, а также такую ​​информацию, как регион, в котором она будет размещена. Наконец, есть важное замечание Базовый уровень против стандартного уровня обмена сообщениями. Для понимания каждого из них вы можете ознакомиться с этой документацией Azure и повлиять на возможность размещения тем и подписок.

Создав пространство имен, мы можем сосредоточиться на двух последних зависимостях; Поставщик токенов и сервисный автобус Uri. Поставщики токенов предоставляют механизм аутентификации, который  NamespaceManagerбудет использоваться. Есть несколько готовых провайдеров, которые можно использовать, и мы рассмотрим некоторые из них более подробно, когда мы перейдем к безопасности. Сейчас мы собираемся использовать провайдер токенов подписи общего доступа и указать имя и ключ политики подписи по умолчанию (SAS) по умолчанию. Эти две части информации можно найти в разделе «Настройка» недавно созданного пространства имен служебной шины на портале Azure. С помощью имени и ключа политики мы можем создать Token Provider:

TokenProvider tokenProvider =
TokenProvider.CreateSharedAccessSignatureTokenProvider(AccountInfo.PolicyName,
AccountInfo.Key);

ПРИМЕЧАНИЕ . AccountInfoПросто введите правильное имя и ключ политики подписи общего доступа.

Uri служебной шины предоставляет NamespaceManagerконечную точку службы, с которой будут работать операции. Здесь будет использоваться наше ранее созданное пространство имен. Конечная точка службы, такая как, mynamespace.servicebus.windows.net может быть создана с использованием одного из статических методов из класса ServiceBusEnvironment и с указанием нашего пространства имен и протокола.

//”sb” defines the scheme of the service Uri
Uri serviceBusUri = ServiceBusEnvironment.CreateServiceUri("sb", "shopit", string.Empty);

С помощью инструментов Visual Studio Azure многие шаблоны проектов Azure Service Bus с помощью пакета NuGet конфигурации Azure Service Bus. Но понимание этих базовых зависимостей поможет облегчить вашу собственную реализацию при необходимости.

Создание менеджера пространства имен

Зависимости в руках, теперь мы можем генерировать наши  NamespaceManager:

NamespaceManager namespaceManger = new NamespaceManager(serviceBusUri, tokenProvider);

По желанию, существует также переопределение, которое позволит нам указать объект NamespaceManagerSettings, который позволит нам дополнительно указать дополнительные спецификации, такие как периоды времени ожидания операции, а также политику повторных попыток (то есть политику экспоненциальной повторной попытки или настраиваемую политику повторных попыток).

NamespaceManagerSettings settings = new NamespaceManagerSettings {OperationTimeout = new TimeSpan(0, 1, 0), TokenProvider = tokenProvider};
NamespaceManager namespaceManger = new NamespaceManager(serviceBusUri, settings);

Асинхронный подход

Одной из лучших рекомендаций, на которые часто ссылаются, является использование асинхронного подхода к разработке для операций вашей служебной шины. Это легко сделать, воспользовавшись асинхронными методами, предоставленными вам для большинства ваших операций. Вы обнаружите, что библиотека предоставляет вам 2 различных асинхронных шаблона, из которых можно выбрать модель асинхронного программирования или асинхронный шаблон на основе задач . Мы поговорим об этом больше, когда перейдем к рекомендациям. Чтобы попытаться сохранить верность лучшему опыту, а также тому факту, что большинство демонстраций проводятся синхронно, здесь большинство примеров будет доставляться асинхронно (каламбур)!

Очереди служебной шины

QueueDescription queueDescription = await namespaceManager.CreateQueueAsync(queueName);

QueueDescription newQueueDescription = null;
if (!await _namespaceManager.QueueExistsAsync(queueName))
{
    newQueueDescription = await _namespaceManager.CreateQueueAsync(queueName);
}

QueueDescription queueDescription = newQueueDescription ?? await _namespaceManager.GetQueueAsync(queueName);

QueueDescription newQueueDescription = new QueueDescription("TestOperationQueue")
{
    DefaultMessageTimeToLive = TimeSpan.FromDays(3),
    AutoDeleteOnIdle = TimeSpan.FromHours(1),
    MaxSizeInMegabytes = 2048
};

QueueDescription  queueDescription = await _namespaceManager.UpdateQueueAsync(newQueueDescription );

_namespaceManager.DeleteQueueAsync(queuePath);

Sending Messages

The connection string is parsed to supply the Service Bus Uri (endpoint) and Token Provider (shared access signature) for the MessageFactory when we utilize the FromConnectionString method.

MessagingFactory factory = MessagingFactory.CreateFromConnectionString(AccountInfo.ConnectionString);
MessageSender messageSender = factory.CreateMessageSender(QueueName);

With a MessageSender we are prepared to send messages

//Some object
SingingTelegram singingTelegram = new SingingTelegram
{
    RecipientFirstName = "Elvis",
    RecipientLastName = "Presley",
    SongName = "Won't you come back again?"
};

BrokeredMessage message = new BrokeredMessage(singingTelegram);

await messageSender.SendAsync(message);

But hold on!

What Kind of Message Are You Sending?

Before we can exercise the “distribute” in Distributed System and start flinging messages at our queue, we need to have an understanding of what constitutes a message.  In our case, its simply.  We’re sending a BrokeredMessage.

But there is allot to understand about a BrokeredMessage that can be broken down to characteristics and features.  Allot of the features will be discussed under specifics areas, but for now I wanted to stop and discuss some of the characteristics you need to be aware of when it comes to BrokeredMessages.

First off, a message at its heart is made up of a body and properties. The total max size of a message is 256kb.  The max size of all properties is 64kb.  The max size of the body is the remainder of the max size of the message and current size of the properties.  There are ways to increase the max size (i.e. sessions) but were not discussing that at the moment.

Body

The body is simply any serializable object and would constitute the payload of the message.  Whether you need to use the body of the message is up to you.  You will see that there are a number of different BrokeredMessage constructors.  These are generally all setting the body of the message as in:

BrokeredMessage msg1 = new BrokeredMessage("My Message Body");
BrokeredMessage msg2 = new BrokeredMessage(new SingingTelegram());

Properties

Properties are made up of two parts, system and custom properties.  The custom properties are simply a key/value pair collection that can be any string / object.  They allow us to transport specific custom properties at the header level of our message.  However, custom properties really come into play when we get to the discussion of subscription rules when we get to Topics and Subscriptions.

Another way to think about message properties in general is when discussing a message flowing between different protocols (HTTP/S, AMQP, SMB).  A message originating from HTTP/S request would end up mapping request headers to message properties.

When we want to promote data to the header level of a message we can utilize the custom Properties property as easily as:

BrokeredMessage message = new BrokeredMessage(singingTelegram);

foreach (object info in importantInformation)
{
    message.Properties.Add(info.Key, info.Value);
}

There are a considerable amount of important system properties on a BrokeredMessage that you can adjust.  Many of these we’ll touch on in specific features.  However, a few worth mentioning are:

1. ScheduledEnqueueTimeUtc – Ifyou wanted to delay the visibility of a message on a queue, you can set a future UTC time in which the message will be unavailable until the specified UTC time.

2. ExpiresAtUtc – If set, will be a specific UTC time in which the message will expire.

3. TimeToLive – Similar to ExpiresAtUtc, this is an amount of time that must pass after the message has been enqueued that the message will expire once.  Both this property and the ExpiresAtUtc directly affect when a message is expired, one at a set time, while the other after an amount of time.

Again, this is not an exhaustive list and it would benefit you to be aware of some of the other influential properties such as how MessageId is used for message duplication, or optional properties such as Label and ContentType can be used.

Receiving Messages

To generate a MessageReceiver, we would need a MessagingFactory and you can refer back to Sending Messages for reference.

MessageReceiver messageReceiver = 
await _messagingFactory.CreateMessageReceiverAsync(queueName);

From here we can start to receive messages.  We can utilize the Timespan parameter of the ReceiveAsync to specify how long to poll before giving up.  In addition, the following example will do a continuous loop until the cancelation token is canceled.

while (!cancellationToken.IsCancellationRequested)
{
    //Wait up to 1 minute for a message
    var msg = await messageReceiver.ReceiveAsync(TimeSpan.FromMinutes(1)).ConfigureAwait(false);
    await ProcessAndReleaseMessageAsync(msg);
    await Task.Delay(TimeSpan.FromMinutes(1), cancellationToken);
}

private async Task ProcessAndReleaseMessageAsync(BrokeredMessage message)
{
    MessageProcessingAction action = MessageProcessingAction.Abandon;
    try
    {
        //Process message
        action = MessageProcessingAction.Complete;
        await UpdateMessageState(message, action);
    }
    catch (Exception ex)
    {
        //log
    }
    finally
    {
        //C# 6.0 allows await calls in a finally blocks
        UpdateMessageState(message, action);
    }
}

private async Task UpdateMessageState(BrokeredMessage message, MessageProcessingAction action)
{
    switch (action)
    {
        case MessageProcessingAction.Complete:
            await message.CompleteAsync();
            break;
        case MessageProcessingAction.Abandon:
            await message.AbandonAsync();
            break;
        case MessageProcessingAction.Deadletter:
            await message.DeadLetterAsync();
            break;
        default:
            await message.AbandonAsync();
            break;
    }
}

Simply put, we wait up to 1 min for a message to be pulled from the queue.  Upon successfully pulling a message, the message is processed and then we explicitly notifying Azure Service Bus with the state of the message by calling CompleteAsync or AbandonAsync.  This is because of a default mode of the Queue where messages received are received in a PeekLock mode.

PeekLock specifies that a message is only temporarily checked out for some duration of time to be processed.  Unless the message is checked in as complete will it be removed from the queue.  There are other consequences that would directly affect a message being removed from the queue such as time to live setting of the queue or message.  But as far as message handling by our receiver, we will need to complete, abandon or dead-letter the message.  We’ll talk more about dead-letter queue and moving messages to there later.

As far as the example above, a couple of points; obviously, a switch statement could be more elegantly refactored, but would also obscure what I wanted to demonstrate about updating the message state so that the queue can properly handle it.  Whether that means leaving it on the queue in the case where our attempt to process it failed, or setting it to be complete so the queue removes it.

PeekLock is not the only mode that messages can be received.  Let’s look at message receive modes as well as other Queue behaviors and features I have withheld up until now.

Dealing with Queues, Again!

I have purposely tried not to over complicate queues with all the different features.  But before closing out Part 1 on Service bus, I did want to return to talk about a few important points that are worth noting.

As I already mentioned, there are 2 modes of message retrieval, PeekLock and ReceiveAndDelete (technically there is a 3rd Session + Peek, but not in the scope of this discussion).  We have already discussed PeekLock, so RecieveAndDelete is a less durable option as it removes the message from the queue when it is retrieved.  Therefore, as you might expect, any exception in your message processing that you have not accounted for can cause a loss of that message.  We can change this when creating our MessageReceiver:

MessageReceiver messageReceiver = 
await _messagingFactory.CreateMessageReceiverAsync(queueName, ReceiveMode.ReceiveAndDelete);

As mentioned under the queue properties that we can set the LockDuration of a message when received under the PeekLock mode which dictates how long a message can be retrained by a receiver before being released back on the queue to be processed.  This is at a default of 1 min., but can be set to a max 5 min.  In the case that a message’s LockUntilUtc property shows it is about to expire, we can request a renewal of the message lock:

await msg.RenewLockAsync();

Queues have two modes of delivery, Pull and Forward.  So far we have seen pull when our MessageReceiver pulls on request to the queue to receive a message.  But we can forward messages by setting the ForwardTo property of our queue so that any messages delivered will automatically be forwarded onto another queue.

QueueDescription queueDescription = await namespaceManager.CreateQueueAsync(new QueueDescription(QueueName)
{
    ForwardTo = destinationQueueName
});

Part 1 Conclusion

To recap, we learned allot of the differences that make up the Azure Service Bus, how to manage all aspects of Queues, as well as how to send and receive messages through Queues.  Now, we have already covered allot with the Service Bus, but there is allot more to cover.  Don’t worry though, I’ll let you take a breather and return in Part II to cover topics such as Topics and Subscriptions, Dead Lettering, Auto Forwarding, Transactions, Batch processing, Security and Service Bus best practices.