如何用 Redis Streams 构建 .NET 事件存储？

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 行业资讯

如何用 Redis Streams 构建 .NET 事件存储？

2025-10-10

浏览次数：次

返回列表

使用 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 构建 .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")
});

读取某个聚合的所有事件：

AdMaker AI

从0到爆款高转化AI广告生成器

65 查看详情 AdMaker AI

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);