.NetCore使用Redis，StackExchange.Redis队列，发布与订阅，分布式锁的简单使用-程序员宅基地

技术标签：网络多线程数据库分布式 redis

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<ApiResultObject> 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<ApiResultObject> 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<ApiResultObject> 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<ApiResultObject> 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 入队
        /// <summary>
        /// 入队right
        /// </summary>
        /// <param name="queueName"></param>
        /// <param name="redisValue"></param>
        /// <returns></returns>
        public static long EnqueueListRightPush(RedisKey queueName, RedisValue redisValue)
        {
            using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr))
            {
                return conn.GetDatabase().ListRightPush(queueName, redisValue);
            }
        }
        /// <summary>
        /// 入队left
        /// </summary>
        /// <param name="queueName"></param>
        /// <param name="redisvalue"></param>
        /// <returns></returns>
        public static long EnqueueListLeftPush(RedisKey queueName, RedisValue redisvalue)
        {
            using (var conn = ConnectionMultiplexer.Connect(connectionRedisStr))
            {
                return conn.GetDatabase().ListLeftPush(queueName, redisvalue);
            }
        }
        /// <summary>
        /// 入队left异步
        /// </summary>
        /// <param name="queueName"></param>
        /// <param name="redisvalue"></param>
        /// <returns></returns>
        public static async Task<long> EnqueueListLeftPushAsync(RedisKey queueName, RedisValue redisvalue)
        {
            using (var conn = await ConnectionMultiplexer.ConnectAsync(connectionRedisStr))
            {
                return await conn.GetDatabase().ListLeftPushAsync(queueName, redisvalue);
            }
        }
        /// <summary>
        /// 获取队列的长度
        /// </summary>
        /// <param name="queueName"></param>
        /// <returns></returns>
        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<RedisChannel, RedisValue> 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
    }
}

本文链接：https://blog.csdn.net/kaixincheng2009/article/details/108231895

原作者删帖不实内容删帖广告或垃圾文章投诉

智能推荐

oracle 12c 集群安装后的检查_12c查看crs状态-程序员宅基地

文章浏览阅读1.6k次。安装配置gi、安装数据库软件、dbca建库见下：http://blog.csdn.net/kadwf123/article/details/784299611、检查集群节点及状态：[root@rac2 ~]# olsnodes -srac1 Activerac2 Activerac3 Activerac4 Active[root@rac2 ~]_12c查看crs状态

解决jupyter notebook无法找到虚拟环境的问题_jupyter没有pytorch环境-程序员宅基地

文章浏览阅读1.3w次，点赞45次，收藏99次。我个人用的是anaconda3的一个python集成环境，自带jupyter notebook，但在我打开jupyter notebook界面后，却找不到对应的虚拟环境，原来是jupyter notebook只是通用于下载anaconda时自带的环境，其他环境要想使用必须手动下载一些库：1.首先进入到自己创建的虚拟环境(pytorch是虚拟环境的名字)activate pytorch2.在该环境下下载这个库conda install ipykernelconda install nb__jupyter没有pytorch环境

国内安装scoop的保姆教程_scoop-cn-程序员宅基地

文章浏览阅读5.2k次，点赞19次，收藏28次。选择scoop纯属意外，也是无奈，因为电脑用户被锁了管理员权限，所有exe安装程序都无法安装，只可以用绿色软件，最后被我发现scoop，省去了到处下载XXX绿色版的烦恼，当然scoop里需要管理员权限的软件也跟我无缘了（譬如everything）。推荐添加dorado这个bucket镜像，里面很多中文软件，但是部分国外的软件下载地址在github，可能无法下载。以上两个是官方bucket的国内镜像，所有软件建议优先从这里下载。上面可以看到很多bucket以及软件数。如果官网登陆不了可以试一下以下方式。_scoop-cn

Element ui colorpicker在Vue中的使用_vue el-color-picker-程序员宅基地

