Управление несколькими клиентами очереди / темы?

Question

Проблема в том, что я пытаюсь создать единый фасад для работы с очередями и темами, сохраняя семантику Send против Publish

Например:

public interface IServiceBus
{
    Task Send<T>(T message, string destination, SendOptions options = null);
    Task Publish<T>(T message, string topic, SendOptions options = null);
}

Send() отправит сообщение в очередь, а Publish() опубликует сообщение в теме. Поэтому мне нужно иметь экземпляр IQueueClient и ITopicClient, чтобы это произошло; Я вставил бы их в мою IServiceBus реализацию как зависимости и вызвал бы их соответственно.

Проблема в том, что QueueClient и TopicClients требуют, чтобы вы указывали их назначения при обновлении клиентов, что не позволяет мне разрешить это в качестве параметров для моей IServiceBus реализации.

Я мог бы просто создать клиента во время создания сообщения, но это супер неэффективно. Я искал хотя бы диспетчер соединений, который мог бы выступать в роли фабрики для клиентов, но, похоже, MessagingFactory не входит в этот SDK (Microsoft.Azure.ServiceBus 3.4.0).

Так что вопросы - Есть ли какая-то фабрика, которую я могу использовать, которая позволила бы мне создавать подходящих клиентов по требованию с той же эффективностью, которую можно было бы получить, повторно используя клиентов? - Есть ли какой-то тип переопределения или альтернативный объект клиента, который я должен использовать для достижения этого эффекта? Эти два клиента действительно ограничивают.

Answer 1

Поскольку я считаю, что мы можем предположить, что экземпляры QueueClient и TopicClient являются поточно-ориентированными , то, что вы можете сделать, это зарегистрировать разрешенный конкретный класс IServiceBus как синглтон в вашем контейнере IoC.

В конкретном ServiceBus вы можете создать кеш ранее просмотренных клиентов тем и очередей:

private readonly ConcurrentDictionary<string, ITopicClient> _topicClientCache
    = new ConcurrentDictionary<string, ITopicClient>();
private readonly ConcurrentDictionary<string, IQueueClient> _queueClientCache
    = new ConcurrentDictionary<string, IQueueClient>();

Затем в вашем Publish методе

public async Task Publish<T>(T message, string destination, ...)
{
    // i.e. destination is the topic
    if (!_topicClientCache.TryGetValue(destination, out var topicClient))
    {
        topicClient = new TopicClient(_myServiceBusConnectionString, destination);
        _topicClientCache.TryAdd(destination, topicClient);
    }
    ... create and serialize message into azureMessage here
    await topicClient.SendAsync(azureMessage);
}

То же самое относится и кваша реализация Send - она ​​проверит соответствие _queueClientCache пункта назначения (имя очереди), создаст его и кеширует при первом обнаружении.

Answer 2

Мне наконец удалось встретить кого-то, у кого была похожая проблема. Оказывается, они удалили MessagingFactory, но сделали соединение многоразовым. У каждого клиента есть перегрузка конструктора, которая принимает соединение, поэтому я зарегистрировал соединение как одноэлементное и внедрил его вместо клиента, а затем просто создал клиентов по требованию.

см .: https://github.com/Azure/azure-service-bus-dotnet/issues/556

Мое решение выглядело примерно так (полная реализация опущена для краткости)

public class AzureServiceBus : IServiceBus
{
    public AzureServiceBus(ServiceBusConnection connection, string replyTo)
    {
        _connection = connection;
        _replyTo = replyTo;
        _retryPolicy = new RetryExponential(
            TimeSpan.FromSeconds(1),
            TimeSpan.FromMinutes(1),
            10);
    }

    public async Task Send<T>(T message, string destination)
    {
        var client = new QueueClient(_connection, destination, ReceiveMode.PeekLock, _retryPolicy);

        // ... do work
    }

    public async Task Publish<T>(T message, string topic, SendOptions options = null)
    {
        var client = new TopicClient(_connection, topic, _retryPolicy);

        // ... do work
    }
}