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 um comentário

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:

When you think about Roslyn source code, you should think about performance-oriented design. I would like to share some performance techniques...
Decidi aprender a programar com R. Aqui está algo que escrevi. ## defining a function makeCacheMatrix <- function(x = matrix())...
Publicado originalmente no meu blog em 2011 (infelizmente, este conteúdo não está mais disponível). Também publiquei no Linkedin. A publicação...
Há pouco menos de um ano, aceitei o desafio de liderar o time de tecnologia e estratégia de negócios da...
Most of my client’s applications code is for parsing, caching, storing, aggregating, protecting and sharing data! It is not the...
Eu sei que sou privilegiado, em última instância, por poder oferecer, através do meu trabalho, algo que a sociedade valoriza....
× Precisa de ajuda?