Начальные грабли в работе с Service Bus for Windows Server12.09.2016 15:03

Краткое описание о том, что такое Service Bus for Windows Server вообще

Это реализация Service Bus от Майкрософт, которая очень близка к Windows Azure Service Bus on Windows, но не требующая самого Azure. То есть этакая довольно удобная и продвинутая шина. Умеет предоставлять как стандартные очереди (queue), так и их продвинутый вариант топики (topic), который умеет отдавать одно и то же сообщение на несколько разных своих подписок (subscription). Так как сталкиваться на практике мне удалось только с топиками/подписками, то дальше речь пойдет только о них.

То есть потребители публикуют свои сообщения в топик. Топик передает их во все свои подписки. Подписки, в свою очередь, проверяют сообщения на предмет того, нужны ли они им или нет, сопоставляя их со своим списком правил (фильтр). Все пригодные сообщения далее передаются тем клиентам, кто подписан на эти самые подписки. Причем если на одну и ту же подписку подписано несколько клиентов, то сообщение получит только один из них. Все довольно стандартно.

Первые шаги и первые же грабли

Внимание, тут и далее приведен не продакшн-код, приложенный код лишь призван служить функционирующей иллюстрацией текста

var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString);
var publisher = messageFactory.CreateTopicClient(topicName);
var listener = messageFactory.CreateSubscriptionClient(topicName, subscriptionName);

listener.OnMessage(message => Console.WriteLine($"Message received: {message.GetBody()}"));

var brokeredMessage = new BrokeredMessage("some test message");
publisher.Send(brokeredMessage);

var stopwatch = new Stopwatch();
int messagesToSent = 200;
int messagesProccessed = 0;

listener.OnMessage(message => 
{
    Console.WriteLine($"Message received: {message.GetBody()}");
    messagesProccessed++;

    if (messagesProccessed == messagesToSent)
    {
        stopwatch.Stop();
        Console.WriteLine($"Time passed: {stopwatch.Elapsed}");
    }
});

stopwatch.Start();
for (var i = 0; i < messagesToSent; i ++)
{
    var brokeredMessage = new BrokeredMessage($"Message №{i}");
    publisher.Send(brokeredMessage);
}

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

• PeekLock — сообщение получается, но не удаляется из подписки, а лишь получает лок. Чтобы оно удалилось, клиент должен явно подтвердить его успешную обработку вызвав Commit (). Иначе же по истечению лока или по вызову Abandon () от самого клиента будет произведена попытка доставить это сообщение снова.
• ReceiveAndDelete — сообщение получается и сразу удаляется из подписки. Если обработка пошла не так, повторно это сообщение уже не получить. Зато работает чуть быстрее PeekLock, так как не вешает локи.

listener.OnMessage(message =>
{
    Console.WriteLine($"Message received: {message.GetBody()}");
    messagesProccessed++;

    if (messagesProccessed == messagesToSent)
    {
        stopwatch.Stop();
        Console.WriteLine($"Time passed: {stopwatch.Elapsed}");
    }

    message.Complete(); // Это все, что поменялось от прошлого примера с замером
});

Почему так происходит? Ведь если что-то было бы не так, можно было бы ожидать эксепшна? На самом деле, эксепшн есть. Просто по какой-то не совсем понятной причине, Майкрософт сделали крайне неочевидный способ получения информации об этих самых исключениях.

Метод подписки на сообщения OnMessage принимает необязательный параметр OnMessageOptions, который позволяет подписаться на событие ExceptionReceived. То есть те самые «скрытые исключения».

var onMessageOptions = new OnMessageOptions();
onMessageOptions.ExceptionReceived += (sender, args) => 
    Console.WriteLine($"Exception received: {args.Exception}");

listener.OnMessage(message => 
{
    Console.WriteLine($"Message received: {message.GetBody()}");
    messagesProccessed++;

    if (messagesProccessed == messagesToSent)
    {
        stopwatch.Stop();
        Console.WriteLine($"Time passed: {stopwatch.Elapsed}");
    }

    message.Complete();
}, onMessageOptions); // не забываем передавать onMessageOptions

Предоставленная блокировка недействительна. У нее истек срок действия, либо сообщение уже удалено из очереди…TrackingId:54630ae4–6e4f-4979–8fc8-b66e5314079c_GAPC_BAPC, TimeStamp:24.08.2016 21:20:08

var stopwatch = new Stopwatch();
int messagesToSent = 200;
int messagesProccessed = 0;

