using Newtonsoft.Json; using StackExchange.Redis; using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Configuration; using System.Linq; using System.Threading.Tasks; namespace XImaging.Automation.Library.HxDriverLib.Tools { public static class RedisConnectionHelp { /// /// 系统自定义Key前缀 /// public static readonly string SysCustomKey = ConfigurationManager.AppSettings["RedisKey"] ?? ""; /// /// 连接字符串 /// private static readonly string RedisConnectionString = ConfigurationManager.AppSettings["RedisHosts"] ?? "127.0.0.1:6379"; private static readonly object Locker = new object(); private static ConnectionMultiplexer _instance; /// /// 线程安全的字典 /// private static readonly ConcurrentDictionary ConnectionCache = new ConcurrentDictionary(); /// /// 单例获取 /// public static ConnectionMultiplexer Instance { get { if (_instance == null) { lock (Locker) { if (_instance == null || !_instance.IsConnected) { _instance = GetManager(); } } } return _instance; } } /// /// 缓存获取 /// /// /// public static ConnectionMultiplexer GetConnectionMultiplexer(string connectionString) { if (!ConnectionCache.ContainsKey(connectionString)) { ConnectionCache[connectionString] = GetManager(connectionString); } return ConnectionCache[connectionString]; } /// /// 获取连接 /// /// /// private static ConnectionMultiplexer GetManager(string connectionString = null) { connectionString = connectionString ?? RedisConnectionString; var connect = ConnectionMultiplexer.Connect(connectionString); //注册如下事件 connect.ConnectionFailed += MuxerConnectionFailed; connect.ConnectionRestored += MuxerConnectionRestored; connect.ErrorMessage += MuxerErrorMessage; connect.ConfigurationChanged += MuxerConfigurationChanged; connect.HashSlotMoved += MuxerHashSlotMoved; connect.InternalError += MuxerInternalError; connect.ConfigurationChangedBroadcast += MuxerConfigurationChangedBroadcast; return connect; } #region 事件 /// /// 重新配置广播时(通常意味着主从同步更改) /// /// /// private static void MuxerConfigurationChangedBroadcast(object sender, EndPointEventArgs e) { Console.WriteLine($"{nameof(MuxerConfigurationChangedBroadcast)}: {e.EndPoint}"); } /// /// 配置更改时 /// /// /// private static void MuxerConfigurationChanged(object sender, EndPointEventArgs e) { Console.WriteLine("Configuration changed: " + e.EndPoint); } /// /// 发生内部错误时(主要用于调试) /// /// /// private static void MuxerErrorMessage(object sender, RedisErrorEventArgs e) { Console.WriteLine("ErrorMessage: " + e.Message); } /// /// 重新建立连接之前的错误 /// /// /// private static void MuxerConnectionRestored(object sender, ConnectionFailedEventArgs e) { Console.WriteLine("ConnectionRestored: " + e.EndPoint); } /// /// 连接失败 , 如果重新连接成功你将不会收到这个通知 /// /// /// private static void MuxerConnectionFailed(object sender, ConnectionFailedEventArgs e) { Console.WriteLine("重新连接:Endpoint failed: " + e.EndPoint + ", " + e.FailureType + (e.Exception == null ? "" : (", " + e.Exception.Message))); } /// /// 更改集群时 /// /// /// private static void MuxerHashSlotMoved(object sender, HashSlotMovedEventArgs e) { Console.WriteLine("HashSlotMoved:NewEndPoint" + e.NewEndPoint + ", OldEndPoint" + e.OldEndPoint); } /// /// redis类库错误 /// /// /// private static void MuxerInternalError(object sender, InternalErrorEventArgs e) { Console.WriteLine("InternalError:Message" + e.Exception.Message); } #endregion } /// /// Redis操作 /// public class RedisStackExchangeHelper { private readonly ConnectionMultiplexer _conn; public string CustomKey; private int DbNum { get; } #region 构造函数 public RedisStackExchangeHelper(int dbNum = 0) : this(dbNum, null) { } public RedisStackExchangeHelper(int dbNum, string readWriteHosts) { DbNum = dbNum; _conn = string.IsNullOrWhiteSpace(readWriteHosts) ? RedisConnectionHelp.Instance : RedisConnectionHelp.GetConnectionMultiplexer(readWriteHosts); } #endregion 构造函数 #region String #region 同步方法 /// /// 保存单个key value /// /// Redis Key /// 保存的值 /// 过期时间 /// public bool StringSet(string key, string value, TimeSpan? expiry = default(TimeSpan?)) { key = AddSysCustomKey(key); return Do(db => db.StringSet(key, value, expiry)); } /// /// 保存多个key value /// /// 键值对 /// public bool StringSet(List> keyValues) { List> newkeyValues = keyValues.Select(p => new KeyValuePair(AddSysCustomKey(p.Key), p.Value)).ToList(); return Do(db => db.StringSet(newkeyValues.ToArray())); } /// /// 保存一个对象 /// /// /// /// /// /// public bool StringSet(string key, T obj, TimeSpan? expiry = default(TimeSpan?)) { key = AddSysCustomKey(key); string json = ConvertJson(obj); return Do(db => db.StringSet(key, json, expiry)); } /// /// 获取单个key的值 /// /// Redis Key /// public string StringGet(string key) { key = AddSysCustomKey(key); return Do(db => db.StringGet(key)); } /// /// 获取多个Key /// /// Redis Key集合 /// public RedisValue[] StringGet(List listKey) { List newKeys = listKey.Select(AddSysCustomKey).ToList(); return Do(db => db.StringGet(ConvertRedisKeys(newKeys))); } /// /// 获取一个key的对象 /// /// /// /// public T StringGet(string key) { key = AddSysCustomKey(key); return Do(db => ConvertObj(db.StringGet(key))); } /// /// 为数字增长val /// /// /// 可以为负 /// 增长后的值 public long StringIncrement(string key, long val = 1) { key = AddSysCustomKey(key); return Do(db => db.StringIncrement(key, val)); } /// /// 为数字减少val /// /// /// 可以为负 /// 减少后的值 public long StringDecrement(string key, long val = 1) { key = AddSysCustomKey(key); return Do(db => db.StringDecrement(key, val)); } #endregion 同步方法 #region 异步方法 /// /// 保存单个key value /// /// Redis Key /// 保存的值 /// 过期时间 /// public async Task StringSetAsync(string key, string value, TimeSpan? expiry = default(TimeSpan?)) { key = AddSysCustomKey(key); return await Do(db => db.StringSetAsync(key, value, expiry)); } /// /// 保存多个key value /// /// 键值对 /// public async Task StringSetAsync(List> keyValues) { List> newkeyValues = keyValues.Select(p => new KeyValuePair(AddSysCustomKey(p.Key), p.Value)).ToList(); return await Do(db => db.StringSetAsync(newkeyValues.ToArray())); } /// /// 保存一个对象 /// /// /// /// /// /// public async Task StringSetAsync(string key, T obj, TimeSpan? expiry = default(TimeSpan?)) { key = AddSysCustomKey(key); string json = ConvertJson(obj); return await Do(db => db.StringSetAsync(key, json, expiry)); } /// /// 获取单个key的值 /// /// Redis Key /// public async Task StringGetAsync(string key) { key = AddSysCustomKey(key); return await Do(db => db.StringGetAsync(key)); } /// /// 获取多个Key /// /// Redis Key集合 /// public async Task StringGetAsync(List listKey) { List newKeys = listKey.Select(AddSysCustomKey).ToList(); return await Do(db => db.StringGetAsync(ConvertRedisKeys(newKeys))); } /// /// 获取一个key的对象 /// /// /// /// public async Task StringGetAsync(string key) { key = AddSysCustomKey(key); string result = await Do(db => db.StringGetAsync(key)); return ConvertObj(result); } /// /// 为数字增长val /// /// /// 可以为负 /// 增长后的值 public async Task StringIncrementAsync(string key, double val = 1) { key = AddSysCustomKey(key); return await Do(db => db.StringIncrementAsync(key, val)); } /// /// 为数字减少val /// /// /// 可以为负 /// 减少后的值 public async Task StringDecrementAsync(string key, double val = 1) { key = AddSysCustomKey(key); return await Do(db => db.StringDecrementAsync(key, val)); } #endregion 异步方法 #endregion String #region Hash #region 同步方法 /// /// 判断某个数据是否已经被缓存 /// /// /// /// public bool HashExists(string key, string dataKey) { key = AddSysCustomKey(key); return Do(db => db.HashExists(key, dataKey)); } /// /// 存储数据到hash表 /// /// /// /// /// /// public bool HashSet(string key, string dataKey, T t) { key = AddSysCustomKey(key); return Do(db => { string json = ConvertJson(t); return db.HashSet(key, dataKey, json); }); } /// /// 移除hash中的某值 /// /// /// /// public bool HashDelete(string key, string dataKey) { key = AddSysCustomKey(key); return Do(db => db.HashDelete(key, dataKey)); } /// /// 移除hash中的多个值 /// /// /// /// public long HashDelete(string key, List dataKeys) { key = AddSysCustomKey(key); //List dataKeys1 = new List() {"1","2"}; return Do(db => db.HashDelete(key, dataKeys.ToArray())); } /// /// 从hash表获取数据 /// /// /// /// /// public T HashGet(string key, string dataKey) { key = AddSysCustomKey(key); return Do(db => { string value = db.HashGet(key, dataKey); return ConvertObj(value); }); } /// /// 为数字增长val /// /// /// /// 可以为负 /// 增长后的值 public double HashIncrement(string key, string dataKey, double val = 1) { key = AddSysCustomKey(key); return Do(db => db.HashIncrement(key, dataKey, val)); } /// /// 为数字减少val /// /// /// /// 可以为负 /// 减少后的值 public double HashDecrement(string key, string dataKey, double val = 1) { key = AddSysCustomKey(key); return Do(db => db.HashDecrement(key, dataKey, val)); } /// /// 获取hashkey所有Redis key /// /// /// /// public List HashKeys(string key) { key = AddSysCustomKey(key); return Do(db => { RedisValue[] values = db.HashKeys(key); return ConvetList(values); }); } #endregion 同步方法 #region 异步方法 /// /// 判断某个数据是否已经被缓存 /// /// /// /// public async Task HashExistsAsync(string key, string dataKey) { key = AddSysCustomKey(key); return await Do(db => db.HashExistsAsync(key, dataKey)); } /// /// 存储数据到hash表 /// /// /// /// /// /// public async Task HashSetAsync(string key, string dataKey, T t) { key = AddSysCustomKey(key); return await Do(db => { string json = ConvertJson(t); return db.HashSetAsync(key, dataKey, json); }); } /// /// 移除hash中的某值 /// /// /// /// public async Task HashDeleteAsync(string key, string dataKey) { key = AddSysCustomKey(key); return await Do(db => db.HashDeleteAsync(key, dataKey)); } /// /// 移除hash中的多个值 /// /// /// /// public async Task HashDeleteAsync(string key, List dataKeys) { key = AddSysCustomKey(key); //List dataKeys1 = new List() {"1","2"}; return await Do(db => db.HashDeleteAsync(key, dataKeys.ToArray())); } /// /// 从hash表获取数据 /// /// /// /// /// public async Task HashGeAsync(string key, string dataKey) { key = AddSysCustomKey(key); string value = await Do(db => db.HashGetAsync(key, dataKey)); return ConvertObj(value); } /// /// 为数字增长val /// /// /// /// 可以为负 /// 增长后的值 public async Task HashIncrementAsync(string key, string dataKey, double val = 1) { key = AddSysCustomKey(key); return await Do(db => db.HashIncrementAsync(key, dataKey, val)); } /// /// 为数字减少val /// /// /// /// 可以为负 /// 减少后的值 public async Task HashDecrementAsync(string key, string dataKey, double val = 1) { key = AddSysCustomKey(key); return await Do(db => db.HashDecrementAsync(key, dataKey, val)); } /// /// 获取hashkey所有Redis key /// /// /// /// public async Task> HashKeysAsync(string key) { key = AddSysCustomKey(key); RedisValue[] values = await Do(db => db.HashKeysAsync(key)); return ConvetList(values); } #endregion 异步方法 #endregion Hash #region List #region 同步方法 /// /// 移除指定ListId的内部List的值 /// /// /// public void ListRemove(string key, T value) { key = AddSysCustomKey(key); Do(db => db.ListRemove(key, ConvertJson(value))); } /// /// 获取指定key的List /// /// /// public List ListRange(string key) { key = AddSysCustomKey(key); return Do(redis => { var values = redis.ListRange(key); return ConvetList(values); }); } /// /// 入队 /// /// /// public void ListRightPush(string key, T value) { key = AddSysCustomKey(key); Do(db => db.ListRightPush(key, ConvertJson(value))); } /// /// 出队 /// /// /// /// public T ListRightPop(string key) { key = AddSysCustomKey(key); return Do(db => { var value = db.ListRightPop(key); return ConvertObj(value); }); } /// /// 入栈 /// /// /// /// public void ListLeftPush(string key, T value) { key = AddSysCustomKey(key); Do(db => db.ListLeftPush(key, ConvertJson(value))); } /// /// 出栈 /// /// /// /// public T ListLeftPop(string key) { key = AddSysCustomKey(key); return Do(db => { var value = db.ListLeftPop(key); return ConvertObj(value); }); } /// /// 获取集合中的数量 /// /// /// public long ListLength(string key) { key = AddSysCustomKey(key); return Do(redis => redis.ListLength(key)); } #endregion 同步方法 #region 异步方法 /// /// 移除指定ListId的内部List的值 /// /// /// public async Task ListRemoveAsync(string key, T value) { key = AddSysCustomKey(key); return await Do(db => db.ListRemoveAsync(key, ConvertJson(value))); } /// /// 获取指定key的List /// /// /// public async Task> ListRangeAsync(string key) { key = AddSysCustomKey(key); var values = await Do(redis => redis.ListRangeAsync(key)); return ConvetList(values); } /// /// 入队 /// /// /// public async Task ListRightPushAsync(string key, T value) { key = AddSysCustomKey(key); return await Do(db => db.ListRightPushAsync(key, ConvertJson(value))); } /// /// 出队 /// /// /// /// public async Task ListRightPopAsync(string key) { key = AddSysCustomKey(key); var value = await Do(db => db.ListRightPopAsync(key)); return ConvertObj(value); } /// /// 入栈 /// /// /// /// public async Task ListLeftPushAsync(string key, T value) { key = AddSysCustomKey(key); return await Do(db => db.ListLeftPushAsync(key, ConvertJson(value))); } /// /// 出栈 /// /// /// /// public async Task ListLeftPopAsync(string key) { key = AddSysCustomKey(key); var value = await Do(db => db.ListLeftPopAsync(key)); return ConvertObj(value); } /// /// 获取集合中的数量 /// /// /// public async Task ListLengthAsync(string key) { key = AddSysCustomKey(key); return await Do(redis => redis.ListLengthAsync(key)); } #endregion 异步方法 #endregion List #region SortedSet 有序集合 #region 同步方法 /// /// 添加 /// /// /// /// public bool SortedSetAdd(string key, T value, double score) { key = AddSysCustomKey(key); return Do(redis => redis.SortedSetAdd(key, ConvertJson(value), score)); } /// /// 删除 /// /// /// public bool SortedSetRemove(string key, T value) { key = AddSysCustomKey(key); return Do(redis => redis.SortedSetRemove(key, ConvertJson(value))); } /// /// 获取全部 /// /// /// public List SortedSetRangeByRank(string key) { key = AddSysCustomKey(key); return Do(redis => { var values = redis.SortedSetRangeByRank(key); return ConvetList(values); }); } /// /// 获取集合中的数量 /// /// /// public long SortedSetLength(string key) { key = AddSysCustomKey(key); return Do(redis => redis.SortedSetLength(key)); } #endregion 同步方法 #region 异步方法 /// /// 添加 /// /// /// /// public async Task SortedSetAddAsync(string key, T value, double score) { key = AddSysCustomKey(key); return await Do(redis => redis.SortedSetAddAsync(key, ConvertJson(value), score)); } /// /// 删除 /// /// /// public async Task SortedSetRemoveAsync(string key, T value) { key = AddSysCustomKey(key); return await Do(redis => redis.SortedSetRemoveAsync(key, ConvertJson(value))); } /// /// 获取全部 /// /// /// public async Task> SortedSetRangeByRankAsync(string key) { key = AddSysCustomKey(key); var values = await Do(redis => redis.SortedSetRangeByRankAsync(key)); return ConvetList(values); } /// /// 获取集合中的数量 /// /// /// public async Task SortedSetLengthAsync(string key) { key = AddSysCustomKey(key); return await Do(redis => redis.SortedSetLengthAsync(key)); } #endregion 异步方法 #endregion SortedSet 有序集合 #region key #region 同步 /// /// 删除单个key /// /// redis key /// 是否删除成功 public bool KeyDelete(string key) { key = AddSysCustomKey(key); return Do(db => db.KeyDelete(key)); } /// /// 删除多个key /// /// rediskey /// 成功删除的个数 public long KeyDelete(List keys) { List newKeys = keys.Select(AddSysCustomKey).ToList(); return Do(db => db.KeyDelete(ConvertRedisKeys(newKeys))); } /// /// 判断key是否存储 /// /// redis key /// public bool KeyExists(string key) { key = AddSysCustomKey(key); return Do(db => db.KeyExists(key)); } /// /// 重新命名key /// /// 就的redis key /// 新的redis key /// public bool KeyRename(string key, string newKey) { key = AddSysCustomKey(key); return Do(db => db.KeyRename(key, newKey)); } /// /// 设置Key的时间 /// /// redis key /// /// public bool KeyExpire(string key, TimeSpan? expiry = default(TimeSpan?)) { key = AddSysCustomKey(key); return Do(db => db.KeyExpire(key, expiry)); } #endregion #region 异步 /// /// 删除单个key /// /// redis key /// 是否删除成功 public async Task KeyDeleteAsync(string key) { key = AddSysCustomKey(key); return await Do(db => db.KeyDeleteAsync(key)); } /// /// 删除多个key /// /// rediskey /// 成功删除的个数 public async Task KeyDeleteAsync(List keys) { List newKeys = keys.Select(AddSysCustomKey).ToList(); return await Do(db => db.KeyDeleteAsync(ConvertRedisKeys(newKeys))); } /// /// 判断key是否存储 /// /// redis key /// public async Task KeyExistsAsync(string key) { key = AddSysCustomKey(key); return await Do(db => db.KeyExistsAsync(key)); } /// /// 重新命名key /// /// 就的redis key /// 新的redis key /// public async Task KeyRenameAsync(string key, string newKey) { key = AddSysCustomKey(key); return await Do(db => db.KeyRenameAsync(key, newKey)); } /// /// 设置Key的时间 /// /// redis key /// /// public async Task KeyExpireAsync(string key, TimeSpan? expiry = default(TimeSpan?)) { key = AddSysCustomKey(key); return await Do(db => db.KeyExpireAsync(key, expiry)); } #endregion #endregion key #region 发布订阅 #region 同步 /// /// 订阅 /// /// /// public void Subscribe(string subChannel, Action handler = null) { ISubscriber sub = _conn.GetSubscriber(); sub.Subscribe(subChannel, (channel, message) => { if (handler == null) { Console.WriteLine(subChannel + " 订阅收到消息:" + message); } else { handler(channel, message); } }); } /// /// 发布 /// /// /// /// /// public long Publish(string channel, T msg) { ISubscriber sub = _conn.GetSubscriber(); return sub.Publish(channel, ConvertJson(msg)); } /// /// 取消订阅 /// /// public void Unsubscribe(string channel) { ISubscriber sub = _conn.GetSubscriber(); sub.Unsubscribe(channel); } /// /// 取消全部订阅 /// public void UnsubscribeAll() { ISubscriber sub = _conn.GetSubscriber(); sub.UnsubscribeAll(); } #endregion 发布订阅 #region 异步 /// /// 订阅 /// /// /// public async Task SubscribeAsync(string subChannel, Action handler = null) { ISubscriber sub = _conn.GetSubscriber(); await sub.SubscribeAsync(subChannel, (channel, message) => { if (handler == null) { Console.WriteLine(subChannel + " 订阅收到消息:" + message); } else { handler(channel, message); } }); } /// /// 发布 /// /// /// /// /// public async Task PublishAsync(string channel, T msg) { ISubscriber sub = _conn.GetSubscriber(); return await sub.PublishAsync(channel, ConvertJson(msg)); } /// /// 取消订阅 /// /// public async Task UnsubscribeAsync(string channel) { ISubscriber sub = _conn.GetSubscriber(); await sub.UnsubscribeAsync(channel); } /// /// 取消全部订阅 /// public async Task UnsubscribeAllAsync() { ISubscriber sub = _conn.GetSubscriber(); await sub.UnsubscribeAllAsync(); } #endregion 发布订阅 #endregion #region 其他 public ITransaction CreateTransaction() { return GetDatabase().CreateTransaction(); } public IDatabase GetDatabase() { return _conn.GetDatabase(DbNum); } public IServer GetServer(string hostAndPort) { return _conn.GetServer(hostAndPort); } /// /// 设置前缀 /// /// public void SetSysCustomKey(string customKey) { CustomKey = customKey; } #endregion 其他 #region 辅助方法 private string AddSysCustomKey(string oldKey) { var prefixKey = CustomKey ?? RedisConnectionHelp.SysCustomKey; return prefixKey + oldKey; } private T Do(Func func) { var database = _conn.GetDatabase(DbNum); return func(database); } private string ConvertJson(T value) { string result = value is string ? value.ToString() : JsonConvert.SerializeObject(value); return result; } private T ConvertObj(RedisValue value) { if (value.IsNullOrEmpty) return default(T); return JsonConvert.DeserializeObject(value); } private List ConvetList(RedisValue[] values) { List result = new List(); foreach (var item in values) { var model = ConvertObj(item); result.Add(model); } return result; } private RedisKey[] ConvertRedisKeys(List redisKeys) { return redisKeys.Select(redisKey => (RedisKey)redisKey).ToArray(); } /// /// 转化为timespan /// /// 过期时间,秒 /// public TimeSpan ToTimeSpan(double expTime) { return DateTime.Now.AddSeconds(expTime) - DateTime.Now; } #endregion 辅助方法 } }