一、参考的博文javascript
a : http://www.cnblogs.com/lori/archive/2012/04/12/2443708.html —— 主要的实现思路html
b: http://www.cnblogs.com/liqingwen/archive/2017/04/06/6672452.html —— RedisHelper 类java
c : https://www.cnblogs.com/stopfalling/p/5375492.html —— 应用场景说明jquery
二、原理说明web
博文a 中的老师,提供了Redis 实现消息队列的总体思路,言简意赅,但部分类库a 老师并未提供,所以我参照了博文b 中老师的RedisHelper 类,主要借鉴的方法为ListLeftPop及ListRightPush,及实现消息队列的核心思想,先进先出。redis
博文c 中老师详细介绍了几种消息队列的 业务场景,是我所看的全部业务场景描述中最为详细,清晰的,结合博文a 老师的总体思路,对消息队列的实现上有了较为清晰的认识。数据库
总体思路在博文a 中老师已经介绍,及依靠 mvc 框架,经过web端用户提交事件,实现消息入列,经过定时器(Timer)实现按照时间间隔的消息出列,Redis 做为存储媒介,存储消json
息内容。服务器
三、具体实现并发
3.1 前期准备
(1)开发工具:VS 2017;redis-desktop-manager 客户端;Redis版本:Redis-x64-3.2.100
(2)新建MVC5 框架,配置 web.config 文件,RouteConfig文件
a、新建MVC5 框架
b、配置web.config 文件(Default key 主要用于区别不一样的key值,此处使用了解决方法名称)
c、配置路由(使用经典模式便可,路由的详细配置说明可参见个人另外一篇博文 )
(3)目前.net 框架中支持Redis 的dll包括ServiceStackRedis 及 StackExchange,前者下载地址在Git hub 上,但该类库在6000并发后会抛出license exception 的异常。所以
我这次使用的dll 为基于微软的StackExchage dll,该dll的下载方式是,在NuGet中,找到StackExchage.Redis 类包,下载便可,见下图。
(4)建立消息实体对象,以下图所示
(此处userId 为 字符型,int 型在序列化时,会出现异常,请注意。)
3.2 引入Redis 类库,此部分未作大量修改,代码拷贝自 博文b
using System; using System.Linq; using StackExchange.Redis; using System.Configuration; using System.Collections.Generic; using System.Runtime.Serialization.Formatters.Binary; using System.IO; using System.Threading.Tasks; using System.Runtime.Serialization; namespace MVC5Project.Redis.BaseOnStackExchage { public class RedisHelper { /// <summary> /// 链接字符串 /// </summary> private static readonly string ConnectionString; /// <summary> /// redis 链接对象 /// </summary> private static IConnectionMultiplexer _connMultiplexer; /// <summary> /// 默认的key值(用来看成RedisKey的前缀)【此部分为自行修改的,无心义】 /// </summary> public static string DefaultKey { get; private set; } /// <summary> /// 锁 /// </summary> private static readonly object Locker = new object(); /// <summary> /// 数据库访问对象 /// </summary> private readonly IDatabase _db; /// <summary> /// 采用双重锁单例模式,保证数据访问对象有且仅有一个 /// </summary> /// <returns></returns> public IConnectionMultiplexer GetConnectionRedisMultiplexer() { if ((_connMultiplexer == null || !_connMultiplexer.IsConnected)) { lock (Locker) { if ((_connMultiplexer == null || !_connMultiplexer.IsConnected)) { _connMultiplexer = ConnectionMultiplexer.Connect(ConnectionString); } } } return _connMultiplexer; } /// <summary> /// 添加事务处理 /// </summary> /// <returns></returns> public ITransaction GetTransaction() { //建立事务 return _db.CreateTransaction(); } /// <summary> /// 静态的构造函数, /// 构造函数是属于类的,而不是属于实例的 /// 就是说这个构造函数只会被执行一次。也就是在建立第一个实例或引用任何静态成员以前,由.NET自动调用。 /// </summary> static RedisHelper() { ConnectionString = ConfigurationManager.ConnectionStrings["RedisConnectionString"].ConnectionString; _connMultiplexer = ConnectionMultiplexer.Connect(ConnectionString); DefaultKey = ConfigurationManager.AppSettings["Redis.DefaultKey"]; RegisterEvent(); } /// <summary> /// 重载构造器 /// </summary> /// <param name="db"></param> public RedisHelper(int db = -1) { _db = _connMultiplexer.GetDatabase(db); } #region private method /// <summary> /// 添加 key 的前缀 /// </summary> /// <param name="key"></param> /// <returns></returns> private static string AddKeyPrefix(string key) { return $"{DefaultKey}:{key}"; } /// <summary> /// 序列化 /// </summary> /// <param name="obj"></param> /// <returns></returns> private static byte[] Serialize(object obj) { try { if (obj == null) return null; var binaryFormatter = new BinaryFormatter(); using (var memoryStream = new MemoryStream()) { binaryFormatter.Serialize(memoryStream, obj); var data = memoryStream.ToArray(); return data; } } catch (SerializationException ex) { throw ex; } } /// <summary> /// 反序列化 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="data"></param> /// <returns></returns> private static T Deserialize<T>(byte[] data) { if (data == null) return default(T); var binaryFormatter = new BinaryFormatter(); using (var memoryStream = new MemoryStream(data)) { var result = (T)binaryFormatter.Deserialize(memoryStream); return result; } } #endregion #region stringGet /// <summary> /// 设置key,并保存字符串(若是key 已存在,则覆盖) /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <param name="expried"></param> /// <returns></returns> public bool StringSet(string redisKey, string redisValue, TimeSpan? expried = null) { redisKey = AddKeyPrefix(redisKey); return _db.StringSet(redisKey, redisValue, expried); } /// <summary> /// 保存多个key-value /// </summary> /// <param name="keyValuePairs"></param> /// <returns></returns> public bool StringSet(IEnumerable<KeyValuePair<RedisKey, RedisValue>> keyValuePairs) { keyValuePairs = keyValuePairs.Select(x => new KeyValuePair<RedisKey, RedisValue>(AddKeyPrefix(x.Key), x.Value)); return _db.StringSet(keyValuePairs.ToArray()); } /// <summary> /// 获取字符串 /// </summary> /// <param name="redisKey"></param> /// <param name="expired"></param> /// <returns></returns> public string StringGet(string redisKey, TimeSpan? expired = null) { try { redisKey = AddKeyPrefix(redisKey); return _db.StringGet(redisKey); } catch (TypeAccessException ex) { throw ex; } } /// <summary> /// 存储一个对象,该对象会被序列化存储 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <param name="expired"></param> /// <returns></returns> public bool StringSet<T>(string redisKey, T redisValue, TimeSpan? expired = null) { redisKey = AddKeyPrefix(redisKey); var json = Serialize(redisKey); return _db.StringSet(redisKey, json, expired); } /// <summary> /// 获取一个对象(会进行反序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="expired"></param> /// <returns></returns> public T StringSet<T>(string redisKey, TimeSpan? expired = null) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(_db.StringGet(redisKey)); } /// <summary> /// 保存一个字符串值 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <param name="expired"></param> /// <returns></returns> public async Task<bool> StringSetAsync(string redisKey, string redisValue, TimeSpan? expired = null) { redisKey = AddKeyPrefix(redisKey); return await _db.StringSetAsync(redisKey, redisValue, expired); } /// <summary> /// 保存一个字符串值 /// </summary> /// <param name="keyValuePairs"></param> /// <returns></returns> public async Task<bool> StringSetAsync(IEnumerable<KeyValuePair<RedisKey, RedisValue>> keyValuePairs) { keyValuePairs = keyValuePairs.Select(x => new KeyValuePair<RedisKey, RedisValue>(AddKeyPrefix(x.Key), x.Value)); return await _db.StringSetAsync(keyValuePairs.ToArray()); } /// <summary> /// 获取单个值 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <param name="expired"></param> /// <returns></returns> public async Task<string> StringGetAsync(string redisKey, string redisValue, TimeSpan? expired = null) { redisKey = AddKeyPrefix(redisKey); return await _db.StringGetAsync(redisKey); } /// <summary> /// 存储一个对象(该对象会被序列化保存) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <param name="expired"></param> /// <returns></returns> public async Task<bool> StringSetAsync<T>(string redisKey, string redisValue, TimeSpan? expired = null) { redisKey = AddKeyPrefix(redisKey); var json = Serialize(redisValue); return await _db.StringSetAsync(redisKey, json, expired); } /// <summary> /// 获取一个对象(反序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <param name="expired"></param> /// <returns></returns> public async Task<T> StringGetAsync<T>(string redisKey, string redisValue, TimeSpan? expired = null) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(await _db.StringGetAsync(redisKey)); } #endregion #region string operation /// <summary> /// 判断字段是否在hash中 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public bool HashExist(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return _db.HashExists(redisKey, hashField); } /// <summary> /// 从hash 中删除字段 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public bool HashDelete(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return _db.HashDelete(redisKey, hashField); } /// <summary> /// 从hash中移除指定字段 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public long HashDelete(string redisKey, IEnumerable<RedisValue> hashField) { redisKey = AddKeyPrefix(redisKey); return _db.HashDelete(redisKey, hashField.ToArray()); } /// <summary> /// 在hash中设定值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <param name="value"></param> /// <returns></returns> public bool HashSet(string redisKey, string hashField, string value) { redisKey = AddKeyPrefix(redisKey); return _db.HashSet(redisKey, hashField, value); } /// <summary> /// 从Hash 中获取值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public RedisValue HashGet(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return _db.HashGet(redisKey, hashField); } /// <summary> /// 从Hash 中获取值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public RedisValue[] HashGet(string redisKey, RedisValue[] hashField) { redisKey = AddKeyPrefix(redisKey); return _db.HashGet(redisKey, hashField); } /// <summary> /// 从hash 返回全部的key值 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public IEnumerable<RedisValue> HashKeys(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.HashKeys(redisKey); } /// <summary> /// 根据key返回hash中的值 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public RedisValue[] HashValues(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.HashValues(redisKey); } /// <summary> /// /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <param name="value"></param> /// <returns></returns> public bool HashSet<T>(string redisKey, string hashField, T value) { redisKey = AddKeyPrefix(redisKey); var json = Serialize(value); return _db.HashSet(redisKey, hashField, json); } /// <summary> /// 在hash 中获取值 (反序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public T HashGet<T>(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(_db.HashGet(redisKey, hashField)); } /// <summary> /// 判断字段是否存在hash 中 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public async Task<bool> HashExistsAsync(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return await _db.HashExistsAsync(redisKey, hashField); } /// <summary> /// 从hash中移除指定字段 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public async Task<bool> HashDeleteAsync(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return await _db.HashDeleteAsync(redisKey, hashField); } /// <summary> /// 从hash中移除指定字段 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public async Task<long> HashDeleteAsync(string redisKey, IEnumerable<RedisValue> hashField) { redisKey = AddKeyPrefix(redisKey); return await _db.HashDeleteAsync(redisKey, hashField.ToArray()); } /// <summary> /// 在hash 设置值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <param name="value"></param> /// <returns></returns> public async Task<bool> HashSetAsync(string redisKey, string hashField, string value) { redisKey = AddKeyPrefix(redisKey); return await _db.HashSetAsync(redisKey, hashField, value); } /// <summary> /// 在hash 中设定值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashFields"></param> /// <returns></returns> public async Task HashSetAsync(string redisKey, IEnumerable<HashEntry> hashFields) { redisKey = AddKeyPrefix(redisKey); await _db.HashSetAsync(redisKey, hashFields.ToArray()); } /// <summary> /// 在hash 中设定值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public async Task<RedisValue> HashGetAsync(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return await _db.HashGetAsync(redisKey, hashField); } /// <summary> /// 在hash 中获取值 /// </summary> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <param name="value"></param> /// <returns></returns> public async Task<IEnumerable<RedisValue>> HashGetAsync(string redisKey, RedisValue[] hashField, string value) { redisKey = AddKeyPrefix(redisKey); return await _db.HashGetAsync(redisKey, hashField); } /// <summary> /// 从hash返回全部的字段值 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<IEnumerable<RedisValue>> HashKeysAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.HashKeysAsync(redisKey); } /// <summary> /// 返回hash中全部的值 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<IEnumerable<RedisValue>> HashValuesAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.HashValuesAsync(redisKey); } /// <summary> /// 在hash 中设定值(序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <param name="value"></param> /// <returns></returns> public async Task<bool> HashSetAsync<T>(string redisKey, string hashField, T value) { redisKey = AddKeyPrefix(redisKey); var json = Serialize(value); return await _db.HashSetAsync(redisKey, hashField, json); } /// <summary> /// 在hash中获取值(反序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="hashField"></param> /// <returns></returns> public async Task<T> HashGetAsync<T>(string redisKey, string hashField) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(await _db.HashGetAsync(redisKey, hashField)); } #endregion #region list operation /// <summary> /// 移除并返回key所对应列表的第一个元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public string ListLeftPop(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.ListLeftPop(redisKey); } /// <summary> /// 移除并返回key所对应列表的最后一个元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public string ListRightPop(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.ListRightPop(redisKey); } /// <summary> /// 移除指定key及key所对应的元素 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public long ListRemove(string redisKey, string redisValue) { redisKey = AddKeyPrefix(redisKey); return _db.ListRemove(redisKey, redisValue); } /// <summary> /// 在列表尾部插入值,若是键不存在,先建立再插入值 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public long ListRightPush(string redisKey, string redisValue) { redisKey = AddKeyPrefix(redisKey); return _db.ListRightPush(redisKey, redisValue); } /// <summary> /// 在列表头部插入值,若是键不存在,先建立再插入值 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public long ListLeftPush(string redisKey, string redisValue) { redisKey = AddKeyPrefix(redisKey); return _db.ListLeftPush(redisKey, redisValue); } /// <summary> /// 返回列表上该键的长度,若是不存在,返回0 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public long ListLength(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.ListLength(redisKey); } /// <summary> /// 返回在该列表上键所对应的元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public IEnumerable<RedisValue> ListRange(string redisKey) { try { redisKey = AddKeyPrefix(redisKey); return _db.ListRange(redisKey); } catch (Exception ex) { throw ex; } } /// <summary> /// 移除并返回存储在该键列表的第一个元素 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <returns></returns> public T ListLeftPop<T>(string redisKey) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(_db.ListLeftPop(redisKey)); } /// <summary> /// 移除并返回该列表上的最后一个元素 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <returns></returns> public T ListRightPop<T>(string redisKey) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(_db.ListRightPop(redisKey)); } /// <summary> /// 在列表尾部插入值,若是键不存在,先建立再插入值 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public long ListRightPush<T>(string redisKey, T redisValue) { redisKey = AddKeyPrefix(redisKey); return _db.ListRightPush(redisKey, Serialize(redisValue)); } /// <summary> /// 在列表头部插入值,若是键不存在,建立后插入值 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public long ListLeftPush<T>(string redisKey, T redisValue) { redisKey = AddKeyPrefix(redisKey); return _db.ListRightPush(redisKey, Serialize(redisValue)); } /// <summary> /// 移除并返回存储在该键列表的第一个元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<string> ListLeftPopAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.ListLeftPopAsync(redisKey); } /// <summary> /// 移除并返回存储在该键列表的最后一个元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<string> ListRightPopAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.ListRightPopAsync(redisKey); } /// <summary> /// 移除列表指定键上与值相同的元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<long> ListRemoveAsync(string redisKey,string redisValue) { redisKey = AddKeyPrefix(redisKey); return await _db.ListRemoveAsync(redisKey, redisValue); } /// <summary> /// 在列表尾部差入值,若是键不存在,先建立后插入 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public async Task<long> ListRightPushAsync(string redisKey,string redisValue) { redisKey = AddKeyPrefix(redisKey); return await ListRightPushAsync(redisKey, redisValue); } /// <summary> /// 在列表头部插入值,若是键不存在,先建立后插入 /// </summary> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public async Task<long> ListLeftPushAsync(string redisKey,string redisValue) { redisKey = AddKeyPrefix(redisKey); return await _db.ListLeftPushAsync(redisKey, redisValue); } /// <summary> /// 返回列表上的长度,若是不存在,返回0 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<long> ListLengthAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.ListLengthAsync(redisKey); } /// <summary> /// 返回在列表上键对应的元素 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<IEnumerable<RedisValue>> ListRangeAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.ListRangeAsync(redisKey); } /// <summary> /// 移除并返回存储在key对应列表的第一个元素 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <returns></returns> public async Task<T> ListLeftPopAsync<T>(string redisKey) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(await _db.ListLeftPopAsync(redisKey)); } /// <summary> /// 移除并返回存储在key 对应列表的最后一个元素 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <returns></returns> public async Task<T> ListRightPopAsync<T>(string redisKey) { redisKey = AddKeyPrefix(redisKey); return Deserialize<T>(await _db.ListRightPopAsync(redisKey)); } /// <summary> /// 在列表尾部插入值,若是值不存在,先建立后写入值 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public async Task<long> ListRightPushAsync<T>(string redisKey,string redisValue) { redisKey = AddKeyPrefix(redisKey); return await _db.ListRightPushAsync(redisKey, Serialize(redisValue)); } /// <summary> /// 在列表头部插入值,若是值不存在,先建立后写入值 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="redisValue"></param> /// <returns></returns> public async Task<long> ListLeftPushAsync<T>(string redisKey,string redisValue) { redisKey = AddKeyPrefix(redisKey); return await _db.ListLeftPushAsync(redisKey, Serialize(redisValue)); } #endregion #region sorted set operation /// <summary> /// sortedset 新增 /// </summary> /// <param name="redisKey"></param> /// <param name="member"></param> /// <param name="score"></param> /// <returns></returns> public bool SortedSetAdd(string redisKey,string member,double score) { redisKey = AddKeyPrefix(redisKey); return _db.SortedSetAdd(redisKey, member, score); } /// <summary> /// 在有序集合中返回指定范围的元素,默认状况下由低到高 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public IEnumerable<RedisValue> SortedSetRangeByRank(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.SortedSetRangeByRank(redisKey); } /// <summary> /// 返回有序集合的个数 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public long SortedSetLength(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.SortedSetLength(redisKey); } /// <summary> /// 返回有序集合的元素个数 /// </summary> /// <param name="redisKey"></param> /// <param name="member"></param> /// <returns></returns> public bool SortedSetLength(string redisKey,string member) { redisKey = AddKeyPrefix(redisKey); return _db.SortedSetRemove(redisKey, member); } /// <summary> /// sorted set Add /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="member"></param> /// <param name="score"></param> /// <returns></returns> public bool SortedSetAdd<T>(string redisKey,T member,double score) { redisKey = AddKeyPrefix(redisKey); var json = Serialize(member); return _db.SortedSetAdd(redisKey, json, score); } /// <summary> /// sorted set add /// </summary> /// <param name="redisKey"></param> /// <param name="member"></param> /// <param name="score"></param> /// <returns></returns> public async Task<bool> SortedSetAddAsync(string redisKey,string member,double score) { redisKey = AddKeyPrefix(redisKey); return await _db.SortedSetAddAsync(redisKey, member, score); } /// <summary> /// 在有序集合中返回指定范围的元素,默认状况下由低到高 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<IEnumerable<RedisValue>> SortedSetRangeByRankAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.SortedSetRangeByRankAsync(redisKey); } /// <summary> /// 返回有序集合的元素个数 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<long> SortedSetLengthAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.SortedSetLengthAsync(redisKey); } /// <summary> /// 返回有序集合的元素个数 /// </summary> /// <param name="redisKey"></param> /// <param name="member"></param> /// <returns></returns> public async Task<bool> SortedSetRemoveAsync(string redisKey,string member) { redisKey = AddKeyPrefix(redisKey); return await _db.SortedSetRemoveAsync(redisKey, member); } /// <summary> /// SortedSet 新增 /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisKey"></param> /// <param name="member"></param> /// <param name="score"></param> /// <returns></returns> public async Task<bool> SortedSetAddAsync<T>(string redisKey,T member,double score) { redisKey = AddKeyPrefix(redisKey); var json = Serialize(member); return await _db.SortedSetAddAsync(redisKey, json, score); } #endregion #region key operation /// <summary> /// 移除指定key /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public bool KeyDelete(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.KeyDelete(redisKey); } /// <summary> /// 删除指定key /// </summary> /// <param name="redisKeys"></param> /// <returns></returns> public long KeyDelete(IEnumerable<string> redisKeys) { var keys = redisKeys.Select(x => (RedisKey)AddKeyPrefix(x)); return _db.KeyDelete(keys.ToArray()); } /// <summary> /// 检验key是否存在 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public bool KeyExists(string redisKey) { redisKey = AddKeyPrefix(redisKey); return _db.KeyExists(redisKey); } /// <summary> /// 重命名key /// </summary> /// <param name="oldKeyName"></param> /// <param name="newKeyName"></param> /// <returns></returns> public bool KeyReName(string oldKeyName,string newKeyName) { oldKeyName = AddKeyPrefix(oldKeyName); return _db.KeyRename(oldKeyName, newKeyName); } /// <summary> /// 设置key 的过时时间 /// </summary> /// <param name="redisKey"></param> /// <param name="expired"></param> /// <returns></returns> public bool KeyExpire(string redisKey,TimeSpan?expired = null) { redisKey = AddKeyPrefix(redisKey); return _db.KeyExpire(redisKey, expired); } /// <summary> /// 移除指定的key /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<bool> KeyDeleteAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.KeyDeleteAsync(redisKey); } /// <summary> /// 删除指定的key /// </summary> /// <param name="redisKeys"></param> /// <returns></returns> public async Task<long> KeyDeleteAsync(IEnumerable<string> redisKeys) { var keys = redisKeys.Select(x => (RedisKey)AddKeyPrefix(x)); return await _db.KeyDeleteAsync(keys.ToArray()); } /// <summary> /// 检验key 是否存在 /// </summary> /// <param name="redisKey"></param> /// <returns></returns> public async Task<bool> KeyExistsAsync(string redisKey) { redisKey = AddKeyPrefix(redisKey); return await _db.KeyExistsAsync(redisKey); } /// <summary> /// 重命名key /// </summary> /// <param name="redisKey"></param> /// <param name="redisNewKey"></param> /// <returns></returns> public async Task<bool> KeyRenameAsync(string redisKey,string redisNewKey) { redisKey = AddKeyPrefix(redisKey); return await _db.KeyRenameAsync(redisKey, redisNewKey); } /// <summary> /// 设置 key 时间 /// </summary> /// <param name="redisKey"></param> /// <param name="expired"></param> /// <returns></returns> public async Task<bool> KeyExpireAsync(string redisKey,TimeSpan? expired) { redisKey = AddKeyPrefix(redisKey); return await _db.KeyExpireAsync(redisKey, expired); } #endregion #region Subscribe /// <summary> /// 订阅 /// </summary> /// <param name="channel">频道</param> /// <param name="handle">事件</param> public void Subscribe(RedisChannel channel,Action<RedisChannel,RedisValue> handle) { //getSubscriber() 获取到指定服务器的发布者订阅者的链接 var sub = _connMultiplexer.GetSubscriber(); //订阅执行某些操做时改变了 优先/主动 节点广播 sub.Subscribe(channel, handle); } /// <summary> /// 发布 /// </summary> /// <param name="channel"></param> /// <param name="message"></param> /// <returns></returns> public long Publish(RedisChannel channel,RedisValue message) { var sub = _connMultiplexer.GetSubscriber(); return sub.Publish(channel, message); } /// <summary> /// 发布(使用序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="channel"></param> /// <param name="message"></param> /// <returns></returns> public long Publish<T>(RedisChannel channel,T message) { var sub = _connMultiplexer.GetSubscriber(); return sub.Publish(channel, Serialize(message)); } /// <summary> /// 订阅 /// </summary> /// <param name="redisChannel"></param> /// <param name="handle"></param> /// <returns></returns> public async Task SubscribeAsync(RedisChannel redisChannel,Action<RedisChannel,RedisValue> handle) { var sub = _connMultiplexer.GetSubscriber(); await sub.SubscribeAsync(redisChannel, handle); } /// <summary> /// 发布 /// </summary> /// <param name="redisChannel"></param> /// <param name="message"></param> /// <returns></returns> public async Task<long> PublishAsync(RedisChannel redisChannel,RedisValue message) { var sub = _connMultiplexer.GetSubscriber(); return await sub.PublishAsync(redisChannel, message); } /// <summary> /// 发布(使用序列化) /// </summary> /// <typeparam name="T"></typeparam> /// <param name="redisChannel"></param> /// <param name="message"></param> /// <returns></returns> public async Task<long> PublishAsync<T>(RedisChannel redisChannel,T message) { var sub = _connMultiplexer.GetSubscriber(); return await sub.PublishAsync(redisChannel, Serialize(message)); } #endregion #region register event /// <summary> /// 注册事件 /// </summary> private static void RegisterEvent() { _connMultiplexer.ConnectionRestored += ConnMultiplexer_ConnectionRestored; _connMultiplexer.ConnectionFailed += ConnMultiplexer_ConnectionFailed; _connMultiplexer.ErrorMessage += ConnMultiplexer_ErrorMessage; _connMultiplexer.ConfigurationChanged += ConnMultiplexer_ConfigurationChanged; _connMultiplexer.HashSlotMoved += ConnMultiplexer_HashSlotMoved; _connMultiplexer.InternalError += ConnMultiplexer_InternalError; _connMultiplexer.ConfigurationChangedBroadcast += ConnMultiplexer_ConfigurationChangedBroadcast; } /// <summary> /// 从新配置广播时(主从同步更改) /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_ConfigurationChangedBroadcast(object sender, EndPointEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_ConfigurationChangedBroadcast)}: {e.EndPoint}"); } /// <summary> /// 发生内部错误时(调试用) /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_InternalError(object sender,InternalErrorEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_InternalError)}: {e.Exception}"); } /// <summary> /// 更改集群时 /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_HashSlotMoved(object sender,HashSlotMovedEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_HashSlotMoved)}: {nameof(e.OldEndPoint)}-{e.OldEndPoint} To {nameof(e.NewEndPoint)}-{e.NewEndPoint} "); } /// <summary> /// 配置更改时 /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_ConfigurationChanged(object sender, EndPointEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_ConfigurationChanged)}: {e.EndPoint}"); } /// <summary> /// 发生错误时 /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_ErrorMessage(object sender, RedisErrorEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_ErrorMessage)}: {e.Message}"); } /// <summary> /// 物理链接失败时 /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_ConnectionFailed(object sender, ConnectionFailedEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_ConnectionFailed)}: {e.Exception}"); } /// <summary> /// 创建物理链接时 /// </summary> /// <param name="sender"></param> /// <param name="e"></param> private static void ConnMultiplexer_ConnectionRestored(object sender, ConnectionFailedEventArgs e) { Console.WriteLine($"{nameof(ConnMultiplexer_ConnectionRestored)}: {e.Exception}"); } #endregion #region physical save - undo #endregion } }
3.3 引入博文a 老师的消息出列方法,代码以下(此部分进行了部分修改)
using MVC5Project.Models; using MVC5Project.Redis.BaseOnStackExchage; namespace MVC5Project.MSMQ { public class MessageQueue { static System.Timers.Timer timer = new System.Timers.Timer(5000); public static ChatModels CurrentChatModels = new ChatModels(); static MessageQueue() { timer.AutoReset = true; timer.Enabled = true; timer.Elapsed += new System.Timers.ElapsedEventHandler(timer_Elapsed); timer.Start(); } private static void timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e) { var redisClient = new RedisHelper(2);
// 消息出列 CurrentChatModels = redisClient.ListLeftPop<ChatModels>("MessageQuene"); } } }
3.4 web 端页面处理,在Home的Index 页面中
@{ } <!DOCTYPE html> <html> <head> <meta http-equiv="Content-Type" content="text/html; charset=utf-8" /> <meta charset="utf-8" /> <title> Testing MSMQ </title> </head> <body> <script type="text/javascript"> function onSubmit() { alert('111'); $.post("/Home/Index", { userId: $('#userId').val(),chat:$('#chat').val() }, function () { }); } </script> <div> <form action="~/Controllers/Home" method="post"> <input id="userId" value="chenk"/> <input id="chat" value="this is my first messageQuene info !" /> <input type="button" value="submit" onclick="onSubmit()"/> </form> <label>PageInfo:</label>@ViewBag.PageInfo <label>MessageQuene:</label>@ViewBag.MessageQuene <label>Pop:</label>@ViewBag.Pop </div> <script src="~/Scripts/jquery-1.10.2.min.js"></script> </body> </html>
3.5 controller 处理消息入列内容
/// <summary> /// display page /// </summary> /// <returns></returns> public ActionResult Index(){ var redisClient = new RedisHelper(2); ViewBag.PageInfo = "this page is Home"; List<ChatModels> isError = null; ViewData["pop"] = MessageQueue.CurrentChatModels == null ? "没有记录" : MessageQueue.CurrentChatModels.chat; //目前ListRange()方法会出现RedisTimeOutException,并未找到问题根源,可是不影响代码执行。 ViewData["MSMQ"] = redisClient.ListRange("MessageQuene") == null ? isError = new List<ChatModels>() : redisClient.ListRange("MessageQuene").Cast<ChatModels>().ToList(); return View(); }
/// <summary> /// submit form action /// </summary> /// <returns></returns> [HttpPost] public ActionResult Index(FormCollection form){ var redisClient = new RedisHelper(2); List<ChatModels> isError = null; //消息入列 redisClient.ListRightPush("MessageQuene", new ChatModels { userId = form["userId"], chat = form["chat"] }); ViewData["MessageQuene"] = redisClient.ListRange("MessageQuene") == null ? isError = new List<ChatModels>() : redisClient.ListRange("MessageQuene").Cast<ChatModels>().ToList(); return View(); }
四、执行结果查看
a、在点击submit 以前 db(2) 为空。
b、点击submit
c、完成消息入列
d、5秒钟后,数据会从db(2) 中移出。完成消息出列。
五、总结
以上,为一个完整的Redis 模拟存储消息的实现,在过程当中也有几个未能完善的问题
(1)、StackExchage.Redis 提供的 ListRange() 方法会抛出RedisTimeOutException ,该问题并未在网上找到致使的缘由。
(2)、经过ListRightPush<T>方法 插入的数据,在客户端中并未能正确显示,此问题还需进一步解决。