Exemplos de produtor e consumidor | RabbitMQ

• Exemplos de produtor e consumidor | RabbitMQ

---

Propriedades do produtor

danger

Quem deve criar a fila: 🔗 Clique aqui Routing-Key e Binding-Key: 🔗 Clique aqui

static void Main(string[] args)
{
    var factory = new ConnectionFactory() { HostName = "localhost" };

    using (var connection = factory.CreateConnection())
    using (var channel = connection.CreateModel())
    {
     
        string message = "This is sample message";

        var body = Encoding.UTF8.GetBytes(message);

        var properties = channel.CreateBasicProperties();
        properties.Persistent = true; // 🔹 mensagem persistente
        properties.ContentType = "text/plain";
        properties.DeliveryMode = 2; // 2 = persistente
        properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds());

        channel.BasicPublish(
            exchange: "",
            routingKey: "sample",
            basicProperties: properties,
            body: body
        );

        Console.WriteLine($"[x] Sent a message : {message}");
    }

    Console.WriteLine("Press any key to exit.");
    Console.ReadKey();

}

Versão <= 6.x

Propriedades do consumidor

static void Main(string[] args)
{
    var factory = new ConnectionFactory() { HostName = "localhost" };

    using (var connection = factory.CreateConnection())

    using (var channel = connection.CreateModel())
    {
        channel.QueueDeclare(
            queue: "sample",
            durable: false,
            exclusive: false,
            autoDelete: false,
            arguments: null);

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine($"[x] Received a message : {message}");

        };

        channel.BasicConsume(
            queue: "sample",
            autoAck: true,
//A execução de uma tarefa pode levar alguns segundos. Você pode se perguntar o que acontece se um dos consumidores iniciar uma longa tarefa e morrer com ela apenas parcialmente concluída. Com nosso código atual, uma vez que o RabbitMQ entrega uma mensagem ao consumidor, ele imediatamente a marca para exclusão. Nesse caso, se você matar um trabalhador, perderemos a mensagem que estava apenas processando. Também perderemos todas as mensagens que foram despachadas para esse trabalhador em particular, mas que ainda não foram tratadas.
//Mas não queremos perder nenhuma tarefa. Se um trabalhador morrer, gostaríamos que a tarefa fosse entregue a outro trabalhador.
//Para garantir que uma mensagem nunca seja perdida, o RabbitMQ suporta reconhecimentos de mensagens. Uma confirmação é enviada de volta pelo consumidor para informar ao RabbitMQ que uma determinada mensagem foi recebida, processada e que o RabbitMQ está livre para excluí-la
            consumer: consumer);

        Console.WriteLine(" Press any key to exit.");
        Console.ReadKey();

    }
}

Versão >= 7.x

Propriedades do produtor

using RabbitMQ.Client;
using System.Text;

static async Task Main(string[] args)
{
    var factory = new ConnectionFactory()
    {
        Uri = new Uri("amqp://guest:guest@localhost:5672")
    };

    await using var connection = await factory.CreateConnectionAsync();
    await using var channel = await connection.CreateChannelAsync();

    string message = "This is sample message";
    var body = Encoding.UTF8.GetBytes(message);

    // ✅ Instanciação direta (novo padrão)
    var properties = new BasicProperties
    {
        ContentType = "text/plain",
        DeliveryMode = DeliveryModes.Persistent,
        MessageId = Guid.NewGuid().ToString(),
        Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()),
        Headers = new Dictionary<string, object>
        {
            { "origin", "producer-7x" }
        }
    };

    await channel.BasicPublishAsync(
        exchange: "",
        routingKey: "sample",
        mandatory: false,
        basicProperties: properties,
        body: body
    );

    Console.WriteLine($"[x] Sent: {message}");

}

---

Propriedades do consumidor - Exemplo de um Worker

info

• A API do RabbitMQ.Client mudou no .NET 7+ e a interface IModel foi renomeada para IChannel na versão 7.x do cliente para refletir melhor o modelo de AMQP 0‑9‑1.
• EventingBasicConsumer (sincrono) se transformou em AsyncEventingBasicConsumer (assíncrono)

using RabbitMQ.Client;
using RabbitMQ.Client.Events;

using System.Text;

public sealed class Worker : BackgroundService
{
    private readonly ILogger<Worker> _logger;
    private readonly IConfiguration _configuration;

    private IConnection? _connection;
    private IChannel? _channel;
    private string? _queue;

    public Worker(
        ILogger<Worker> logger,
        IConfiguration configuration)
    {
        _logger = logger;
        _configuration = configuration;
    }

    public override async Task StartAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Worker iniciando (RabbitMQ Consumer)");

        string? connectionString = _configuration.GetConnectionString("rabbitmq");

        if (string.IsNullOrWhiteSpace(connectionString))
            throw new InvalidOperationException("ConnectionStrings:rabbitmq não configurada.");

        ConnectionFactory factory = new ConnectionFactory
        {
            Uri = new Uri(connectionString),
            AutomaticRecoveryEnabled = true,
            NetworkRecoveryInterval = TimeSpan.FromSeconds(10),
        };

        _connection = await factory.CreateConnectionAsync(cancellationToken);
        _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);

        _queue = _configuration["RabbitMq:Queue"];
        if (string.IsNullOrWhiteSpace(_queue))
            throw new InvalidOperationException("RabbitMq:Queue não configurada.");

        _ = await _channel.QueueDeclareAsync(queue: _queue, durable: true, exclusive: false, autoDelete: false, arguments: null, cancellationToken: cancellationToken);

        await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: ushort.Parse(_configuration["RabbitMq:PrefetchCount"] ?? "10"), global: false, cancellationToken: cancellationToken);

        await base.StartAsync(cancellationToken);
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        AsyncEventingBasicConsumer consumer = new AsyncEventingBasicConsumer(_channel!);

        consumer.ReceivedAsync += HandleMessageAsync;

        _ = await _channel!.BasicConsumeAsync(queue: _queue!, autoAck: false, consumer: consumer, cancellationToken: stoppingToken);

        _logger.LogInformation("Consumer conectado à fila {Queue}", _queue);

        while (!stoppingToken.IsCancellationRequested)
        {
            await Task.Delay(1000, stoppingToken); // apenas para manter o serviço ativo
        }
    }

    private async Task HandleMessageAsync(object sender, BasicDeliverEventArgs ea)
    {
        try
        {
            string message = Encoding.UTF8.GetString(ea.Body.ToArray());
            _logger.LogInformation("Mensagem recebida: {Message}", message);

            // 🔹 Simula processamento
            await Task.Delay(500);

            await _channel!.BasicAckAsync(ea.DeliveryTag, multiple: false);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Erro ao processar mensagem");

            await _channel!.BasicNackAsync(
                deliveryTag: ea.DeliveryTag,
                multiple: false,
                requeue: false);
        }
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("Worker parando");

        _ = (_channel?.CloseAsync(cancellationToken));
        _ = (_connection?.CloseAsync(cancellationToken: cancellationToken));

        await base.StopAsync(cancellationToken);
    }

    public override void Dispose()
    {
        _channel?.Dispose();
        _connection?.Dispose();
        base.Dispose();
    }
}