var onMessageOptions = new OnMessageOptions
{
    AutoComplete = false // неочевидный параметр
};
onMessageOptions.ExceptionReceived += (sender, args) => 
    Console.WriteLine($"Exception received: {args.Exception}");

listener.OnMessage(message => 
{
    Console.WriteLine($"Message received: {message.GetBody()}");
    messagesProccessed++;

    if (messagesProccessed == messagesToSent)
    {
        stopwatch.Stop();
        Console.WriteLine($"Time passed: {stopwatch.Elapsed}");
    }

    message.Complete();
}, onMessageOptions);

stopwatch.Start();
for (var i = 0; i < messagesToSent; i ++)
{
    var brokeredMessage = new BrokeredMessage($"Message №{i}");
    publisher.Send(brokeredMessage);
}

Резюмируя:

• Практически всегда надо подписываться на onMessageOptions.ExceptionReceived, иначе можно не заметить ряд проблем в работе кода
• Помнить, что по умолчанию сервисбас пытается коммитить сообщения за вас. Чаще всего это поведение стоит отключать

Абстрации и грабли два

Создается вот такого вида класс:

class Listener : IListener
{
    private readonly MessagingFactory _messageFactory;
    private readonly SubscriptionClient _client;

    public event Func OnReceivedAsync;

    public Listener(string connectionString, string topicName, string subscriptionName, ReceiveMode receiveMode)
    {
        _messageFactory = MessagingFactory.CreateFromConnectionString(connectionString);
        _client = _messageFactory.CreateSubscriptionClient(topicName, subscriptionName, receiveMode);

        var onMessageOptions = new OnMessageOptions
        {
            AutoComplete = false
        };

        onMessageOptions.ExceptionReceived += (sender, args) =>
            Console.WriteLine($"Exception received: {args.Exception}");

        _client.OnMessageAsync(bm => OnReceivedAsync?.Invoke(bm.GetBody()), onMessageOptions);
    }
}

var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString);
var publisher = messageFactory.CreateTopicClient(topicName);
            
int messagesToSent = 20;
for (var i = 0; i < messagesToSent; i++)
{
    var brokeredMessage = new BrokeredMessage($"Message №{i}");
    publisher.Send(brokeredMessage);
}

var listener = new Listener(connectionString, topicName, subscriptionName, ReceiveMode.ReceiveAndDelete);
listener.OnReceivedAsync += x =>
{
    Console.WriteLine($"Message received: {x}");
    return Task.FromResult(true);
};

Причем ошибка будет выловлена только в случае подписки на onMessageOptions.ExceptionReceived, если этого не сделать (а не делают это почему-то очень часто), то о наличии проблемы можно узнать только по косвенным и иногда очень трудноуловимым багам в поведении кода.

Что тут не так? Ну, ответ довольно очевиден, и если бы не встречался так часто, я бы о нем, наверное, и не упоминал. Когда в конструкторе абстракции Listener вызывается _client.OnMessageAsync — подписчик уже начинает принимать сообщения. Поэтому ряд из них (в зависимости от того, как далеко разнесены конструктор и подписка на listener.OnReceivedAsync будет пропущен и попадет на пустой OnReceivedAsync?.Invoke, логично возвращая null. Отсюда и NullReferenceException.

Что с этим делать? Самое простое — разнести создание экземпляра и подписку, например так:

class Listener : IListener
{
    private readonly MessagingFactory _messageFactory;
    private readonly SubscriptionClient _client;

    public Listener(string connectionString, string topicName, string subscriptionName, ReceiveMode receiveMode)
    {
        _messageFactory = MessagingFactory.CreateFromConnectionString(connectionString);
        _client = _messageFactory.CreateSubscriptionClient(topicName, subscriptionName, receiveMode);
    }

    public void Subscribe(Func handleMessage)
    {
        var onMessageOptions = new OnMessageOptions
        {
            AutoComplete = false
        };

        onMessageOptions.ExceptionReceived += (sender, args) =>
            Console.WriteLine($"Exception received: {args.Exception}");

        _client.OnMessageAsync(bm => handleMessage(bm.GetBody()), onMessageOptions);
    }
}

var listener = new Listener(connectionString, topicName, subscriptionName, ReceiveMode.ReceiveAndDelete);
listener.Subscribe(x =>
{
    Console.WriteLine($"Message received: {x}");
    return Task.FromResult(true);
});

Теперь потери сообщений при создании класса не происходит.

Резюмируя:

• Подписка на onMessageOptions.ExceptionReceived актуальней актуального
• Помнить, что получение сообщений начинается сразу после вызова client.OnMessageAsync и учитывать это в проектировании абстраций

Грабли номер три

var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString);
var publisher = messageFactory.CreateTopicClient(topicName);
var listener1 = messageFactory.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.ReceiveAndDelete);
var listener2 = messageFactory.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.ReceiveAndDelete);
            
