高并发时，使用Redis应注意的问题【缓存穿透、缓存击穿.、缓存雪崩】-白红宇的个人博客

高并发时，使用Redis应注意的问题【缓存穿透、缓存击穿.、缓存雪崩】

发布日期：2021-05-09 01:18:05 浏览次数：19 分类：技术文章

本文共 13920 字，大约阅读时间需要 46 分钟。

十年河东，十年河西，莫欺少年穷

学无止境，精益求精

首先说下，我的 Redis 系列博客如下：

进入正文

缓存的出现解决了数据库压力的问题，但是当以下情况发生的时候，缓存就不在起到作用了，缓存穿透、缓存击穿、缓存雪崩这三种情况。

1. 缓存穿透：

我们的程序中用缓存的时候一般采取的是先去缓存中查询我们想要的缓存数据，如果缓存中不存在我们想要的数据的话，缓存就失去了作用（譬如缓存失效），这时我们就是需要伸手向DB库要数据，如果这种动作过多数据库就崩溃了。

这种情况需要我们去预防了，比如说：我们向缓存获取一个用户信息，但是故意去输入一个缓存中不存在的用户Key，这样就避过了缓存，把压力重新转移到数据上面了。

对于这种问题我们可以采取：

因为缓存查不到用户信息，数据库也查询不到用户信息，我们就把访问的数据进行缓存，这时候就可以避免重复访问，顺利把压力重新转向缓存中，有人会有疑问了，当访问的参数有上万个都是不重复的参数，并且都是可以躲避缓存的怎么办，我们同样把数据存起来设置一个较短过期时间清理缓存。

示例代码如下：

[HttpGet]        [Route("RedisGet")]        public IActionResult RedisGet(string key)        {            if (rd.KeyExists(key))            {                /*                 * 如果缓存中存在，则直接返回结果                 */                var result = rd.StringGet(key);                return Ok(result);            }            else            {                /*                 * 如果缓存中不存在，则需要结合数据库进行查询，但必须采用相应的策略，防止恶意【缓存击穿】。                 * 数据库查询部分，                 * 如果数据库查询到结果，则对结果进行缓存，并返回结果。                 * 如果数据库查询不到结果，则对请求的数据进行缓存，防止缓存击穿。                 * 除了上述比较被动的防御以外，我们还可以采取一段时间内限制请求次数来达到恶意攻击行为。                 */                return Ok();            }        }

2. 缓存击穿：

事情是这样的，对于一些设置了过期时间的缓存KEY，过期的时候，程序被高并发访问了（此时缓存已失效），这个时候由于缓存失效，访问压力也就转移到了数据库身上，高并发情况下，数据库往往扛不住那么多请求。

针对这种情况，我们可以使用互斥锁（Mutex）来解决问题，

互斥锁原理：通俗的描述就是，一万个用户访问了，但是只有一个用户可以拿到访问数据库的权限。

当这个用户拿到这个权限之后重新创建缓存，这个时候剩下的访问者因为没有拿到权限，就原地等待着去访问缓存。

逻辑上‘永不过期’：有人就会想了，我设置了过期时间，但我的系统中有一个定时的服务一直在跑，这个服务是用于判断缓存是否即将过期，如果发现即将过期的缓存，通过定时服务来更新缓存，这个时候缓存中的数据在逻辑上就会‘永不过期’了。

比如，定时服务每10分钟跑一次，但当我们发现缓存的过期时间小于10分钟了，我们通过服务来更新缓存，达到‘永不过期’的目的。

互斥锁解决方案：

