Invio di messaggi con RabbitMQ

7 minuti

È semplice scrivere codice che crea code, invia messaggi e riceve messaggi da RabbitMQ. In una soluzione .NET Aspire è possibile anche creare il contenitore RabbitMQ e stabilirne le connessioni da microservizi.

Nel proprio negozio di attrezzature outdoor, si è deciso di implementare RabbitMQ come broker di messaggi centralizzato per il sito Web del catalogo prodotti rivolto ai clienti. Si vuole usare l’integrazione RabbitMQ di .NET Aspire per gestire questo broker e le relative code.

In questa unità si apprenderà come creare un contenitore RabbitMQ e usarlo per inviare e ricevere messaggi.

Usare l’integrazione RabbitMQ di .NET Aspire

Quando si usa RabbitMQ da .NET, in genere è necessario creare un oggetto ConnectionFactory con una stringa di connessione e quindi usarlo per stabilire connessioni al servizio. In un progetto .NET Aspire è più semplice gestire la connessione RabbitMQ perché:

Si registrano una connessione e una stringa di connessione nel progetto AppHost.
Quando si passa un riferimento al servizio per utilizzare i progetti, questi possono sfruttare l'inserimento delle dipendenze per ottenere una connessione a RabbitMQ. Non devono creare e configurare le proprie connessioni.

Configurazione di RabbitMQ nell'host dell'app

In .NET Aspire è necessario installare l’integrazione di hosting Rabbit MQ nell'host delle app:

dotnet add package Aspire.Hosting.RabbitMQ

È ora possibile registrare il servizio RabbitMQ e passarlo ai progetti che lo usano:

// Service registration
var rabbit = builder.AddRabbitMQ("messaging");

// Service consumption
builder.AddProject<Projects.CatalogAPI>()
       .WithReference(rabbit);

AppHost gestisce la connessione per tutti i progetti della soluzione.

Configurazione di Rab

Aggiungere quindi l’integrazione RabbitMQ di .NET Aspire a ogni progetto che lo usa:

dotnet add package Aspire.RabbitMQ.Client

Per ottenere un riferimento al broker di messaggi RabbitMQ, chiamare il metodo AddRabbitMQClient():

builder.AddRabbitMQClient("messaging");

È ora possibile usare l'inserimento delle dipendenze per ottenere la connessione a RabbitMQ:

public class CatalogAPI(IConnection rabbitConnection)
{
    // Send and receive messages here
}

Con la connessione, il passaggio successivo consiste nel creare un canale di messaggistica, come illustrato di seguito:

var channel = connection.CreateModel();

Invio di messaggi

Dopo aver creato il canale di messaggistica, è possibile usarlo per configurare code, scambi e altre integrazioni della propria topologia di messaggistica. Ad esempio, per creare una coda, usare questo codice:

channel.QueueDeclare(queue: "catalogEvents",
    durable: false,
    exclusive: false,
    autoDelete: false,
    arguments: null);

Si usa il metodo BasicPublish per inviare un messaggio a questa coda, ma il messaggio prevede che il corpo sia una matrice di byte:

var body = Encoding.UTF8.GetBytes("Getting all items in the catalog.");

channel.BasicPublish(exchange: string.Empty,
    routingKey: "catalogEvents",
    basicProperties: null,
    body: body);

Ricezione di messaggi

Nell’integrazione ricevente creare il canale di messaggistica e la coda, nello stesso modo usato per il mittente. Assicurarsi che il nome della coda corrisponda a quello creato nell’integrazione di invio. In caso contrario, si creeranno due code separate e i messaggi non arriveranno alla destinazione corretta.

È necessario creare un nuovo metodo EventingBasicConsumer() e registrare un metodo per gestire l'evento Received:

var consumer = new EventingBasicConsumer(channel);
consumer.Received += ProcessMessageAsync;

Il gestore di messaggi utilizza un oggetto BasicDeliverEventArgs per ottenere le proprietà del messaggio, incluso il corpo del messaggio. È necessario ricordare di deserializzare il corpo del messaggio:

private void ProcessMessageAsync(object? sender, BasicDeliverEventArgs args)
{
    string messagetext = Encoding.UTF8.GetString(args.Body.ToArray());
    logger.LogInformation("The message is: {text}", messagetext);
}

Infine, per controllare la presenza di nuovi messaggi nella coda, chiamare il metodo BasicConsume();

channel.BasicConsume(queue:  queueName,
                    autoAck: true, 
                    consumer: consumer);