.NetCore使用Redis,StackExchange.Redis队列,发布与订阅,分布式锁的简单使用

dotNET知音

共 14700字,需浏览 30分钟

 ·

2020-08-27 22:51

https://www.cnblogs.com/Fengge518/p/13556182.html

环境:之前一直是使用serverStack.Redis的客服端,

今天来使用一下StackExchange.Redis(个人感觉更加的人性化一些,也是免费的,性能也不会差太多),
版本为StackExchange.Redis V2.1.58 ,Core3.1

简单的说明(专业的术语参考资料网络和官网):官网地址:https://www.redis.net.cn/

Redis(Remote Dictionary Server ),即远程字典服务是一个开源的 ,由C语言编写、支持网络、可基于内存亦可持久化的日志型、Key-Value数据库,并提供多种语言的API。

Redis 是一个高性能的key-value数据库。Redis的出现,很大程度补偿了memcached这类key/value存储的不足,

提供了Java,C/C++,C#,PHP,JavaScript,Perl,Object-C,Python,Ruby,Erlang等客户端,使用很方便。

优点:

1 Redis读写性能优异,从内存当中进行IO读写速度快,支持超过100K+每秒的读写频率。

2 Redis支持Strings,

Lists, Hashes, Sets,Ordered Sets等数据类型操作。

3 Redis支持数据持久化,支持AOF和RDB两种持久化方式

4 Redis支持主从复制,主机会自动将数据同步到从机,可以进行读写分离。

5 Redis的所有操作都是原子性的,同时Redis还支持对几个操作全并后的原子性执行。

6 Redis是单线程多CPU,不需要考虑加锁释放锁,也就没有死锁的问题,效率高。

1:redis队列值入队出队,截图效果:

优化之前入队1000条数据,测试结果将近50秒,这实在太慢,不可忍受!

 优化后的效果:为5.55s的样子

 

2:redis发布与订阅截图效果:(一个发布者,四个订阅者) 订阅者都会收到相同的信息

 3:redis秒杀,截图如下:单个进程秒杀ok

开多个进程时,会有超卖的现象:(那我加lock锁呢?结果也是会有超卖的现象,此时下面的分布式锁可以解决)

 4:加上redis分布式锁的测试效果截图:

但是这样会比较耗资源,库存已经没有了,就应该不要再去执行下去了

分布式锁ok库存为零就不在请求直接抛异常即可

 上面通过测试的截图,简单的介绍了,Redis的队列(入队和出队),Redis发布与订阅,Redis分布式锁的使用,现在直接上代码 :

出队入队的WebApi Core3.1

 

using System;using System.Collections.Generic;using System.Linq;using System.Threading.Tasks;
namespace WebApp.Controllers{ using Microsoft.AspNetCore.Mvc; using RedisPublishAndSubHelper; [Route("api/[Controller]")] [ApiController] public class RedisTestController { [HttpGet("EnqueueMsg")] public async Task EnqueueMsgAsync(string rediskey, string redisValue) { ApiResultObject obj = new ApiResultObject(); try { long enqueueLong = default; for (int i = 0; i < 1000; i++) { enqueueLong = await MyRedisSubPublishHelper.EnqueueListLeftPushAsync(rediskey, redisValue + i); } obj.Code = ResultCode.Success; obj.Data = "入队的数据长度:" + enqueueLong; obj.Msg = "入队成功!"; } catch (Exception ex) {
obj.Msg = $"入队异常,原因:{ex.Message}"; } return obj; } [HttpGet("DequeueMsg")] public async Task DequeueMsgAsync(string rediskey) { ApiResultObject obj = new ApiResultObject(); try { string dequeueMsg = await MyRedisSubPublishHelper.DequeueListPopRightAsync(rediskey); obj.Code = ResultCode.Success; obj.Data = $"出队的数据是:{dequeueMsg}"; obj.Msg = "入队成功!"; } catch (Exception ex) { obj.Msg = $"入队异常,原因:{ex.Message}"; } return obj; } }}

出队入队的后端code WebApi:

 

using System;using System.Collections.Generic;using System.Linq;using System.Threading.Tasks;
namespace WebApp.Controllers{ using Microsoft.AspNetCore.Mvc; using RedisPublishAndSubHelper; [Route("api/[Controller]")] [ApiController] public class RedisTestController { [HttpGet("EnqueueMsg")] public async Task EnqueueMsgAsync(string rediskey, string redisValue) { ApiResultObject obj = new ApiResultObject(); try { long enqueueLong = default; for (int i = 0; i < 1000; i++) { enqueueLong = await MyRedisSubPublishHelper.EnqueueListLeftPushAsync(rediskey, redisValue + i); } obj.Code = ResultCode.Success; obj.Data = "入队的数据长度:" + enqueueLong; obj.Msg = "入队成功!"; } catch (Exception ex) {
obj.Msg = $"入队异常,原因:{ex.Message}"; } return obj; } [HttpGet("DequeueMsg")] public async Task DequeueMsgAsync(string rediskey) { ApiResultObject obj = new ApiResultObject(); try { string dequeueMsg = await MyRedisSubPublishHelper.DequeueListPopRightAsync(rediskey); obj.Code = ResultCode.Success; obj.Data = $"出队的数据是:{dequeueMsg}"; obj.Msg = "入队成功!"; } catch (Exception ex) { obj.Msg = $"入队异常,原因:{ex.Message}"; } return obj; } }}

