Conexão Persistente com RabbitMQ em Aplicações .NET

Quando estamos desenvolvendo aplicações distribuídas, não devemos nos perguntar se teremos problemas de conectividade. No lugar disso, devemos nos perguntar quando teremos problemas!

Assuma que você está desenvolvendo uma aplicação que precisa se manter conectada ao RabbitMQ. Entretanto, por alguma razão, a conexão que você estabeleceu falhou (acredite, mais cedo ou mais tarde, ela irá falhar). Como proceder?

Eis uma abordagem interessante, que venho adotando, baseada em um exemplo fornecido pela Microsoft.

public delegate Policy PersisterConnectionPolicyFactory(ILogger<PersisterConnection> logger);

public class PersisterConnection :
    IDisposable
{
    public static readonly PersisterConnectionPolicyFactory DefaultPolicyFactory = (logger) =>
    {
        return Policy.Handle<SocketException>()
            .Or<BrokerUnreachableException>()
            .WaitAndRetry(5, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
                (ex, time) => { logger.LogWarning(ex.ToString()); }
            );
    };

    private readonly IConnectionFactory _factory;
    private readonly ILogger<PersisterConnection> _logger;
    private readonly PersisterConnectionPolicyFactory _policyFactory;

    public PersisterConnection(
        IConnectionFactory factory,
        ILogger<PersisterConnection> logger,
        PersisterConnectionPolicyFactory policyFactory
        )
    {
        _factory = factory;
        _logger = logger;
        _policyFactory = policyFactory ?? DefaultPolicyFactory;
    }

    private IConnection _connection;
    public bool IsConnected => 
        _connection != null && 
        _connection.IsOpen && 
        !Disposed;

    private readonly object _lockObject = new object();
        

    public bool TryConnect()
    {
        _logger.LogInformation("RabbitMQ Client is trying to connect");

        lock (_lockObject)
        {
            if (IsConnected) return true;

            var policy = _policyFactory(_logger);
            policy.Execute(() =>
            {
                _connection = _factory.CreateConnection();
            });

            if (!IsConnected)
            {
                _logger.LogCritical("FATAL ERROR: RabbitMQ connections could not be created and opened");
                return false;
            }

            WatchConnectionHealth();

            _logger.LogInformation($"New connection to {_connection.Endpoint.HostName}.");

            return true;
        }
    }

    private void WatchConnectionHealth()
    {
        _connection.ConnectionShutdown += (sender, e) =>
        {
            if (Disposed) return;
            _logger.LogWarning("A RabbitMQ connection is shutdown. Trying to re-connect...");
            TryConnect();
        };

        _connection.CallbackException += (sender, e) =>
        {
            if (Disposed) return;
            _logger.LogWarning("A RabbitMQ connection throw exception. Trying to re-connect...");
            TryConnect();
        };

        _connection.ConnectionBlocked += (sender, e) =>
        {
            if (Disposed) return;
            _logger.LogWarning("A RabbitMQ connection is blocked. Trying to re-connect...");
            TryConnect();
        };
    }

    public IModel CreateModel()
    {
        if (!IsConnected)
        {
            throw new InvalidOperationException(
                "There are RabbitMQ connections available to perform this action"
                );
        }

        return _connection.CreateModel();
    }

    public void Dispose()
    {
        if (Disposed) return;
        Disposed = true;

        try
        {
            _connection.Dispose();
        }
        catch (IOException ex)
        {
            _logger.LogCritical(ex.ToString());
        }
    }
    public bool Disposed { get; private set; }
}

Essa implementação fica “monitorando” a sanidade da conexão com RabbitMQ. Caso ela seja perdida, uma nova conexão é criada. Observe que estou usando Polly para implementação do Retry Pattern. Especial atenção para o delegate utilizado para demandar uma Factory.

UPDATE 1: Luiz Carlos Faria, no Facebook, me alertou para o fato do próprio driver do RabbitMQ oferecer uma opção para recuperação automática. Entretanto, esteja atento a algumas decisões de design do driver caso resolva adotar essa opção.

Se tiver interesse em entender mais sobre microsserviços, recomendo que acesse o Guia de Conteúdo para Microsserviços deste site.

Capa: Gustavo Zambelli

Compartilhe este insight:

2 respostas

  1. E ai, esta seria uma boa estratégia para manter persistência de conexão também para bancos relacionais como MSSQL ou MySql? Ou seja, seria uma boa estratégia para qualquer conexão persistente?

Deixe uma resposta

O seu endereço de e-mail não será publicado. Campos obrigatórios são marcados com *

Elemar Júnior

Sou fundador e CEO da EximiaCo e atuo como tech trusted advisor ajudando diversas empresas a gerar mais resultados através da tecnologia.

Elemar Júnior

Sou fundador e CEO da EximiaCo e atuo como tech trusted advisor ajudando diversas empresas a gerar mais resultados através da tecnologia.

Mais insights para o seu negócio

Veja mais alguns estudos e reflexões que podem gerar alguns insights para o seu negócio:

Criar uma visualização para o fluxo de trabalho é uma das primeiras recomendações do Kanban e tem uma série de...
For years, I have known developers who designed beautiful architectures. A lot of them are questioning the need for a...
Compete ao arquiteto corporativo a responsabilidade de descrever, analisar e comunicar os diversos aspectos da arquitetura corporativa, bem como registrar...
In this post, let’s talk about how to implement Value Types correctly and improve the performance of your applications. The...
Como você pode se considerar um profissional se você não sabe se todo seu código funciona? Como você pode saber...
When designing systems that need to scale you always need to remember that [tweet]more computing power does not necessarily mean...