StackExchange.Redis中的Redis Streams操作指南

Redis Streams概述

Redis Streams是Redis 5.0引入的一种新型数据结构，它本质上是一个只追加的日志结构。与Redis的其他数据结构不同，Streams特别适合处理消息队列、事件溯源等场景。StackExchange.Redis作为.NET平台最流行的Redis客户端，完整实现了Redis Streams的所有相关命令。

Streams的核心特点包括：

• 每条消息都有唯一的ID
• 支持消息的消费者组模式
• 可以按时间范围或ID范围查询消息
• 支持消息的持久化和复制

写入Streams

基本写入操作

在StackExchange.Redis中，使用StreamAdd方法向Stream添加消息：

var db = redis.GetDatabase();
var messageId = db.StreamAdd("events_stream", "field1", "value1");

每条消息可以包含多个字段-值对：

var values = new NameValueEntry[]
{
    new NameValueEntry("user_id", "1001"),
    new NameValueEntry("action", "login"),
    new NameValueEntry("ip", "192.168.1.1")
};

var messageId = db.StreamAdd("user_activity", values);

高级写入选项

StreamAdd方法提供了多个可选参数：

// 自定义消息ID并限制Stream长度
db.StreamAdd("events_stream", "field1", "value1", 
    messageId: "custom-id-123", 
    maxLength: 1000);

参数说明：

• messageId：自定义消息ID，格式通常为"<毫秒时间戳>-<序列号>"
• maxLength：Stream最大长度，超过时会自动修剪旧消息

读取Streams

基本读取操作

使用StreamRead从指定位置开始读取：

// 从ID "0-0"开始读取所有消息
var messages = db.StreamRead("events_stream", "0-0");

批量读取和多Stream读取

// 从多个Stream读取
var streams = db.StreamRead(new StreamPosition[]
{
    new StreamPosition("stream1", "0-0"),
    new StreamPosition("stream2", "0-0")
}, countPerStream: 10);

范围查询

使用StreamRange可以查询特定范围内的消息：

// 查询所有消息（"-"表示最小ID，"+"表示最大ID）
var allMessages = db.StreamRange("events_stream");

// 查询最新的100条消息，按时间倒序
var recentMessages = db.StreamRange("events_stream",
    count: 100,
    messageOrder: Order.Descending);

Stream信息查询

获取Stream的元数据信息：

var info = db.StreamInfo("events_stream");

Console.WriteLine($"总消息数: {info.Length}");
Console.WriteLine($"第一条消息ID: {info.FirstEntry.Id}");
Console.WriteLine($"最后一条消息ID: {info.LastEntry.Id}");
Console.WriteLine($"消费者组数量: {info.ConsumerGroupCount}");

消费者组操作

创建消费者组

// 从最新消息开始消费
db.StreamCreateConsumerGroup("events_stream", "cg1", "$");

// 从最早的消息开始消费
db.StreamCreateConsumerGroup("events_stream", "cg2", "0-0");

消费消息

// 消费者1读取5条新消息
var messages1 = db.StreamReadGroup("events_stream", "cg1", "consumer1", ">", count: 5);

// 消费者2读取5条新消息
var messages2 = db.StreamReadGroup("events_stream", "cg1", "consumer2", ">", count: 5);

处理待处理消息

// 获取待处理消息信息
var pendingInfo = db.StreamPending("events_stream", "cg1");

// 获取特定消费者的待处理消息详情
var pendingMessages = db.StreamPendingMessages("events_stream", "cg1",
    count: 10,
    consumerName: "consumer1");

// 确认消息处理完成
foreach(var msg in pendingMessages)
{
    db.StreamAcknowledge("events_stream", "cg1", msg.MessageId);
}

消息所有权转移

// 将consumer1的待处理消息转移给consumer2
var messages = db.StreamPendingMessages("events_stream", "cg1",
    count: 5,
    consumerName: "consumer1");

db.StreamClaim("events_stream", "cg1", "consumer2", 0,
    messages.Select(m => m.MessageId).ToArray());

最佳实践

1. 消息ID生成：通常使用自动生成的ID（时间戳+序列号），除非有特殊需求

2. 消费者组设计：

   • 每个逻辑消费者应该有自己的消费者名称
   • 合理设置消费者组的起始位置
   • 及时确认已处理的消息

3. 性能考虑：

   • 批量读取消息（使用count参数）
   • 定期检查待处理消息，避免堆积
   • 对于长时间运行的任务，考虑使用消息所有权转移机制

4. 错误处理：

   • 处理消息时应该捕获异常
   • 实现重试机制
   • 考虑死信队列处理无法处理的消息

通过StackExchange.Redis提供的丰富API，开发者可以高效地利用Redis Streams构建可靠的消息处理系统。