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

许可证

保留所有权利