.NetCore使用Redis,StackExchange.Redis队列,发布与订阅,分布式锁的简单使用
共 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;
[ ]
[ ]
public class RedisTestController
{
[ ]
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;
}
[ ]
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;
[ ]
[ ]
public class RedisTestController
{
[ ]
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;
}
[ ]
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)
{
{
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();
}
{
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();
}
}
}
}
完整的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";
}
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);
}
}
///
/// 入队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);
}
}
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;
}
}
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();
}
}
}
}
}
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);
}
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}");
}
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;
}
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}");
}
}
}
}