Azure Event Grid .NET
Azure.Messaging.EventGrid (.NET)
用于向 Azure Event Grid 主题 (topics)、域 (domains) 和命名空间 (namespaces) 发布事件的客户端库。
安装
# 适用于主题和域(推送交付)
dotnet add package Azure.Messaging.EventGrid
适用于命名空间(拉取交付)
dotnet add package Azure.Messaging.EventGrid.Namespaces
适用于 CloudNative CloudEvents 互操作
dotnet add package Microsoft.Azure.Messaging.EventGrid.CloudNativeCloudEvents当前版本: 4.28.0 (stable)
环境变量
# 主题/域端点
EVENT_GRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events
EVENT_GRID_TOPIC_KEY=<access-key>
命名空间端点(用于拉取交付)
EVENT_GRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net
EVENT_GRID_TOPIC_NAME=<topic-name>
EVENT_GRID_SUBSCRIPTION_NAME=<subscription-name>客户端层级
推送交付 (主题/域)
└── EventGridPublisherClient
├── SendEventAsync(EventGridEvent)
├── SendEventsAsync(IEnumerable<EventGridEvent>)
├── SendEventAsync(CloudEvent)
└── SendEventsAsync(IEnumerable<CloudEvent>)
拉取交付 (命名空间)
├── EventGridSenderClient
│ └── SendAsync(CloudEvent)
└── EventGridReceiverClient
├── ReceiveAsync()
├── AcknowledgeAsync()
├── ReleaseAsync()
└── RejectAsync()
身份验证
API 密钥验证
using Azure;
using Azure.Messaging.EventGrid;
EventGridPublisherClient client = new(
new Uri("https://mytopic.eastus-1.eventgrid.azure.net/api/events"),
new AzureKeyCredential("<access-key>"));
Microsoft Entra ID (推荐)
using Azure.Identity;
using Azure.Messaging.EventGrid;
EventGridPublisherClient client = new(
new Uri("https://mytopic.eastus-1.eventgrid.azure.net/api/events"),
new DefaultAzureCredential());
SAS 令牌验证
string sasToken = EventGridPublisherClient.BuildSharedAccessSignature(
new Uri(topicEndpoint),
DateTimeOffset.UtcNow.AddHours(1),
new AzureKeyCredential(topicKey));
var sasCredential = new AzureSasCredential(sasToken);
EventGridPublisherClient client = new(
new Uri(topicEndpoint),
sasCredential);
发布事件
EventGridEvent 架构
EventGridPublisherClient client = new(
new Uri(topicEndpoint),
new AzureKeyCredential(topicKey));
// 单个事件
EventGridEvent egEvent = new(
subject: "orders/12345",
eventType: "Order.Created",
dataVersion: "1.0",
data: new { OrderId = "12345", Amount = 99.99 });
await client.SendEventAsync(egEvent);
// 批量事件
List<EventGridEvent> events = new()
{
new EventGridEvent(
subject: "orders/12345",
eventType: "Order.Created",
dataVersion: "1.0",
data: new OrderData { OrderId = "12345", Amount = 99.99 }),
new EventGridEvent(
subject: "orders/12346",
eventType: "Order.Created",
dataVersion: "1.0",
data: new OrderData { OrderId = "12346", Amount = 149.99 })
};
await client.SendEventsAsync(events);
CloudEvent 架构
CloudEvent cloudEvent = new(
source: "/orders/system"cloudEvent.Subject = "orders/12345";
cloudEvent.Id = Guid.NewGuid().ToString();
cloudEvent.Time = DateTimeOffset.UtcNow;
await client.SendEventAsync(cloudEvent);
// CloudEvents 批处理
List<CloudEvent> cloudEvents = new()
{
new CloudEvent("/orders", "Order.Created", new { OrderId = "1" }),
new CloudEvent("/orders", "Order.Updated", new { OrderId = "2" })
};
await client.SendEventsAsync(cloudEvents);
### 发布到 Event Grid 域 (Domain)// 为了进行域路由,事件必须指定 Topic 属性
List<EventGridEvent> events = new()
{
new EventGridEvent(
subject: "orders/12345",
eventType: "Order.Created",
dataVersion: "1.0",
data: new { OrderId = "12345" })
{
Topic = "orders-topic" // 域主题名称
},
new EventGridEvent(
subject: "inventory/item-1",
eventType: "Inventory.Updated",
dataVersion: "1.0",
data: new { ItemId = "item-1" })
{
Topic = "inventory-topic"
}
};
await client.SendEventsAsync(events);
### 自定义序列化using System.Text.Json;
var serializerOptions = new JsonSerializerOptions
{
PropertyNamingPolicy = JsonNamingPolicy.CamelCase
};
var customSerializer = new JsonObjectSerializer(serializerOptions);
EventGridEvent egEvent = new(
subject: "orders/12345",
eventType: "Order.Created",
dataVersion: "1.0",
data: customSerializer.Serialize(new OrderData { OrderId = "12345" }));
await client.SendEventAsync(egEvent);
## 拉取交付 (命名空间)
向命名空间主题发送事件
var senderClient = new EventGridSenderClient(
new Uri(namespaceEndpoint),
topicName,
new AzureKeyCredential(topicKey));
// 发送单个事件
CloudEvent cloudEvent = new("employee_source", "Employee.Created",
new { Name = "John", Age = 30 });
await senderClient.SendAsync(cloudEvent);
// 批量发送
await senderClient.SendAsync(new[]
{
new CloudEvent("source", "type", new { Name = "Alice" }),
new CloudEvent("source", "type", new { Name = "Bob" })
});
### 接收并处理事件var receiverClient = new EventGridReceiverClient(
new Uri(namespaceEndpoint),
topicName,
subscriptionName,
new AzureKeyCredential(topicKey));
// 接收事件
ReceiveResult result = await receiverClient.ReceiveAsync(maxEvents: 10);
List<string> lockTokensToAck = new();
List<string> lockTokensToRelease = new();
foreach (ReceiveDetails detail in result.Details)
{
CloudEvent cloudEvent = detail.Event;
string lockToken = detail.BrokerProperties.LockToken;
try
{
// 处理事件
Console.WriteLine($"Event: {cloudEvent.Type}, Data: {cloudEvent.Data}");
lockTokensToAck.Add(lockToken);
}
catch (Exception)
{
// 释放以进行重试
lockTokensToRelease.Add(lockToken);
}
}
// 确认已成功处理的事件
if (lockTokensToAck.Any())
{
await receiverClient.AcknowledgeAsync(lockTokensToAck);
}
// 释放事件以进行重试
if (lockTokensToRelease.Any())
{
await receiverClient.ReleaseAsync(lockTokensToRelease);
}
### 拒绝事件 (死信)// 拒绝无法处理的事件
await receiverClient.RejectAsync(new[] { lockToken });
## 消费事件 (
Azure Functions)
EventGridEvent 触发器
public static class EventGridFunction
{
[FunctionName("ProcessEventGridEvent")]
public static void Run(
[EventGridTrigger] EventGridEvent eventGridEvent,
ILogger log)
{
log.LogInformation($"Event Type: {eventGridEvent.EventType}");
log.LogInformation($"Subject: {eventGridEvent.Subject}");
log.LogInformation($"Data: {eventGridEvent.Data}");
}
}
### CloudEvent 触发器using Azure.Messaging;
using Microsoft.Azure.Functions.Worker;
public class CloudEventFunction
{
[Function("ProcessCloudEvent")]
public void Run(
[EventGridTrigger] CloudEvent cloudEvent,
FunctionContext context)
{
var logger = context.GetLogger("ProcessCloudEvent");
logger.LogInformation($"Event Type: {cloudEvent.Type}");
logger.LogInformation($"Source: {cloudEvent.Source}");
logger.LogInformation($"Data: {cloudEvent.Data}");
}
}
## 解析事件
解析 EventGridEvent
foreach (EventGridEvent egEvent in events)
{
if (egEvent.TryGetSystemEventData(out object systemEvent))
{
// 处理系统事件
switch (systemEvent)
{
case StorageBlobCreatedEventData blobCreated:
Console.WriteLine($"Blob created: {blobCreated.Url}");
break;
}
}
else
{
// 处理自定义事件
var customData = egEvent.Data.ToObjectFromJson<MyCustomData>();
}
}
### 解析 CloudEventCloudEvent[] cloudEvents = CloudEvent.ParseMany(BinaryData.FromString(json));
foreach (CloudEvent cloudEvent in cloudEvents)
{
var data = cloudEvent.Data.ToObjectFromJson<MyEventData>();
Console.WriteLine($"Type: {cloudEvent.Type}, Data: {data}");
}
## 系统事件// 常见的系统事件类型
using Azure.Messaging.EventGrid.SystemEvents;
// 存储事件
StorageBlobCreatedEventData blobCreated;
StorageBlobDeletedEventData blobDeleted;
// 资源事件
ResourceWriteSuccessEventData resourceCreated;
ResourceDeleteSuccessEventData resourceDeleted;
// App Service 事件
WebAppUpdatedEventData webAppUpdated;
// 容器注册表事件
ContainerRegistryImagePushedEventData imagePushed;
// IoT Hub 事件
IotHubDeviceCreatedEventData deviceCreated;
## 关键类型参考
| 类型 | 用途 |
|------|---------|
| EventGridPublisherClient | 发布到主题/域 |
| EventGridSenderClient | 发送到命名空间主题 |
| EventGridReceiverClient | 从命名空间订阅中接收 |
| EventGridEvent | Event Grid 原生架构 |
| CloudEvent | CloudEvents 1.0 架构 |
| ReceiveResult | 拉取传递响应 |
| ReceiveDetails | 包含代理属性的事件 |
| BrokerProperties | 锁定令牌、传递次数 |
事件架构对比
| 特性 | EventGridEvent | CloudEvent |
|---------|----------------|------------|
| 标准 | Azure 专用 | CNCF 标准 |
| 必填字段 | subject, eventType, dataVersion, data | source, type |
| 扩展性 | 有限 | 扩展属性 |
| 互操作性 | 仅限 Azure | 跨平台 |
最佳实践
1. 使用 CloudEvents —
1. 优先使用 CloudEvents 进行新实现(行业标准)
2. 批量事件 — 单次调用发送多个事件以提高效率
3. 使用 Entra ID — 优先使用托管标识而非访问密钥
4. 幂等处理程序 — 事件可能会被多次交付
5. 设置事件 TTL — 为命名空间事件配置生存时间 (Time-to-Live)
6. 处理部分失败 — 单独确认/释放事件
7. 使用死信队列 — 为失败事件配置死信队列
8. 验证 Schema — 在处理前验证事件数据
错误处理
try
{
await client.SendEventAsync(cloudEvent);
}
catch (RequestFailedException ex) when (ex.Status == 401)
{
Console.WriteLine("身份验证失败 - 请检查凭据");
}
catch (RequestFailedException ex) when (ex.Status == 403)
{
Console.WriteLine("授权失败 - 请检查 RBAC 权限");
}
catch (RequestFailedException ex) when (ex.Status == 413)
{
Console.WriteLine("有效负载过大 - 每个事件最大 1MB,每批次总计最大 1MB");
}
catch (RequestFailedException ex)
{
Console.WriteLine($"Event Grid 错误: {ex.Status} - {ex.Message}");
}
## 故障转移模式try
{
var primaryClient = new EventGridPublisherClient(primaryUri, primaryKey);
await primaryClient.SendEventsAsync(events);
}
catch (RequestFailedException)
{
// 故障转移至次要区域
var secondaryClient = new EventGridPublisherClient(secondaryUri, secondaryKey);
await secondaryClient.SendEventsAsync(events);
}
``
相关 SDK
| SDK | 用途 | 安装 |
|-----|---------|---------|
|
Azure.Messaging.EventGrid | 主题/域 (本 SDK) | dotnet add package Azure.Messaging.EventGrid |
| Azure.Messaging.EventGrid.Namespaces | 拉取交付 (Pull delivery) | dotnet add package Azure.Messaging.EventGrid.Namespaces |
| Azure.Identity | 身份验证 | dotnet add package Azure.Identity |
| Microsoft.Azure.WebJobs.Extensions.EventGrid | Azure Functions 触发器 | dotnet add package Microsoft.Azure.WebJobs.Extensions.EventGrid` |
参考链接
| 资源 | URL |
|----------|-----|
| NuGet 包 | https://www.nuget.org/packages/Azure.Messaging.EventGrid |
| API 参考 | https://learn.microsoft.com/dotnet/api/azure.messaging.eventgrid |
| 快速入门 | https://learn.microsoft.com/azure/event-grid/custom-event-quickstart |
| 拉取交付 | https://learn.microsoft.com/azure/event-grid/pull-delivery-overview |
| GitHub 源码 | https://github.com/Azure/azure-sdk-for-net/tree/main/sdk/eventgrid/Azure.Messaging.EventGrid |
使用场景
本技能适用于执行概览中所描述的工作流或操作。局限性
- 仅在任务与上述范围明确匹配时使用此技能。
- 不要将输出视为针对特定环境的验证、测试或专家评审的替代方案。
- 如果缺少必要的输入、权限、安全边界或成功标准,请停止并请求澄清。