使用 RabbitMQ 傳送訊息

  • 7 分鐘

您可以輕鬆撰寫程式碼，從 RabbitMQ 建立佇列、傳送訊息及接收訊息。 在 .NET Aspire 解決方案中，您也可以協助建立 RabbitMQ 容器，並從微服務連線到該容器。

在您的戶外設備零售商店中，您已決定將 RabbitMQ 實作為集中式訊息代理程式，用於面對客戶的產品目錄網站。 您想要使用 .NET Aspire RabbitMQ 整合來管理此訊息代理程式及其佇列。

在本單元中，您將了解如何建立 RabbitMQ 容器，並用該容器來傳送和接收訊息。

使用 .NET Aspire RabbitMQ 整合

當您從 .NET 使用 RabbitMQ 時，通常必須建立具有連接字串的 ConnectionFactory 物件，然後使用該物件來建立服務的連線。 在 .NET Aspire 專案中，RabbitMQ 連線較容易管理，因為：

  • 您會在 AppHost 專案中註冊連線和連接字串。
  • 當您將服務的參考傳遞至取用專案時，他們可以使用相依性插入來取得 RabbitMQ 的連線。 他們不需要建立及設定自己的連線。

在主控處理程序中設定 RabbitMQ

在 .NET Aspire 中，您必須在主控處理程序中安裝 Rabbit MQ 主控整合：

dotnet add package Aspire.Hosting.RabbitMQ

現在，您可以註冊 RabbitMQ 服務，並將其傳遞給使用該服務的專案：

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

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

AppHost 會管理解決方案中所有專案的連線。

設定 Rab

接下來，將 .NET Aspire RabbitMQ 整合新增至使用該整合的每個專案：

dotnet add package Aspire.RabbitMQ.Client

若要取得 RabbitMQ 訊息代理程式的參考，請呼叫 AddRabbitMQClient() 方法：

builder.AddRabbitMQClient("messaging");

現在，您可以使用相依性插入來取得 RabbitMQ 的連線：

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

有了連線後，下一個步驟是建立傳訊通道，如下所示：

var channel = connection.CreateModel();

傳送訊息

擁有傳訊通道之後，您就可以使用該通道來設定傳訊拓撲的佇列、交換和其他整合。 例如，若要建立佇列，請使用下列程式碼：

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

您可以使用 BasicPublish 方法將訊息傳送至此佇列，但訊息會預期本文是位元組陣列：

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

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

接收訊息

在接收整合中，您會以與傳送者相同的方式建立傳訊通道和佇列。 請確定佇列名稱符合您在傳送整合中建立的佇列名稱。 否則，您會建立兩個不同的佇列，而訊息不會到達正確的目的地。

您必須建立新的 EventingBasicConsumer() 方法並註冊方法來處理 Received 事件：

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

訊息處理常式會使用 BasicDeliverEventArgs 物件來取得訊息的屬性，包括訊息本文。 您必須記得將訊息本文還原序列化：

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

最後，若要檢查佇列是否有新訊息，請呼叫 BasicConsume() 方法;

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

深入了解