microsoft/azure-eventgrid-dotnet

Azure.Messaging.EventGrid (.NET)

用于向 Azure Event Grid 主题、域和命名空间发布事件的客户端库。

安装

# 主题和域（推送交付）
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（稳定版）

环境变量

EVENT_GRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events  # 必需：Event Grid 主题或域终结点
EVENT_GRID_TOPIC_KEY=<access-key>  # 仅在 AzureKeyCredential 身份验证时需要
EVENT_GRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net  # 可选：Event Grid 命名空间终结点
EVENT_GRID_TOPIC_NAME=<topic-name>  # 必需：Event Grid 主题名称
EVENT_GRID_SUBSCRIPTION_NAME=<subscription-name>  # 可选：Event Grid 订阅名称
AZURE_TOKEN_CREDENTIALS=prod  # 仅在生产环境中使用 DefaultAzureCredential 时需要

客户端层次结构

推送交付（主题/域）
└── 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 令牌凭据

using Azure.Identity;
using Azure.Messaging.EventGrid;

// 本地开发：DefaultAzureCredential。生产环境：设置 AZURE_TOKEN_CREDENTIALS=prod 或 AZURE_TOKEN_CREDENTIALS=<specific_credential>
var credential = new DefaultAzureCredential(
    DefaultAzureCredential.DefaultEnvironmentVariableName
);
// 或者在生产环境中直接使用特定凭据：
// 请参阅 https://learn.microsoft.com/dotnet/api/overview/azure/identity-readme?view=azure-dotnet#credential-classes
// var credential = new ManagedIdentityCredential();
EventGridPublisherClient client = new(
    new Uri("https://mytopic.eastus-1.eventgrid.azure.net/api/events"),
    credential);

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",
    type: "Order.Created",
    data: new { OrderId = "12345", Amount = 99.99 });

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 域

// 事件必须指定 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);

拉取交付（命名空间）

向命名空间主题发送事件

using Azure;
using Azure.Messaging;
using Azure.Messaging.EventGrid.Namespaces;

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($"事件: {cloudEvent.Type}, 数据: {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 触发器

using Azure.Messaging.EventGrid;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.EventGrid;

public static class EventGridFunction
{
    [FunctionName("ProcessEventGridEvent")]
    public static void Run(
        [EventGridTrigger] EventGridEvent eventGridEvent,
        ILogger log)
    {
        log.LogInformation($"事件类型: {eventGridEvent.EventType}");
        log.LogInformation($"主题: {eventGridEvent.Subject}");
        log.LogInformation($"数据: {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($"事件类型: {cloudEvent.Type}");
        logger.LogInformation($"源: {cloudEvent.Source}");
        logger.LogInformation($"数据: {cloudEvent.Data}");
    }
}

解析事件

解析 EventGridEvent

// 从 JSON 字符串
string json = "..."; // Event Grid webhook 负载
EventGridEvent[] events = EventGridEvent.ParseMany(BinaryData.FromString(json));

foreach (EventGridEvent egEvent in events)
{
    if (egEvent.TryGetSystemEventData(out object systemEvent))
    {
        // 处理系统事件
        switch (systemEvent)
        {
            case StorageBlobCreatedEventData blobCreated:
                Console.WriteLine($"Blob 已创建: {blobCreated.Url}");
                break;
        }
    }
    else
    {
        // 处理自定义事件
        var customData = egEvent.Data.ToObjectFromJson<MyCustomData>();
    }
}

解析 CloudEvent

CloudEvent[] cloudEvents = CloudEvent.ParseMany(BinaryData.FromString(json));

foreach (CloudEvent cloudEvent in cloudEvents)
{
    var data = cloudEvent.Data.ToObjectFromJson<MyEventData>();
    Console.WriteLine($"类型: {cloudEvent.Type}, 数据: {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 — 新实现优先使用 CloudEvents（行业标准）
2. 批处理事件 — 单个调用发送多个事件以提高效率
3. 使用 Entra ID — 优先使用托管标识而非访问密钥
4. 幂等处理 — 事件可能被多次投递
5. 设置事件 TTL — 为命名空间事件配置生存时间
6. 处理部分失败 — 单独确认/释放事件
7. 使用死信 — 为失败事件配置死信
8. 验证架构 — 在处理前验证事件数据

错误处理

using Azure;

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	拉取交付	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