TJC.Cyclops.Redis 2025.12.3.2

Cyclops.Redis

项目概述

Cyclops.Redis 是 Cyclops.Framework 中的 Redis 客户端集成组件，提供了强大而灵活的 Redis 操作封装。该组件支持多种 Redis 部署模式（单实例、哨兵、集群），并封装了缓存、队列、分布式锁、发布订阅等常用功能，简化了 Redis 在 .NET 应用程序中的使用。

核心功能模块

1. Redis 连接管理

• 多种部署模式支持：支持单实例（Instance）、哨兵（Sentinel）和集群（Cluster）三种部署模式
• 连接池管理：基于 StackExchange.Redis 实现高效的连接池管理
• 配置灵活：提供丰富的配置选项，支持密码认证、连接超时、重试策略等

// 连接管理核心实现位于 RedisClient 类

2. 缓存操作

• 统一缓存接口：实现 IServiceCache 接口，提供标准化的缓存操作
• 对象序列化：自动处理对象的 JSON 序列化与反序列化
• 过期时间管理：支持设置缓存过期时间和检查过期状态
• 哈希存储：使用 Redis 哈希结构存储缓存数据，便于管理

// RedisCache 类提供完整的缓存功能实现

3. Redis 队列

• 高性能队列：基于 Redis List 实现的高性能分布式队列
• 任务状态跟踪：支持队列任务的状态管理（未开始、处理中等）
• 异步操作：提供同步和异步的队列操作方法
• 超时管理：支持设置队列数据的过期时间

// RedisQueue<T> 类实现了类型化的队列操作

4. 分布式锁

• 安全的锁实现：基于 SET NX PX 命令和 Lua 脚本实现的安全分布式锁
• 锁续期机制：支持锁的续期操作，防止长时间任务执行期间锁过期
• 等待和重试：支持配置等待超时和重试间隔
• 安全释放：使用 Lua 脚本确保只有锁的持有者才能释放锁

// RedisDistributedLock 类实现了安全的分布式锁机制

5. Redis 流（Stream）

• 生产者-消费者模式：提供 Redis Stream 的生产者和消费者封装
• 消息分组处理：支持基于消费者组的消息处理
• 消息持久化：利用 Redis Stream 的持久化特性
• 消息回溯：支持消息的确认和回溯机制

// RedisProducer 和 RedisConsumer 类实现了流操作

6. 发布订阅系统

• 发布者-订阅者模式：提供 Redis 发布订阅的完整封装
• 频道管理：支持多频道的发布和订阅
• 键过期监听：提供键过期事件的监听机制

// RedisPublisher 和 RedisSubscriber 类实现了发布订阅功能
// RedisKeyExpireListener 类实现了键过期监听

技术栈

• .NET 8.0：基于最新的 .NET 平台构建
• StackExchange.Redis：高性能的 Redis 客户端库
• JSON 序列化：内置 JSON 序列化支持
• Lua 脚本：用于实现原子操作（如分布式锁的安全释放）

环境依赖

• .NET 8.0 或更高版本
• Redis 服务器：支持 Redis 3.2 及以上版本（使用流功能需要 Redis 5.0+）

安装与配置

1. 安装包

将 Cyclops.Redis 项目引用添加到您的 .NET 项目中。

2. 配置 Redis 连接

在应用程序配置文件中添加 Redis 连接配置：

{
  "RedisOptions": {
    "Type": "Instance", // 可选值：Instance, Sentinel, Cluster
    "Masters": "localhost:6379",
    "Slaves": "localhost:6380", // 可选，仅 Instance 模式使用
    "Password": "your_password", // 可选
    "DefaultDatabase": 0,
    "ConnectTimeout": 5000,
    "CommandTimeout": 5000,
    "ConnectRetry": 3,
    "KeepAlive": 60,
    "BusyRetry": 3,
    "BusyRetryWaitMS": 500,
    "AllowAdmin": false,
    "ServiceName": "mymaster" // 可选，仅 Sentinel 模式使用
  }
}

代码示例

1. 基本使用 - 缓存操作

// 创建缓存实例
var cache = new RedisCache();

// 设置缓存
cache.Set("user:1", new UserInfo { Id = 1, Name = "张三" });