using System;using System.Threading;namespace ConsoleApp1{    class shareRes    {        public static int count = 0;        public static Mutex mutex = new Mutex();    }    class IncThread    {        int number;        public Thread thrd;        public IncThread(string name, int n)        {            thrd = new Thread(this.run);            number = n;            thrd.Name = name;            thrd.Start();        }        void run()        {            Console.WriteLine(thrd.Name + "正在等待 the mutex");            //申请            shareRes.mutex.WaitOne();            Console.WriteLine(thrd.Name + "申请到 the mutex");            do            {                Thread.Sleep(1000);                shareRes.count++;                Console.WriteLine("In " + thrd.Name + "ShareRes.count is " + shareRes.count);                number--;            } while (number > 0);            Console.WriteLine(thrd.Name + "释放 the nmutex");            //  释放            shareRes.mutex.ReleaseMutex();        }    }    class DecThread    {        int number;        public Thread thrd;        public DecThread(string name, int n)        {            thrd = new Thread(this.run);            number = n;            thrd.Name = name;            thrd.Start();        }        void run()        {            Console.WriteLine(thrd.Name + "正在等待 the mutex");            //申请            shareRes.mutex.WaitOne();            Console.WriteLine(thrd.Name + "申请到 the mutex");            do            {                Thread.Sleep(1000);                shareRes.count--;                Console.WriteLine("In " + thrd.Name + "ShareRes.count is " + shareRes.count);                number--;            } while (number > 0);            Console.WriteLine(thrd.Name + "释放 the nmutex");            //  释放            shareRes.mutex.ReleaseMutex();        }    }    class Program    {        static void Main(string[] args)        {            IncThread mthrd1 = new IncThread("线程1 thread ", 5);            DecThread mthrd2 = new DecThread("线程2 thread ", 5);            mthrd1.thrd.Join();            mthrd2.thrd.Join();            Console.Read();        }    }}

关于互斥锁解决方案，我们可以通过了解互斥锁（Mutex）来进行解决。

逻辑上‘永不过期’解决方案：

需要定义一个定时的服务，具体请参考，总之通过定时检测缓存过期时间来更新即将过期的缓存。

3. 缓存雪崩：

是指多种缓存设置了同一时间过期，这个时候大批量的数据访问来了，（缓存失效）数据库DB的压力又上来了。

解决方法在设置过期时间的时候，在过期时间的基础上增加一个随机数，尽可能的保证缓存不会大面积的同时失效，说白了，就是缓存的过期时间不能大批量相同。

以上便是使用Redis缓存应注意的三个方面及解决方案。

最后，

顺便贴出一个NetCore的缓存帮助类，封装的比较简单，但也是有用的，如下：

using Newtonsoft.Json;using StackExchange.Redis;using System;using System.Collections.Generic;using System.Linq;using System.Text;namespace FranchiseeCommon{    ///     /// Redis操作类    /// 老版用的是ServiceStack.Redis    /// .Net Core使用StackExchange.Redis的nuget包    /// 
    public class RedisHelper    {        //redis数据库连接字符串        private string _conn = "127.0.0.1:6379";        private int _db = 0;        //静态变量 保证各模块使用的是不同实例的相同链接        private static ConnectionMultiplexer connection;        ///         /// 构造函数        /// 
        public RedisHelper() { }        ///         /// 构造函数        /// 
        ///         ///         public RedisHelper(int db, string connectStr)        {            _db = db;            _conn = connectStr;        }        ///         /// 缓存数据库，数据库连接        /// 
        public ConnectionMultiplexer CacheConnection        {            get            {                try                {                    if (connection == null || !connection.IsConnected)                    {                        connection = new Lazy
     
      (() => ConnectionMultiplexer.Connect(_conn)).Value;                    }                }                catch (Exception ex)                {                    return null;                }                return connection;            }        }        /// 
              /// 缓存数据库        /// 
        public IDatabase CacheRedis => CacheConnection.GetDatabase(_db);        #region --KEY/VALUE存取--        /// 
              /// 单条存值        /// 
        /// 
      key        /// 
      The value.        /// 
      
       
        true
        if XXXX, 
       
        false
        otherwise.
              public bool StringSet(string key, string value)        {            return CacheRedis.StringSet(key, value);        }        /// 
              /// 保存单个key value        /// 
        /// 
      Redis Key        /// 
      保存的值        /// 
      过期时间        /// 
              public bool StringSet(string key, string value, TimeSpan? expiry = default(TimeSpan?))        {            return CacheRedis.StringSet(key, value, expiry);        }        /// 
              /// 保存多个key value        /// 
        /// 
      key        /// 
              public bool StringSet(KeyValuePair
      
       [] arr)        {            return CacheRedis.StringSet(arr);        }        /// 
               /// 批量存值        /// 
        /// 
       key        /// 
       The value.        /// 
       
        
         true
         if XXXX, 
        
         false
         otherwise.
               public bool StringSetMany(string[] keysStr, string[] valuesStr)        {            var count = keysStr.Length;            var keyValuePair = new KeyValuePair
       
        [count];            for (int i = 0; i < count; i++)            {                keyValuePair[i] = new KeyValuePair
        
         (keysStr[i], valuesStr[i]);            }            return CacheRedis.StringSet(keyValuePair);        }        /// 
                 /// 保存一个对象        /// 
        /// 
                 /// 
                 /// 
                 /// 
                 public bool SetStringKey
         
          (string key, T obj, TimeSpan? expiry = default(TimeSpan?))        {            string json = JsonConvert.SerializeObject(obj);            return CacheRedis.StringSet(key, json, expiry);        }        /// 
           /// 追加值 /// 
        /// 
                  /// 
                  public void StringAppend(string key, string value)        {            ////追加值，返回追加后长度            long appendlong = CacheRedis.StringAppend(key, value);        }        /// 
           /// 获取单个key的值 /// 
        /// 
          Redis Key        /// 
                  public RedisValue GetStringKey(string key)        {            return CacheRedis.StringGet(key);        }        /// 
           /// 根据Key获取值 /// 
        /// 
          键值        /// 
          
           System.String.
                  public string StringGet(string key)        {            try            {                return CacheRedis.StringGet(key);            }            catch (Exception ex)            {                return null;            }        }        /// 
           /// 获取多个Key /// 
        /// 
          Redis Key集合        /// 
                  public RedisValue[] GetStringKey(List
          
            listKey) { return CacheRedis.StringGet(listKey.ToArray()); } /// 
            /// 批量获取值 /// 
 public string[] StringGetMany(string[] keyStrs) { var count = keyStrs.Length; var keys = new RedisKey[count]; var addrs = new string[count]; for (var i = 0; i < count; i++) { keys[i] = keyStrs[i]; } try { var values = CacheRedis.StringGet(keys); for (var i = 0; i < values.Length; i++) { addrs[i] = values[i]; } return addrs; } catch (Exception ex) { return null; } } /// 
            /// 获取一个key的对象 /// 
 /// 
            /// 
            /// 
            public T GetStringKey
           
            (string key) { return JsonConvert.DeserializeObject
            
             (CacheRedis.StringGet(key)); } #endregion #region --删除设置过期-- /// 
              /// 删除单个key /// 
 /// 
             redis key /// 
             
              是否删除成功
              public bool KeyDelete(string key) { return CacheRedis.KeyDelete(key); } /// 
              /// 删除多个key /// 
 /// 
             rediskey /// 
             
              成功删除的个数
              public long KeyDelete(RedisKey[] keys) { return CacheRedis.KeyDelete(keys); } /// 
              /// 判断key是否存储 /// 
 /// 
             redis key /// 
              public bool KeyExists(string key) { return CacheRedis.KeyExists(key); } /// 
              /// 重新命名key /// 
 /// 
             就的redis key /// 
             新的redis key /// 
              public bool KeyRename(string key, string newKey) { return CacheRedis.KeyRename(key, newKey); } /// 
              /// 删除hasekey /// 
 /// 
              /// 
              /// 
              public bool HaseDelete(RedisKey key, RedisValue hashField) { return CacheRedis.HashDelete(key, hashField); } /// 
              /// 移除hash中的某值 /// 
 /// 
              /// 
              /// 
              /// 
              public bool HashRemove(string key, string dataKey) { return CacheRedis.HashDelete(key, dataKey); } /// 
              /// 设置缓存过期 /// 
 /// 
              /// 
              public void SetExpire(string key, DateTime datetime) { CacheRedis.KeyExpire(key, datetime); } #endregion }}

View Code

NetCore配置文件为：

{  "Logging": {    "LogLevel": {      "Default": "Information",      "Microsoft": "Warning",      "Microsoft.Hosting.Lifetime": "Information"    }  },  "AllowedHosts": "*",  "tokenManagement": {    "secret": "123456123456123456",    "issuer": "webapi.cn",    "audience": "WebApi",    "accessExpiration": 120,    "refreshExpiration": 60  },  "ConnectionStrings": {    "aixueshi_temp1Context": "我的数据库连接;",    "RedisConnectionStrings": "127.0.0.1:6379"  }}

控制器端调用的代码如下：

using System;using System.Collections.Generic;using System.IdentityModel.Tokens.Jwt;using System.Linq;using System.Threading.Tasks;using FranchiseeApi.Helper;using FranchiseeCommon;using FranchiseeDto;using FranchiseeDto.Franchisee;using FranchiseeInterface.Franchisee;using Microsoft.AspNetCore.Authorization;using Microsoft.AspNetCore.Http;using Microsoft.AspNetCore.Mvc;using Microsoft.AspNetCore.Mvc.RazorPages;using Microsoft.Extensions.Configuration;using Newtonsoft.Json;using Newtonsoft.Json.Linq;namespace FranchiseeApi{    [Route("Api/V1/RedisApi")]    [ApiExplorerSettings(GroupName = "V1")]    public class RedisApiController : ControllerBase    {        private IConfigurationRoot ConfigRoot;        private readonly RedisHelper rd;        public RedisApiController(IConfiguration configRoot)        {            ConfigRoot = (IConfigurationRoot)configRoot;            rd = new RedisHelper(0, ConfigRoot["ConnectionStrings:RedisConnectionStrings"]);        }        [HttpGet]        [Route("RedisSet")]        public IActionResult RedisSet()        {            rd.StringSet("sName", "陈卧龙");            return Ok();        }        [HttpGet]        [Route("RedisGet")]        public IActionResult RedisGet(string key)        {            if (rd.KeyExists(key))            {                /*                 * 如果缓存中存在，则直接返回结果                 */                var result = rd.StringGet(key);                return Ok(result);            }            else            {                /*                 * 如果缓存中不存在，则需要结合数据库进行查询，但必须采用相应的策略，防止恶意【缓存击穿】。                 * 数据库查询部分，                 * 如果数据库查询到结果，则对结果进行缓存，并返回结果。                 * 如果数据库查询不到结果，则对请求的数据进行缓存，防止缓存击穿。                 * 除了上述比较被动的防御以外，我们还可以采取一段时间内限制请求次数来达到恶意攻击行为。                 */                return Ok();            }        }    }}

Redis的数据结构应用场景

String：

key-value结构中，value不仅可以是String，也可以是数字类型。可以应用在比如博客粉丝数量、评论数量、阅读数量的缓存。redis也提供了计数器类型的命令(incr、decr等)

Hash：

Hash表中可以储存多个K-V结构元素，可以用来储存用户的信息模块。

key=User123value={“id”: 1,“name”: “BengHiong”,“age”: 21,“location”: “guangdong”}

List：

链表结构，可以应用于显示某一列信息（如用户关注列表、粉丝列表、作品列表等），还可以通过lrange命令进行分页查询，由于redis速度快特性，实现用户不断下拉操作数据仍能快速呈现的效果。

Set：

Set结构自带了排重功能，可以通过交集命令sinterstore实现多个用户共同好友、共同关注、共同爱好的功能，也可以用sdiffstore命令实现体现用户如Q群未加好友的陌生人。

SortedSet：

相比Set多了个socre权重，实现了排序功能。可以应用于例如游戏排行榜、礼物排行榜功能。

Redis为什么能这么快？

完全基于内存，绝大部分请求是纯粹的内存操作，非常快速。数据存在内存中。

结构类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)；

采用单线程，避免了不必要的上下文切换和竞争条件，也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗；

为啥是单线程？

redis由于是内存操作，速度非常快，所以使用单线程不必过于担心某个操作时间占用较长导致其他操作长时间阻塞，而且单线程模式下可以减少线程间的切换和竞争开销，以及不需要加锁机制，提高了性能。（由于是单线程，尽量避免那些操作时间较长的命令如 keys *），多线程的话需要加锁，还需要线程间大量的cache同步。因为每一次命令执行时间短，所以综合下来，单线程才是最优方案

注意： redis 单线程指的是网络请求模块使用了一个线程，即一个线程处理所有网络请求，其他模块仍用了多个线程。（比如持久化操作需要fork一个子进程进行数据备份操作）

Redis的内存淘汰机制

expire time：redis中会给每个key设定过期时间，时间到了之后会有redis进行移除工作，这在缓存中是非常必要的，因为缓存空间毕竟是有限的。那当到了过期时间时，redis是怎么将这个key移除的呢？这就要说到redis的两种删除方式了。定期删除和惰性删除

定期删除：

redis默认是每隔 100ms 就随机抽取一些设置了过期时间的key，检查其是否过期，如果过期就删除。注意这里是随机抽取的。为什么要随机呢？你想一想假如 redis 存了几十万个 key ，每隔100ms就遍历所有的设置过期时间的 key 的话，就会给 CPU 带来很大的负载！

惰性删除：

定期删除可能会导致很多过期 key 到了时间并没有被删除掉。所以就有了惰性删除。假如你的过期 key，靠定期删除没有被删除掉，还停留在内存里，除非你的系统去查一下那个 key，才会被redis给删除掉。这就是所谓的惰性删除，也是够懒的哈！

如何解决 Redis 的并发竞争 Key 问题

所谓 Redis 的并发竞争 Key 的问题也就是多个系统同时对一个 key 进行操作，但是最后执行的顺序和我们期望的顺序不同，这样也就导致了结果的不同！（如果是单系统就不需要考虑这个问题了，因为redis本身是单线程的）

推荐一种方案：分布式锁（zookeeper 和 redis 都可以实现分布式锁）。（如果不存在 Redis 的并发竞争 Key 问题，不要使用分布式锁，这样会影响性能）

基于zookeeper临时有序节点可以实现的分布式锁。大致思想为：每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。完成业务流程后，删除对应的子节点释放锁。

@天才卧龙的博客

上一篇：消息队列 -RabbitMQ

下一篇：NetCore3.1 如何添加带有JWT Token 验证的Swagger

发表评论

关于作者

喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！

-- 愿君每日到此一游！

发表评论

最新留言

关于作者

推荐文章