前言

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

环境准备

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

  • .NET Aspire 工作负载：
  • 集成开发环境 (IDE) 或代码编辑器，例如：
  • 版 17.9 或更高版本（可选）
  • （可选）

设置 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）中存在硬编码连接字符串。

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

创建项目

  1. 在 Visual Studio 顶部，导航到“文件” “新建” “项目”。
  2. 在对话框窗口中，搜索ASP.NET Core并选择ASP.NET Core Web API。选择下一步。
  3. 在“配置新项目”屏幕上：
  • 输入项目名称AspireMessaging。
  • 将其余值保留为默认值，然后选择“下一步”。

添加 Worker Service

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

  1. 在解决方案资源管理器中，右键单击顶级AspireMessaging解决方案节点，然后选择“添加” “新项目”。
  2. 搜索并选择Worker Service模板，然后选择Next。
  3. 对于项目名称，输入AspireMessaging.Worker并选择下一步。
  4. 在附加信息屏幕上：
  • 确保选择.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 batch
    using 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 messages
            processor.ProcessMessageAsync += MessageHandler;

            // add handler to process any errors
            processor.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 processing
            logger.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 messages
    Task ErrorHandler(ProcessErrorEventArgs args)
    {
        logger.LogError(args.Exception, args.Exception.Message);
        return Task.CompletedTask;
    }
}

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

  1. 按 Visual Studio 顶部的运行按钮启动 Aspire 应用程序。.NET Aspire 仪表板应用程序应在浏览器中打开。
  2. 在项目页面的aspireweb行中，单击Endpoints列中的链接以打开 API 的 Swagger UI 页面。
  3. 在 .NET Aspire 仪表板上，导航到AspireWorkerService项目的日志。
  4. 返回 Swagger UI 页面，展开/notify端点并选择Try it out。
  5. 在消息输入框中输入测试消息。
  6. 选择执行以发送测试请求。
  7. 切换回AspireWorkerService日志。看到输出日志中打印的测试消息。

参考链接：

.net8系列的另外两篇文章：