入队以及秒杀分布式锁的客服端的Code:

 

using System;using System.Threading;using System.Threading.Tasks;
namespace RedisPublishService{ using RedisPublishAndSubHelper; class Program { static void Main(string[] args) { #region 入队的code { int Index = 100000; while (Index > 0) { //string msg = Console.ReadLine(); new MyRedisSubPublishHelper().PublishMessage("nihaofengge", $"你好风哥:Guid值是:{DateTime.Now}{Guid.NewGuid().ToString()}"); Console.WriteLine("发布成功!"); Index -= 1; } Console.ReadKey(); } #endregion
#region 秒杀的code { try { Console.WriteLine("秒杀开始。。。。。"); for (int i = 0; i < 200; i++) { Task.Run(() => { MyRedisSubPublishHelper.LockByRedis("mstest"); string productCount = MyRedisHelper.StringGet("productcount"); int pcount = int.Parse(productCount); if (pcount > 0) { long dlong = MyRedisHelper.StringDec("productcount"); Console.WriteLine($"秒杀成功,商品库存:{dlong}"); pcount -= 1; System.Threading.Thread.Sleep(30); } else { Console.WriteLine($"秒杀失败,商品库存为零了!"); throw new Exception("产品秒杀数量为零!");//加载这里会比较保险 } MyRedisSubPublishHelper.UnLockByRedis("mstest"); }).Wait(); } } catch (Exception ex) { Console.WriteLine($"产品已经秒杀完毕,原因:{ex.Message}"); } Console.ReadKey(); } #endregion } }}

完整的code RedisHelper帮助类(测试并使用了部分方法的封装),

 

using System;using System.Collections.Generic;using System.Linq;using System.Threading.Tasks;
namespace RedisPublishAndSubHelper{ using StackExchange.Redis; using StackExchange; using System.Threading;
public class MyRedisHelper { private static readonly string connectionRedisStr = string.Empty; static MyRedisHelper() { //在这里来初始化一些配置信息 connectionRedisStr = "12.23.45.12:6379,connectTimeout=1000,connectRetry=3,syncTimeout=10000"; }
#region Redis string简单的常见同步方法操作 public static bool StringSet(string key, string stringValue, double senconds = 60) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase db = conn.GetDatabase(); return db.StringSet(key, stringValue, TimeSpan.FromSeconds(senconds)); } } public static string StringGet(string key) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase db = conn.GetDatabase(); return db.StringGet(key); } }
public static long StringInc(string key) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase db = conn.GetDatabase(); return db.StringIncrement(key); } }
public static long StringDec(string key) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase db = conn.GetDatabase(); return db.StringDecrement(key); } } public static bool KeyExists(string key) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase db = conn.GetDatabase(); return db.KeyExists(key); } } #endregion
#region List Hash, Set,Zset 大同小异的使用,比较简单,后续有时间再补上
#endregion
#region 入队出队
#region 入队 /// /// 入队right /// /// /// /// public static long EnqueueListRightPush(RedisKey queueName, RedisValue redisValue) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { return conn.GetDatabase().ListRightPush(queueName, redisValue); } } /// /// 入队left /// /// /// /// public static long EnqueueListLeftPush(RedisKey queueName, RedisValue redisvalue) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { return conn.GetDatabase().ListLeftPush(queueName, redisvalue); } } /// /// 入队left异步 /// /// /// /// public static async Task<long> EnqueueListLeftPushAsync(RedisKey queueName, RedisValue redisvalue) { using (var conn = await ConnectionMultiplexer.ConnectAsync(connectionRedisStr)) { return await conn.GetDatabase().ListLeftPushAsync(queueName, redisvalue); } } /// /// 获取队列的长度 /// /// /// public static long EnqueueListLength(RedisKey queueName) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { return conn.GetDatabase().ListLength(queueName); } }
#endregion
#region 出队 public static string DequeueListPopLeft(RedisKey queueName) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase database = conn.GetDatabase(); int count = database.ListRange(queueName).Length; if (count <= 0) { throw new Exception($"队列{queueName}数据为零"); } string redisValue = database.ListLeftPop(queueName); if (!string.IsNullOrEmpty(redisValue)) return redisValue; else return string.Empty; } } public static string DequeueListPopRight(RedisKey queueName) { using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr)) { IDatabase database = conn.GetDatabase(); int count = database.ListRange(queueName).Length; if (count <= 0) { throw new Exception($"队列{queueName}数据为零"); } string redisValue = conn.GetDatabase().ListRightPop(queueName); if (!string.IsNullOrEmpty(redisValue)) return redisValue; else return string.Empty; } } public static async Task<string> DequeueListPopRightAsync(RedisKey queueName) { using (var conn = await ConnectionMultiplexer.ConnectAsync(connectionRedisStr)) { IDatabase database = conn.GetDatabase(); int count = (await database.ListRangeAsync(queueName)).Length; if (count <= 0) { throw new Exception($"队列{queueName}数据为零"); } string redisValue = await conn.GetDatabase().ListRightPopAsync(queueName); if (!string.IsNullOrEmpty(redisValue)) return redisValue; else return string.Empty; } } #endregion
#endregion
#region 分布式锁 public static void LockByRedis(string key, int expireTimeSeconds = 10) { try { IDatabase database1 = ConnectionMultiplexer.Connect(connectionRedisStr).GetDatabase(); while (true) { expireTimeSeconds = expireTimeSeconds > 20 ? 10 : expireTimeSeconds; bool lockflag = database1.LockTake(key, Thread.CurrentThread.ManagedThreadId, TimeSpan.FromSeconds(expireTimeSeconds)); if (lockflag) { break; } } } catch (Exception ex) { throw new Exception($"Redis加锁异常:原因{ex.Message}"); } }
public static bool UnLockByRedis(string key) { ConnectionMultiplexer conn = ConnectionMultiplexer.Connect(connectionRedisStr); try { IDatabase database1 = conn.GetDatabase(); return database1.LockRelease(key, Thread.CurrentThread.ManagedThreadId); } catch (Exception ex) { throw new Exception($"Redis加锁异常:原因{ex.Message}"); } finally { if (conn != null) { conn.Close(); conn.Dispose(); } } } #endregion
}}

 

