Exemplo Ciclo de Vida da Connection e IModel | RabbitMQ | CSharp

• Exemplo Ciclo de Vida da Connection e IModel | RabbitMQ | CSharp

---

Exemplo de como ficaria

• Injeção de dependência da IConnectionFactory

  services.AddSingleton<IConnectionFactory>(sp =>
  {
      IOptions<AppSettings> options = sp.GetRequiredService<IOptions<AppSettings>>();
  
      return new ConnectionFactory
      {
          Uri = new Uri(options.Value.RabbitMqConnectionString),
          AutomaticRecoveryEnabled = true,
          NetworkRecoveryInterval = TimeSpan.FromSeconds(10),
          ClientProvidedName = "Nome-Cliente"
      };
  });

---

• RabbitMQConnectionManager - classe responsável por fechar a conexão

  • Temos essa classe para gerenciar a connection que deve ser como singleton, pois é thread-safe
    • Documentação RabbitMQ: 🔗 Clique aqui

  public sealed class RabbitMQConnectionManager(IConnectionFactory factory) : IDisposable
  {
      public IConnection Connection { get; } = factory.CreateConnection();
  
      public void Dispose()
      {
          if (Connection.IsOpen)
              Connection.Close();
  
          Connection.Dispose();
      }
  }

• Versão >= 7.x do pacote RabbitMQ.Client | Cuidado

  public sealed class RabbitMqConnectionManager : 
      IGerenciadorConexaoMensageria, IAsyncDisposable
  {
      private IConnection? _connection;
      private IChannel? _channel;
  
      public IChannel Channel =>
          _channel ?? throw new InvalidOperationException("Channel não inicializado.");
  
      public async Task InitializeAsync(CancellationToken cancellationToken)
      {
          if (_connection is not null)
              return;
  
          _connection = await _connectionFactory
              .CreateConnectionAsync(cancellationToken);
  
          _channel = await _connection
              .CreateChannelAsync(cancellationToken);
      }
  
      public async ValueTask DisposeAsync()
      {
          if (_channel is not null)
              await _channel.CloseAsync();
  
          if (_connection is not null)
              await _connection.CloseAsync();
      }
  }
  

• Quando a aplicação realizar o shutdown, ao descartar essa classe, irá chamar o dispose que matará a conexão.

• Injeção de dependência do RabbitMQConnectionManager

      services.AddSingleton<RabbitMQConnectionManager>();

  danger

  Cuidado com a classe acima porque mesmo ela sendo Singleton, o método de inicializar não é e pode ocorrer erros de concorrência se chamado ao mesmo tempo. Ali seria o mais correto adicionar um SemaphoreSlim:

      private readonly SemaphoreSlim _lock = new(1, 1);
  
      public async Task InicializarAsync(CancellationToken ct)
      {
          if (_connection is not null && _channel is not null)
              return;
  
          await _lock.WaitAsync(ct);
          try
          {
              if (_connection is not null && _channel is not null)
                  return;
  
              _connection = await _connectionFactory.CreateConnectionAsync(ct);
              _channel = await _connection.CreateChannelAsync(ct);
          }
          finally
          {
              _lock.Release();
          }
      }
  

• Fluxo do shutdown em um Worker

      SIGTERM / CTRL+C
      ↓
      Host.StopAsync()
      ↓
      BackgroundService.StopAsync()
      ↓
      CancellationToken cancelado
      ↓
      DI Container começa Dispose
      ↓
      RabbitMqConnectionManager.DisposeAsync()
      ↓
      Channel.CloseAsync()
      ↓
      Connection.CloseAsync()

---

• MessagePublisher - Publicador de Mensagens

  public sealed class MessagePublisher(ILogger<MessagePublisher> logger,
                                      RabbitMQConnectionManager gerenciadorConexao) : IMessagePublisher
  {
      private readonly ILogger<MessagePublisher> _logger = logger;
      private readonly RabbitMQConnectionManager _gerenciadorConexao = gerenciadorConexao;
  
      public void Publicar<T>(T mensagem, string nomeFila)
      {
          try
          {
              IConnection conexao = _gerenciadorConexao.Connection;
  
              using IModel channel = conexao.CreateModel();
              IBasicProperties propriedades = channel.ObterConfiguracaoPersistenciaAtivada();
  
              byte[] dadosEmBytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(mensagem));
  
              channel.BasicPublish(exchange: string.Empty,
                                  routingKey: nomeFila,
                                  basicProperties: propriedades,
                                  body: dadosEmBytes);
          }
          catch (Exception ex)
          {
              _logger.LogError(ex,
                  "Erro ao publicar mensagem no RabbitMQ. Queue: {QueueName}",
                  nomeFila);
              throw;
          }
      }
  }

• Injeção de dependência do MessagePublisher

      services.AddScoped<IMessagePublisher, MessagePublisher>();

---