c#进阶（4）—— Redis 用于消息队列的存储

1、参考的博文

a : http://www.cnblogs.com/lori/archive/2012/04/12/2443708.html —— 主要的实现思路

b: http://www.cnblogs.com/liqingwen/archive/2017/04/06/6672452.html —— RedisHelper 类

c : https://www.cnblogs.com/stopfalling/p/5375492.html —— 应用场景说明

2、原理说明

博文a 中的老师，提供了Redis 实现消息队列的整体思路，言简意赅，但部分类库a 老师并未提供，因此我参照了博文b 中老师的RedisHelper 类，主要借鉴的方法为ListLeftPop及ListRightPush,及实现消息队列的核心思想，先进先出。

博文c 中老师详细介绍了几种消息队列的业务场景，是我所看的所有业务场景描述中最为详细，清晰的，结合博文a 老师的整体思路，对消息队列的实现上有了较为清晰的认识。

整体思路在博文a 中老师已经介绍，及依靠 mvc 框架，通过web端用户提交事件，实现消息入列，通过定时器（Timer）实现按照时间间隔的消息出列，Redis 作为存储媒介，存储消

息内容。

3、具体实现

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" οnclick="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();}

4、执行结果查看

a、在点击submit 之前 db(2) 为空。

b、点击submit

c、完成消息入列

d、5秒钟后，数据会从db(2) 中移出。完成消息出列。

5、总结

以上，为一个完整的Redis 模拟存储消息的实现，在过程中也有几个未能完善的问题

（1）、StackExchage.Redis 提供的 ListRange() 方法会抛出RedisTimeOutException ，该问题并未在网上找到导致的原因。

（2）、通过ListRightPush<T>方法插入的数据，在客户端中并未能正确显示，此问题还需进一步解决

c#进阶（4）—— Redis 用于消息队列的存储相关推荐

activemq消息丢失_基于Redis实现消息队列的典型方案
基于Redis实现消息队列典型方案 1 概述 2 基于List的 LPUSH+BRPOP 的实现 3 PUB/SUB,订阅/发布模式 4 基于SortedSet有序集合的实现 5 基于 Stream ...
redis实现消息队列的几种方式及其优劣
概述常用的消息队列有,rabbitMq.kafka.RocketMq.ActiveMq等.这些消息队列需要独立安装部署,作为一个中间件来提供服务,虽然有着高性能.高可靠的优点,但是额外部署这些中间件 ...
秒杀抢购异步下单：基于Redis的消息队列秒杀抢购异步下单功能
学习Redis时,练习的实战项目代码--基于Redis的Stream类型的秒杀抢购异步下单. 说明: Redis的stream类型的消息队列实现异步下单功能.Redis版本至少要5.0及以上版本才可以 ...
Redis做消息队列，香吗？
来自:架构师修行之路菜菜哥,我刚做完了一个订单系统,感觉很简单呀说说看,大量的订单状态怎么处理的? 我设计的时候可是考虑了这一点,所以用了异步处理,采用了MQ 那用的什么MQ呢,透露一下呗我用的 ...
【springboot】【redis】springboot+redis实现发布订阅功能，实现redis的消息队列的功能...
springboot+redis实现发布订阅功能,实现redis的消息队列的功能参考:https://www.cnblogs.com/cx987514451/p/9529611.html 思考一个问 ...
PHP + Redis 实现消息队列
Redis做消息队列的好处在于它的轻量级,高并发,延迟敏感,应用场景有即时数据分析.秒杀计数器.缓存等 Redis做消息队列待解决的问题: 1.消息的可靠性: 没有相应的机制保证消息的消费,当消费者 ...
redis实现消息队列
redis是一个开源的key-value存储系统.与Memcached类似,Redis将大部分数据存储在内存中,支持的数据类型包括:字符串.哈希表.链表.集合.有序集合以及基于这些数据类型的相关操作. ...
用redis实现消息队列（实时消费+ack机制）【转】
用redis实现消息队列(实时消费+ack机制) java queue 消息队列 redis 消息队列首先做简单的引入. MQ主要是用来: 解耦应用. 异步化消息流量削峰填谷目前使用的较多的有A ...
使用Redis 实现消息队列
一 .为什么要用Redis实现轻量级MQ? MQ的主要作用: 应用解耦异步化消息流量削峰填谷目前使用比较多的是ActiveMQ . RabbitMQ . ZeroMQ . Kafka . Met ...

c#进阶（4）—— Redis 用于消息队列的存储

c#进阶（4）—— Redis 用于消息队列的存储相关推荐

最新文章

热门文章