使用 redis streams 可构建高性能 .net 事件存储，支持按聚合根分隔离事件流、版本控制与消费者组分发。1. 以 events:{id} 为 key 存储事件流，每条消息含 type、data、timestamp 和 version 字段。2. 利用 stackexchange.redis 写入和读取事件，通过 streamaddasync 和 streamrangeasync 操作数据。3. 创建消费者组实现事件广播，streamreadgroupasync 拉取消息并用 streamacknowledgeasync 确认处理，确保至少一次交付。4. 使用独立版本键或 lua 脚本实现乐观锁，防止并发写冲突。该方案具备持久化、高吞吐与水平扩展能力，适用于事件溯源架构。
用 Redis Streams 构建 .NET 事件存储是一种轻量、高性能的实现方式，特别适合需要持久化事件流并支持消费者组处理的场景。Redis Streams 天然支持消息持久化、多消费者组、消息确认机制，非常适合做事件溯源（Event Sourcing）的基础存储。
1. 设计事件存储的基本结构
事件存储的核心是将领域事件按聚合根（Aggregate Root）分类，追加写入事件流。在 Redis 中，可以用 Stream Key 表示一个聚合实例的事件流，例如：
events:order-12345每个事件作为一条消息写入该 Stream，包含事件类型、时间戳和序列化后的数据。
每条消息字段建议包括：
- type: 事件类型（如 OrderCreated）
- data: JSON 序列化的事件内容
- timestamp: 发生时间
- version: 聚合版本号，用于乐观并发控制
2. 在 .NET 中使用 StackExchange.Redis 操作 Streams
安装 NuGet 包：
Install-Package StackExchange.Redis连接 Redis 并写入事件：
var redis = ConnectionMultiplexer.Connect("localhost");
var db = redis.GetDatabase();
var streamKey = "events:order-12345";
var eventId = await db.StreamAddAsync(streamKey, new NameValueEntry[]
{
new ("type", "OrderCreated"),
new ("data", JsonSerializer.Serialize(new { OrderId = "12345", Amount = 100 })),
new ("timestamp", DateTime.UtcNow.ToString("o")),
new ("version", "1")
});
读取某个聚合的所有事件：
var entries = await db.StreamRangeAsync(streamKey, null, null);
var events = entries.Select(entry => new {
Type = entry["type"],
Data = entry["data"],
Version = int.Parse(entry["version"])
}).ToList();
3. 支持消费者组处理事件（用于事件分发）
如果需要将事件通知给多个服务，可以创建消费者组：
// 创建消费者组（首次执行）
await db.StreamCreateConsumerGroupAsync(streamKey, "payments-service", "$");
// 拉取未处理的消息
var messages = await db.StreamReadGroupAsync(streamKey, "payments-service", "consumer-1", 10, ">");
foreach (var message in messages)
{
// 处理业务逻辑
Console.WriteLine($"Received: {message["type"]}");
// 确认消息处理完成
await db.StreamAcknowledgeAsync(streamKey, "payments-service", message.Id);
}
这种模式支持水平扩展，多个消费者可并行处理不同消息，同时保证每条消息只被组内一个消费者处理。
4. 添加版本控制与并发检查
写入事件前应检查当前版本，防止并发冲突：
可通过 StreamLength 或存储一个单独的版本键（如 version:order-12345）来管理版本号。
写入时先获取当前长度或版本：
var currentVersion = await db.StringGetAsync($"version:{aggregateId}");
if (expectedVersion != (long)currentVersion)
throw new ConcurrencyException();
// 写入事件
await db.StreamAddAsync(streamKey, entries);
// 更新版本
await db.StringSetAsync($"version:{aggregateId}", expectedVersion + 1);
也可通过 Lua 脚本原子化操作：校验版本 + 写入事件 + 自增版本。
基本上就这些。Redis Streams 提供了足够基础的能力，配合 .NET 的序列化和异步处理，可以快速搭建一个高效、可靠的事件存储。关键是设计好 Key 结构、版本控制和消费者组的使用策略。