using System;using System.Collections.Generic;using System.Text;
namespace RedisPublishAndSubHelper{ using StackExchange.Redis; using System.Net.Http; using System.Threading; using System.Threading.Channels; using System.Threading.Tasks;
public class MyRedisSubPublishHelper { private static readonly string redisConnectionStr = "12.32.12.54:6379,connectTimeout=10000,connectRetry=3,syncTimeout=10000"; private static readonly ConnectionMultiplexer connectionMultiplexer = null; static MyRedisSubPublishHelper() { connectionMultiplexer = ConnectionMultiplexer.Connect(redisConnectionStr); }

#region 发布订阅 public void SubScriper(string topticName, Action handler = null) { ISubscriber subscriber = connectionMultiplexer.GetSubscriber(); ChannelMessageQueue channelMessageQueue = subscriber.Subscribe(topticName); channelMessageQueue.OnMessage(channelMessage => { if (handler != null) { string redisChannel = channelMessage.Channel; string msg = channelMessage.Message; handler.Invoke(redisChannel, msg); } else { string msg = channelMessage.Message; Console.WriteLine($"订阅到消息: { msg},Channel={channelMessage.Channel}"); } }); } public void PublishMessage(string topticName, string message) { ISubscriber subscriber = connectionMultiplexer.GetSubscriber(); long publishLong = subscriber.Publish(topticName, message); Console.WriteLine($"发布消息成功:{publishLong}"); } #endregion
#region 入队出队 public static async Task<long> EnqueueListLeftPushAsync(RedisKey queueName, RedisValue redisvalue) { return await connectionMultiplexer.GetDatabase().ListLeftPushAsync(queueName, redisvalue); }
public static async Task<string> DequeueListPopRightAsync(RedisKey queueName) { IDatabase database = connectionMultiplexer.GetDatabase(); int count = (await database.ListRangeAsync(queueName)).Length; if (count <= 0) { throw new Exception($"队列{queueName}数据为零"); } string redisValue = await database.ListRightPopAsync(queueName); if (!string.IsNullOrEmpty(redisValue)) return redisValue; else return string.Empty; } #endregion
#region 分布式锁 public static void LockByRedis(string key, int expireTimeSeconds = 10) { try { IDatabase database = connectionMultiplexer.GetDatabase(); while (true) { expireTimeSeconds = expireTimeSeconds > 20 ? 10 : expireTimeSeconds; bool lockflag = database.LockTake(key, Thread.CurrentThread.ManagedThreadId, TimeSpan.FromSeconds(expireTimeSeconds)); if (lockflag) { break; } } } catch (Exception ex) { throw new Exception($"Redis加锁异常:原因{ex.Message}"); } }
public static bool UnLockByRedis(string key) { try { IDatabase database = connectionMultiplexer.GetDatabase(); return database.LockRelease(key, Thread.CurrentThread.ManagedThreadId); } catch (Exception ex) { throw new Exception($"Redis加锁异常:原因{ex.Message}"); } } #endregion }}


浏览 41
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报