如何用 Azure Event Hubs 构建 .NET 事件溯源？-常见问题-PHP中文网

如何用 Azure Event Hubs 构建 .NET 事件溯源？

小老鼠

发布： 2025-09-26 15:50:02

原创

628人浏览过

使用 Azure Event Hubs 构建 .NET 事件溯源系统，需将领域事件序列化并写入高吞吐消息管道；2. 定义不可变事件如 OrderCreated、OrderShipped，用 System.Text.Json 序列化为字节数组，并通过 ContentType 和 EventType 属性标识类型；3. 使用 EventHubProducerClient 批量发送事件，设置 PartitionKey 确保同一聚合根事件有序；4. 消费端通过 EventProcessorClient 结合 BlobCheckpointStore 实现至少一次处理，业务逻辑需幂等；5. 可选重建聚合时从指定分区读取事件流，高频场景引入快照减少回放开销；6. Event Hubs 适合作为 CQRS 架构中的事件分发通道，虽无版本管理与复杂查询，但保障了事件的有序性、不可变性与可重播性。

如何用 azure event hubs 构建 .net 事件溯源？

用 Azure Event Hubs 构建 .NET 事件溯源系统，核心是将领域事件持久化到高吞吐的消息管道中，并由消费者按顺序处理。Event Hubs 虽非传统事件存储（如 EventStoreDB），但结合正确设计，可支撑轻量级事件溯源架构。

定义领域事件与序列化

事件溯源的基础是不可变的事件流。在 .NET 中，先定义清晰的事件类型：

public record OrderCreated(
    Guid OrderId,
    string CustomerName,
    DateTime CreatedAt
);

public record OrderShipped(
    Guid OrderId,
    DateTime ShippedAt
);

登录后复制

使用 System.Text.Json 或 MessagePack 序列化事件为字节数组。建议附加事件类型名到消息属性，便于反序列化：

将事件写入 EventData 时，设置 ContentType 或 EventType 属性
统一采用 UTF-8 编码避免乱码问题

发布事件到 Event Hubs

使用 Azure.Messaging.EventHubs.Producer.EventHubProducerClient 发布事件：

即构数智人

即构数智人是由即构科技推出的AI虚拟数字人视频创作平台，支持数字人形象定制、短视频创作、数字人直播等。

查看详情

var producer = new EventHubProducerClient(connectionString, eventHubName);

using var eventBatch = await producer.CreateBatchAsync();
var eventData = new EventData(Encoding.UTF8.GetBytes(json));
eventData.ContentType = "application/json";
eventData.ApplicationProperties["EventType"] = "OrderCreated";

if (!eventBatch.TryAdd(eventData))
    throw new Exception("Event too large");

await producer.SendAsync(eventBatch);

登录后复制

启用批次发送提升吞吐量
通过 PartitionKey 确保同一聚合根的事件顺序一致（如使用 OrderId 做哈希）

消费事件并更新读模型

使用 EventProcessorClient 持久化处理检查点，确保至少一次语义：

var processor = new EventProcessorClient(
    checkpointStore,     // 如 BlobCheckpointStore
    consumerGroup,
    connectionString,
    eventHubName);

processor.ProcessEventAsync += async args =>
{
    var json = Encoding.UTF8.GetString(args.Data.Body);
    var eventType = args.Data.ApplicationProperties["EventType"].ToString();

    var domainEvent = DeserializeEvent(json, eventType);
    await UpdateReadModel(domainEvent); // 写入数据库或缓存

    await args.UpdateCheckpointAsync();
};

await processor.StartProcessingAsync();

登录后复制