// 获取缓存
var user = cache.Get<UserInfo>("user:1");

// 删除缓存
cache.Del("user:1");

// 检查缓存是否存在
var exists = cache.ContainsKey("user:1");

2. Redis 队列操作

// 获取 Redis 客户端
var redis = new CyclopsRedis();

// 创建队列实例
var queue = redis.GetRedisQueue<OrderInfo>("order_queue");

// 入队
var order = new OrderInfo { OrderId = "ORD123", Amount = 100.5 };
var queueId = await queue.EnqueueAsync(order);

// 出队
string key;
var dequeuedOrder = await queue.DequeueAsync(out key);

3. 分布式锁使用

// 获取 Redis 客户端
var redis = new CyclopsRedis();

// 创建分布式锁
var distributedLock = redis.CreateDistributedLock("lock:order:123", TimeSpan.FromSeconds(30));

// 尝试获取锁，最多等待5秒，每50ms重试一次
if (await distributedLock.AcquireAsync(TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(50)))
{
    try
    {
        // 执行需要同步的操作
        Console.WriteLine("获取锁成功，执行操作...");
        await Task.Delay(1000);
    }
    finally
    {
        // 释放锁
        await distributedLock.ReleaseAsync();
    }
}
else
{
    Console.WriteLine("获取锁失败");
}

4. Redis 发布订阅

// 获取 Redis 客户端
var redis = new CyclopsRedis();

// 创建订阅者
var subscriber = redis.GetSubscriber("notification_channel");

// 订阅消息
await subscriber.SubscribeAsync((channel, message) => {
    Console.WriteLine($"收到消息: {message}");
});

// 创建发布者
var publisher = redis.GetPublisher("notification_channel");

// 发布消息
await publisher.PublishAsync("Hello, Redis Pub/Sub!");

5. Redis 流操作

// 获取 Redis 客户端
var redis = new CyclopsRedis();

// 创建生产者
var producer = redis.GetRedisProducer("order_stream");

// 发送消息
await producer.ProduceAsync(new Dictionary<string, string>() {
    { "orderId", "ORD456" },
    { "amount", "200.0" }
});

// 创建消费者
var consumer = redis.GetRedisConsumer("order_stream", 0, "order_group");

// 消费消息
var messages = await consumer.ConsumeAsync(count: 10);
foreach (var message in messages)
{
    Console.WriteLine($"消费消息: {message.Id} - {message.Data}");
    // 确认消息
    await consumer.AcknowledgeAsync(message.Id);
}

高级功能

1. 多种连接模式

// 哨兵模式配置示例
var sentinelOptions = new RedisOptions {
    Type = EnumRedisType.Sentinel,
    Masters = "sentinel1:26379,sentinel2:26379",
    ServiceName = "mymaster"
};

// 集群模式配置示例
var clusterOptions = new RedisOptions {
    Type = EnumRedisType.Cluster,
    Masters = "node1:6379,node2:6379,node3:6379"
};

2. 键过期监听

// 获取键过期监听器
var listener = redis.GetKeyExpireListener();

// 订阅键过期事件
await listener.SubscribeAsync((key) => {
    Console.WriteLine($"键过期: {key}");
    // 执行清理或其他操作
});

3. 单例模式使用

// 使用单例实例
var cache = RedisCache.Instance;

// 或使用 CyclopsRedis 单例
var redis = CyclopsRedis.Instance;

性能优化建议

1. 合理设置连接池：根据并发量调整连接池大小
2. 使用异步操作：优先使用异步方法提高并发性能
3. 设置合理的过期时间：避免 Redis 内存无限增长
4. 使用管道和事务：批量操作时使用管道或事务提高性能
5. 注意键命名规范：使用统一的键命名规范，便于管理和查询

故障排查

1. 连接问题：检查 Redis 服务器是否运行，网络是否可达，密码是否正确
2. 超时问题：调整连接超时和命令超时设置
3. 内存问题：监控 Redis 内存使用情况，设置合理的 maxmemory 策略
4. 分布式锁争用：调整锁的过期时间和重试策略

贡献者

• yswenli

许可证

保留所有权利