int messagesToSent = 10;
int messagesProccessed = 0;

var onMessageOptions = new OnMessageOptions
{
    AutoComplete = false
};
onMessageOptions.ExceptionReceived += (sender, args) =>
    Console.WriteLine($"Exception received: {args.Exception}");

listener1.OnMessage(message =>
{
    Console.WriteLine($"listener1: message received: {message.GetBody()}, listener1 is closed: {listener1.IsClosed}");
    messagesProccessed++;
}, onMessageOptions);
            
for (var i = 0; i < messagesToSent; i++)
{
    var brokeredMessage = new BrokeredMessage($"Message №{i}");
    publisher.Send(brokeredMessage);
    Thread.Sleep(50);
                
    if (i == 4)
    {
        Console.WriteLine("Closing listener1");
        listener1.Close();
    }
}
            
listener2.OnMessage(message =>
{
    Console.WriteLine($"listener2: message received : {message.GetBody()}, listener2 is closed: {listener2.IsClosed}");
    messagesProccessed++;
}, onMessageOptions);

listener1: message received: Message №0, listener1 is closed: False
listener1: message received: Message №1, listener1 is closed: False
listener1: message received: Message №2, listener1 is closed: False
listener1: message received: Message №3, listener1 is closed: False
listener1: message received: Message №4, listener1 is closed: False
Closing listener1
listener1: message received: Message №5, listener1 is closed: True
listener2: message received: Message №6, listener2 is closed: False
listener2: message received: Message №7, listener2 is closed: False
listener2: message received: Message №8, listener2 is closed: False
listener2: message received: Message №9, listener2 is closed: False

А само пропущенное сообщение, в отличие от ReceiveAndDelete, все таки будет обработано позже, когда произойдет повторная отправка.

var messageFactory = MessagingFactory.CreateFromConnectionString(connectionString);
var messageFactory1 = MessagingFactory.CreateFromConnectionString(connectionString);
var messageFactory2 = MessagingFactory.CreateFromConnectionString(connectionString);
var publisher = messageFactory.CreateTopicClient(topicName);
var listener1 = messageFactory1.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.PeekLock);
var listener2 = messageFactory2.CreateSubscriptionClient(topicName, subscriptionName, ReceiveMode.PeekLock);
            
int messagesToSent = 10;
int messagesProccessed = 0;

var onMessageOptions = new OnMessageOptions
{
    AutoComplete = false
};
onMessageOptions.ExceptionReceived += (sender, args) =>
    Console.WriteLine($"Exception received: {args.Exception}");

listener1.OnMessage(message =>
{
    try
    {
        Console.WriteLine($"listener1: message received: {message.GetBody()}, listener1 is closed: {listener1.IsClosed}");
        messagesProccessed++;
        message.Complete();
    }
    catch (Exception ex1)
    {
        Console.WriteLine($"listener1 Complete() exception: {ex1.Message}");
        try
        {
            message.Abandon();
        }
        catch (Exception ex2)
        {
            Console.WriteLine($"listener1 Abandon() exception: {ex2.Message}");
        }
    }
}, onMessageOptions);
            
for (var i = 0; i < messagesToSent; i++)
{
    var brokeredMessage = new BrokeredMessage($"Message №{i}");
    publisher.Send(brokeredMessage);
    Thread.Sleep(50);
                
    if (i == 4)
    {
        Console.WriteLine("Closing listener1");
        listener1.Close();
    }
}
            
listener2.OnMessage(message =>
{
    Console.WriteLine($"listener2: message received : {message.GetBody()}, listener2 is closed: {listener2.IsClosed}");
    messagesProccessed++;
    message.Complete();
}, onMessageOptions);

Резюмируя:

• Не все экспешны от сервисбаса прилетают только в onMessageOptions.ExceptionReceived
• При завершение работы с подписчиком надо учитывать особенности работы метода Close ()

Заключение

Комментарии (0)