文章浏览阅读4.5k次，点赞2次，收藏3次。首先要有一个color-picker组件 <el-color-picker v-model="headcolor"></el-color-picker>在data里面data() { return {headcolor: ’ #278add ’ //这里可以选择一个默认的颜色} }然后在你想要改变颜色的地方用v-bind绑定就好了，例如：这里的:sty..._vue el-color-picker

迅为iTOP-4412精英版之烧写内核移植后的镜像_exynos 4412 刷机-程序员宅基地

文章浏览阅读640次。基于芯片日益增长的问题，所以内核开发者们引入了新的方法，就是在内核中只保留函数，而数据则不包含，由用户（应用程序员）自己把数据按照规定的格式编写，并放在约定的地方，为了不占用过多的内存，还要求数据以根精简的方式编写。boot启动时，传参给内核，告诉内核设备树文件和kernel的位置，内核启动时根据地址去找到设备树文件，再利用专用的编译器去反编译dtb文件，将dtb还原成数据结构，以供驱动的函数去调用。firmware是三星的一个固件的设备信息，因为找不到固件，所以内核启动不成功。_exynos 4412 刷机

Linux系统配置jdk_linux配置jdk-程序员宅基地

文章浏览阅读2w次，点赞24次，收藏42次。Linux系统配置jdkLinux学习教程，Linux入门教程（超详细）_linux配置jdk

随便推点

matlab(4)：特殊符号的输入_matlab微米怎么输入-程序员宅基地

文章浏览阅读3.3k次，点赞5次，收藏19次。xlabel('\delta');ylabel('AUC');具体符号的对照表参照下图：_matlab微米怎么输入

C语言程序设计-文件(打开与关闭、顺序、二进制读写)-程序员宅基地

文章浏览阅读119次。顺序读写指的是按照文件中数据的顺序进行读取或写入。对于文本文件，可以使用fgets、fputs、fscanf、fprintf等函数进行顺序读写。在C语言中，对文件的操作通常涉及文件的打开、读写以及关闭。文件的打开使用fopen函数，而关闭则使用fclose函数。在C语言中，可以使用fread和fwrite函数进行二进制读写。‍ Biaoge 于2024-03-09 23:51发布阅读量：7 ️文章类型：【 C语言程序设计】在C语言中，用于打开文件的函数是____，用于关闭文件的函数是____。

Touchdesigner自学笔记之三_touchdesigner怎么让一个模型跟着鼠标移动-程序员宅基地

文章浏览阅读3.4k次，点赞2次，收藏13次。跟随鼠标移动的粒子以grid（SOP）为partical（SOP）的资源模板，调整后连接【Geo组合+point spirit（MAT)】，在连接【feedback组合】适当调整。影响粒子动态的节点【metaball(SOP)+force(SOP)】添加mouse in（CHOP)鼠标位置到metaball的坐标，实现鼠标影响。..._touchdesigner怎么让一个模型跟着鼠标移动

【附源码】基于java的校园停车场管理系统的设计与实现61m0e9计算机毕设SSM_基于java技术的停车场管理系统实现与设计-程序员宅基地

文章浏览阅读178次。项目运行环境配置：Jdk1.8 + Tomcat7.0 + Mysql + HBuilderX（Webstorm也行）+ Eclispe（IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持）。项目技术：Springboot + mybatis + Maven +mysql5.7或8.0+html+css+js等等组成，B/S模式 + Maven管理等等。环境需要1.运行环境：最好是java jdk 1.8，我们在这个平台上运行的。其他版本理论上也可以。_基于java技术的停车场管理系统实现与设计

Android系统播放器MediaPlayer源码分析_android多媒体播放源码分析时序图-程序员宅基地

文章浏览阅读3.5k次。前言对于MediaPlayer播放器的源码分析内容相对来说比较多，会从Java-&amp;gt;Jni-&amp;gt;C/C++慢慢分析，后面会慢慢更新。另外，博客只作为自己学习记录的一种方式，对于其他的不过多的评论。MediaPlayerDemopublic class MainActivity extends AppCompatActivity implements SurfaceHolder.Cal..._android多媒体播放源码分析时序图

java 数据结构与算法 ——快速排序法-程序员宅基地

文章浏览阅读2.4k次，点赞41次，收藏13次。java 数据结构与算法 ——快速排序法_快速排序法