在ASP.NET Core 中使用 .NET Aspire 消息传递组件

前言

云原生应用程序通常需要可扩展的消息传递解决方案，以提供消息队列、主题和订阅等功能。.NET Aspire 组件简化了连接到各种消息传递提供程序（例如 Azure 服务总线）的过程。在本教程中，小编将为大家介绍如何创建一个 ASP.NET Core 应用并将提交的消息将发送到服务总线主题以供订阅者使用。

环境准备

要使用 .NET Aspire，需要在本地安装以下软件：

.NET 8.0
.NET Aspire 工作负载：
使用 Visual Studio 安装程序
使用dotnet workload install aspire命令
Docker 桌面
集成开发环境 (IDE) 或代码编辑器，例如：
Visual Studio 2022 预览版 17.9 或更高版本（可选）
Visual Studio 代码（可选）

设置 Azure 服务总线账户

az group create -n <your-resource-group-name> -location eastus
az servicebus namespace create -g <your-resource-group-name> --name <your-namespace-name> --location eastus
az servicebus topic create --g <your-resource-group-name> --namespace-name <your-namespace-name> --name notifications
az servicebus topic subscription create --g <your-resource-group-name> --namespace-name <your-namespace-name> --topic-name notifications --name mobile

备注：your-resource-group-name和your-namespace-name替换为自己值即可。

Azure 身份验证

可以使用无密码身份验证或连接字符串来完成此快速入门。无密码连接使用 Azure Active Directory 和基于角色的访问控制 (RBAC) 连接到服务总线命名空间。无需担心代码、配置文件或安全存储（例如 Azure Key Vault）中存在硬编码连接字符串。

除此之外，还可以使用连接字符串连接到服务总线命名空间，但建议在实际应用程序和生产环境中使用无密码方法。有关更多信息，请阅读身份验证和授权或访问无密码概述页面。

创建项目

在 Visual Studio 顶部，导航到“文件” “新建” “项目”。
在对话框窗口中，搜索ASP.NET Core并选择ASP.NET Core Web API。选择下一步。
在“配置新项目”屏幕上：

输入项目名称AspireMessaging。
将其余值保留为默认值，然后选择“下一步”。

添加 Worker Service

接下来，将工作线程服务项目添加到解决方案，以检索和处理发往 Azure 服务总线的消息。

在解决方案资源管理器中，右键单击顶级AspireMessaging解决方案节点，然后选择“添加” “新项目”。
搜索并选择Worker Service模板，然后选择Next。
对于项目名称，输入AspireMessaging.Worker并选择下一步。
在附加信息屏幕上：

确保选择.NET 8.0 。
确保选中Enlist in .NET Aspire Orchestration并选择Create。

Visual Studio 将项目添加到您的解决方案中，并使用新的代码行更新项目的Program.cs文件：AspireMessaging.AppHost

builder.AddProject<Projects.AspireMessaging_WorkerService>("aspiremessaging.workerservice");

完整的文件结构：

将 .NET Aspire 组件添加到 API

将.NET Aspire Azure 服务总线组件添加到您的AspireMessaging应用程序：

dotnet add package Aspire.Azure.Messaging.ServiceBus --prerelease

在Razor Pages 项目的Program.csAspireMessaging文件中，添加对扩展方法的调用AddAzureServiceBus：

builder.AddAzureServiceBus("serviceBusConnection");

在项目的_appsettings.json文件中AspireMessaging，添加对应的连接信息：

{"ConnectionStrings": {"serviceBusConnection": "Endpoint=sb://{your_namespace}.servicebus.windows.net/;SharedAccessKeyName=accesskeyname;SharedAccessKey=accesskey"}
}

备注：将{your_namespace}替换为自己的服务总线空间的名称

创建 API 端点

提供一个端点来接收数据并将其发布到服务总线主题并向订阅者广播。将以下端点添加到AspireMessaging项目中以向主题发送消息：

app.MapPost("/notify", static async (ServiceBusClient client, string message) =>
{var sender = client.CreateSender("notifications");// Create a batchusing ServiceBusMessageBatch messageBatch =await sender.CreateMessageBatchAsync();if (messageBatch.TryAddMessage(new ServiceBusMessage($"Message {message}")) is false){// If it's too large for the batch.throw new Exception($"The message {message} is too large to fit in the batch.");}// Use the producer client to send the batch of // messages to the Service Bus topic.await sender.SendMessagesAsync(messageBatch);Console.WriteLine($"A message has been published to the topic.");
})

将 .NET Aspire 组件添加到 Worker Service

将.NET Aspire Azure 服务总线组件添加到AspireMessaging.Worker应用程序：

dotnet add package Aspire.Azure.Messaging.ServiceBus --prerelease

在Razor Pages 项目的Program.csAspireMessaging.Worker文件中，添加对扩展方法的调用AddAzureServiceBus：

builder.AddAzureServiceBus("serviceBusConnection");

在项目的_appsettings.json文件中AspireMessaging.Worker，添加对应的连接信息：

{"ConnectionStrings": {"serviceBusConnection": "Endpoint=sb://{your_namespace}.servicebus.windows.net/;SharedAccessKeyName=accesskeyname;SharedAccessKey=accesskey"}
}

备注：将{your_namespace}替换为自己的服务总线空间的名称

处理来自订阅者的消息

当新消息放入队列时messages，工作服务应检索、处理和删除该消息。更新Worker.cs类以匹配以下代码：

public class Worker(ILogger<Worker> logger,ServiceBusClient client) : BackgroundService
{protected override async Task ExecuteAsync(CancellationToken stoppingToken){while (!stoppingToken.IsCancellationRequested){var processor = client.CreateProcessor("notifications","mobile",new ServiceBusProcessorOptions());// add handler to process messagesprocessor.ProcessMessageAsync += MessageHandler;// add handler to process any errorsprocessor.ProcessErrorAsync += ErrorHandler;// start processing await processor.StartProcessingAsync();logger.LogInformation("Wait for a minute and then press any key to end the processing");Console.ReadKey();// stop processinglogger.LogInformation("\nStopping the receiver...");await processor.StopProcessingAsync();logger.LogInformation("Stopped receiving messages");}}async Task MessageHandler(ProcessMessageEventArgs args){string body = args.Message.Body.ToString();logger.LogInformation("Received: {Body} from subscription.", body);// complete the message. messages is deleted from the subscription.await args.CompleteMessageAsync(args.Message);}// handle any errors when receiving messagesTask ErrorHandler(ProcessErrorEventArgs args){logger.LogError(args.Exception, args.Exception.Message);return Task.CompletedTask;}
}

最后：在本地运行并测试应用程序