消息与事件集成

更新: 2026-10-16 15:20:09   字数: 0 字   时长: 0 分钟

事件驱动架构里，消息中间件通常是「最难在本地跑起来」的一环。Aspire 把它变成一行代码。

支持的中间件

中间件	托管集成包	添加方法	本地实现
RabbitMQ	Aspire.Hosting.RabbitMQ	AddRabbitMQ(name)	容器
Kafka	Aspire.Hosting.Kafka	AddKafka(name)	容器（KRaft）
NATS	Aspire.Hosting.NATS	AddNats(name)	容器
Azure Service Bus	Aspire.Hosting.Azure.ServiceBus	AddAzureServiceBus(name)	本地模拟器或真实资源

RabbitMQ

var builder = DistributedApplication.CreateBuilder(args);

var rabbit = builder.AddRabbitMQ("rabbitmq")
    .WithManagementPlugin()       // 启用管理界面
    .WithDataVolume();

var api = builder.AddProject<Projects.Api>("api")
    .WithReference(rabbit)
    .WaitFor(rabbit);

builder.Build().Run();

服务端消费：

builder.AddServiceDefaults();
builder.AddRabbitMQClient("rabbitmq");

RabbitMQ 的密码与数据卷

与数据库一样，RabbitMQ 使用数据卷时必须固定密码，否则每次运行生成的新密码会让旧数据无法访问：

aspire secret set Parameters:rabbitmq-password "your-password"

Kafka

var kafka = builder.AddKafka("kafka")
    .WithDataVolume()
    .WithKafkaUI();               // 附带 Kafka UI

var worker = builder.AddProject<Projects.Worker>("worker")
    .WithReference(kafka)
    .WaitFor(kafka);

服务端消费：

builder.AddServiceDefaults();
builder.AddKafkaProducer<string, string>("kafka");
builder.AddKafkaConsumer<string, string>("kafka", settings =>
{
    settings.Config.GroupId = "orders";
    settings.Config.AutoOffsetReset = AutoOffsetReset.Earliest;
});

NATS

var nats = builder.AddNats("nats");

builder.AddProject<Projects.Notifier>("notifier")
    .WithReference(nats);

Azure Service Bus

var serviceBus = builder.AddAzureServiceBus("messaging")
    .RunAsEmulator();             // 本地运行时使用模拟器

var queue = serviceBus.AddServiceBusQueue("orders");
var topic = serviceBus.AddServiceBusTopic("events")
    .AddServiceBusSubscription("billing");

builder.AddProject<Projects.Api>("api")
    .WithReference(queue)
    .WithReference(topic);

RunAsEmulator 的价值

RunAsEmulator() 让「本地开发」与「云端部署」使用同一套资源定义：本地跑模拟器，发布时自动切换为真实的 Azure 资源。这是 Aspire「同一模型、两种环境」理念最典型的体现。

在非 .NET 服务中使用

消息中间件同样通过环境变量注入连接信息：

// Node.js（amqplib）
const url = process.env.ConnectionStrings__rabbitmq

# Python（pika）
import os
url = os.getenv("ConnectionStrings__rabbitmq")

完整的变量命名规则见环境变量与配置注入。

设计建议

• 用 WaitFor 而不是重试：服务启动时中间件往往还没就绪，WaitFor 能把这类竞态消除在编排层。
• 把队列/主题当作资源声明：AddServiceBusQueue、AddServiceBusTopic 会让它们在部署时被一并创建，避免「上线忘建队列」。
• 本地不要图省事直连生产中间件：容器化的本地实例是 Aspire 最推荐的形态。

下一步

• 缓存集成
• 云服务集成
• 健康检查