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