分布式锁
处理补货问题的时候，可以使用watch来监控添加数量，防止重复添加，但使用watch的时候只能监控到要改变的key是否改变了，在超卖问题下，不仅要监控key是否改变了还要求各个客户端不能进行操作。这里我们需要使用setnx这个分布式锁来操作：
setnx key value
有值返回设置失败【无控制权】，无值返回设置成功【有控制权】
127.0.0.1:6379> flushdb OK 127.0.0.1:6379> setnx lock-num 1 //拿到锁 OK 127.0.0.1:6379> expire lock-num 10 (integer) 1 127.0.0.1:6379> set num 10 OK 127.0.0.1:6379> incr num (integer) 11 127.0.0.1:6379> del lock-num //删除锁 (integer) 1 127.0.0.1:6379> get num "11" 127.0.0.1:6379>
为了防止客户端拿到锁后宕机，我们一般需要给该锁设置一个过期时间，防止发生死锁，关于分布式锁的实现和优化，详细见下一小节的介绍。
分布式锁的实现及改进策略
在分布式的场景下，我们需要考虑并发问题，避免多个线程同时对同一个资源执行操作，造成不可预期的后果，这个时候就需要给资源加分布式的锁，来保证资源的独占。
简易SetNx锁
对整个消息队列，要想控制线程对消息队列的单独处理，就需要给整个队列加一个分组锁，而分组锁可以用Redis的SetNx实现，语义是：设置key，如果存在则返回flase，如果不存在则设置成功并返回true，即设置和判断是一个原子操作：
队列锁即对队列的控制权，我们可以设定一个约定好的常量值即可，使用Redis的特性SetNx操作，天然的原子操作：只在键 key 不存在的情况下， 将键 key 的值设置为 value 。若键 key 已经存在， 则 SETNX 命令不做任何动作，需要注意的是队列锁需要设置过期时间，如果线程意外关闭没有来得及释放队列锁，会导致死锁。
/// <summary> /// 获取缓存队列的队列锁 /// </summary> /// <returns></returns> public static bool GetQueueLock() { //申请成功标志 const string queueLockkey = "QueueLock"; const int tenantId = TenantIdInRedis; try { using (var redis = new RedisNativeProviderV2(KeySpaceInRedis, tenantId)) { //redis中，如果返回true设置成功代表队列锁空闲，如果返回false设置失败表明队列锁正在被持有 if (RedisSetNx(queueLockkey)) //如果为true,设置队列锁并设置该锁的过期时间 { RedisExpire(queueLockkey, 600);//设置过期时间为10分钟 return true; } } } catch (Exception ex) { //进行查询异常操作 Loggging.Error($"在redis 设置队列锁[{queueLockkey}]异常", ex); //抛出异常 } return false; } /// <summary> /// 给缓存队列的队列解锁 /// </summary> /// <returns></returns> public static bool QueueUnLock() { //申请成功标志 const string queueLockkey = "QueueLock"; const int tenantId = TenantIdInRedis; try { using (var redis = new RedisNativeProviderV2(KeySpaceInRedis, tenantId)) { return RedisDeleteKey(queueLockkey); //redis中，如果返回true设置成功代表原来不存在这样的分组，如果返回false设置失败表明原来存在这样的分组 } } catch (Exception ex) { //进行查询异常操作 Loggging.Error($"在redis 解除队列锁[{queueLockkey}]异常", ex); //抛出异常 } return false; }
SetNx的执代码如下：
/// <summary> /// 如果返回true设置成功代表原来不存在这样的锁，如果返回false设置失败表明原来存在这样的锁 /// </summary> /// <param name="key"></param> /// <returns></returns> public static bool RedisSetNx(string key) { const int tenantId = TenantIdInRedis; try { using (var redis = new RedisNativeProviderV2(KeySpaceInRedis, tenantId)) { return redis.SetNx(key, StringToBytes(key)); } } catch (Exception ex) { //进行锁创建异常的操作 Loggging.Error($"在redis setNx[{key}]:[{key}]异常", ex); } finally { //进行锁创建成功或失败的操作 Loggging.Info($"在redis setNx[{key}]:[{key}]"); } return false; }
单机状态下SetNx锁的改进策略
一个高可用的分布式锁应该满足如下几点要求：
- 互斥性：任意时刻只能有一个客户端拥有锁，不能被多个客户端获取
- 安全性：锁只能被持有该锁的客户端删除，不能被其它客户端删除
- 防死锁：获取锁的客户端因为某些原因而宕机，而未能释放锁，其它客户端也就无法获取该锁，需要有机制来避免该类问题的发生
- 高可用：当部分节点宕机，客户端仍能获取锁或者释放锁
之前的设计，通过setnx实际上只解决了互斥性的问题，其它的都不满足，那么基于问题如何再进行拓展设计呢？
解决SetNx的超时非原子操作问题【互斥性、防死锁】
可以看到简易SetNx锁中，setnx命令无法原子性的设置锁的自身过期时间，也就是说执行setnx命令时我们无法同时设置其过期时间，那么就会出现死锁，例如：客户端A刚执行完setnx，这时候客户端A挂掉了，没有完成给锁设置过期时间，此时就产生了死锁，所有的客户端再也无法获得该锁，这种情况一般采用Lua脚本来实现（因为Redis执行Lua脚本是原子性的），其实从 Redis 2.6.12 版本开始set命令完全可以替代setnx命令，我们看官网的set命令参数
SET key value [EX seconds] [PX milliseconds] [NX|XX]
- EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。
- PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
- NX ：只在键不存在时，才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value 。
- XX ：只在键已经存在时，才对键进行设置操作。
例如：SET key value NX PX 30000 这个命令的作用是在只有这个key不存在的时候才会设置这个key的值（NX选项的作用），超时时间设为30000毫秒（PX选项的作用）。那么我们用set命令带上EX或者PX、以及NX参数就满足了上面提到的互斥性（加锁）、死锁（自动过期）两个要求。
解决非本线程对锁的删除问题【安全性】
如果客户端A拿到锁并设置了锁的过期时间为10S，但是由于某种原因客户端A执行时间超过了10S，此时锁自动过期，那么客户端B拿到了锁，然后客户端A此时正好执行完毕删除锁，但是此时删除的是客户端B加的锁，如何防止这种不安全的情况发生呢？有两种方案：
给锁自动续期，执行完任务再删除
解锁实际就是删除缓存key，简易SetNx锁调用后就立即执行删除，没有判断现在的锁还是不是自己的锁了，我们可以让获得锁的线程开启一个守护线程，用来给自己的锁“续期”。例如设置的超时时间为10秒，则有如下的判断方法：
- 当过去了9S，客户端A还没执行完，守护线程会执行expire指令，把锁再“续期”10S,
- 守护线程从第9S开始执行，每9秒执行一次。
- 当客户端A执行完任务，会显式关掉守护线程。
如果客户端A忽然宕机，由于A线程和守护线程在同一个进程，守护线程也会停下。这把锁到了超时的时候，没人给它续期，也就自动释放了。
设置锁的唯一标识【判断和删除原子操作】
可以在加锁的时候把set的value值设置成一个唯一标识，标识这个锁是谁加的锁，在删除锁的时候判断是不是自己加的那把锁，如果不是则不删除。例如加上自己的线程号作为唯一标识。当然这里会有一个新问题产生，判断是不是自己加的锁和释放锁是两个独立操作，不是原子性，所以我们需要使用Lua脚本执行判断和释放锁。
if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end
集群状态下SetNx锁的改进策略
从纯集群的状态下去考虑可能遇到的问题，当然单机遇到的问题集群一定都会遇到，只是场景更为复合。在大型的应用中，一般Redis服务都是集群形式，主从模式下：
- master挂掉后，slave选举上来成为master时容易出现问题，redis在进行主从复制时是异步完成的，比如在clientA获取锁后，主redis崩溃了，数据还没有复制到从redis中，
- 从redis选举出一个升级为主redis,造成新的主redis没有clientA 设置的锁，这时clientB尝试获取锁，并且能够成功获取锁，导致互斥失效。
这个时候就只能使用终极大招，Redis之父创建的Redlock算法了，有两个前置概念
- TTL：Time To Live;只 redis key 的过期时间或有效生存时间
- clock drift:时钟漂移；指两个电脑间时间流速基本相同的情况下，两个电脑（或两个进程间）时间的差值；如果电脑距离过远会造成时钟漂移值 过大
时钟漂移相对于TTL来说要小的多，我们采用RedLock算法来解决这个问题：在分布式版本的算法里我们假设我们有N个Redis Master节点，这些节点都是完全独立的。不使用任何复制或者其他隐含的分布式协调算法（如果采用的是Redis Cluster集群此方案可能不适用，因为Redis Cluster是按哈希槽 (hash slot)的方式来分配到不同节点上的，明显存在分布式协调算法）。
RedLock算法
我们把N设成5，因此我们需要在不同的计算机或者虚拟机上运行5个master节点来保证他们大多数情况下都不会同时宕机。一个客户端需要做如下操作来获取锁：
- 获取时间：5台机器同时获取当前系统时间（单位是毫秒）
- 轮流请求锁：轮流用相同的key和随机值（客户端的唯一标识）在5个节点上请求锁，在这一步里，客户端在每个master上请求锁时，会有一个和总的锁释放时间相比小的多的超时时间。比如如果锁自动释放时间是10秒钟，那每个节点锁请求的超时时间可能是5-50毫秒的范围，这个可以防止一个客户端在某个宕掉的master节点上阻塞过长时间，如果一个master节点不可用了，我们应该尽快尝试下一个master节点。
- 计算获取锁时间：客户端计算上一步中获取锁所花的时间，只有当客户端在大多数master节点上成功获取了锁（在这里是3个），而且总共消耗的时间不超过锁释放时间，这个锁就认为是获取成功了。
- 锁持有时间：如果锁获取成功了，那现在锁自动释放时间就是最初的锁释放时间减去之前获取锁所消耗的时间【例如获取花了3秒，锁释放为10秒，则不超过，并且实际使用时间为7秒，实际应该再减去时钟漂移，但可忽略不计】。
- 释放锁：如果锁获取失败了，不管是因为获取成功的锁不超过一半（N/2+1)还是因为总消耗时间超过了锁释放时间，客户端都会到每个master节点上释放锁，即便是那些他认为没有获取成功的锁。由于释放锁时会判断这个锁的value是不是自己设置的，如果是才删除，所以在释放锁时非常简单，只要向所有实例都发出释放锁的命令，不用考虑能否成功释放锁【也解决了判定是否是自己加的锁防止误删的问题】
以上就是整个RedLock的算法流程。
RedLock注意事项
有以下几个概念点需要厘清，关于RedLock的注意点：
- RedLock可以看成是同步算法：因为 即使进程间（多个电脑间）没有同步时钟，但是每个进程时间流速大致相同；并且时钟漂移相对于TTL小，可以忽略，所以可以看成同步算法
- RedLock失败重试机制：当client不能获取锁时，应该在随机时间后重试获取锁；并且最好在同一时刻并发的把set命令发送给所有redis实例；而且对于已经获取锁的client在完成任务后要及时释放锁，这是为了节省时间；
- 各节点Redis需要有持久化机制：如果redis没有持久化功能，在clientA获取锁成功后，所有redis重启，clientB能够再次获取到锁，这样违法了锁的排他互斥性; 如果启动AOF永久化存储，重启redis后由于redis过期机制是按照unix时间戳走的，所以在重启后，然后会按照规定的时间过期，不影响业务;但是由于AOF同步到磁盘的方式默认是每秒1次，如果在一秒内断电，会导致数据丢失，立即重启会造成锁互斥性失效;但如果同步磁盘方式使用Always(每一个写命令都同步到硬盘)造成性能急剧下降;所以锁完全有效性和性能方面要有所取舍;
- 锁的有效获取时间计算方式：先假设client获取所有实例，所有实例包含相同的key和过期时间(TTL) ,但每个实例set命令时间不同导致不能同时过期，第一个set命令之前是T1,最后一个set命令后为T2,则此client有效获取锁的最小时间为TTL-(T2-T1)-时钟漂移;
虽然说RedLock算法可以解决单点Redis分布式锁的高可用问题，但如果集群中有节点发生崩溃重启，还是会出现锁的安全性问题
RedLock的极致有效
假设一共有A, B, C, D, E，5个Redis节点，设想发生了如下的事件序列：
- 客户端1成功锁住了A, B, C，获取锁成功（但D和E没有锁住）
- 节点C崩溃重启了，但客户端1在C上加的锁没有持久化下来，丢失了
- 节点C重启后，客户端2锁住了C, D, E，获取锁成功
这样，客户端1和客户端2同时获得了锁（针对同一资源）。针对这样场景，解决方式也很简单， 断电后等待TTL后重启：即使断电情况也能有效保证锁完全有效性及性能高效：redis同步到磁盘方式保持默认每秒，在redis无论停掉后要等待TTL时间后再重启【这种情况下相当于客户端1放弃了自己的锁，客户端2设置后唯一，仍然互斥有效】(延迟重启) ，缺点是 TTL时间内服务相当于暂停状态
Redis的删除策略
什么是过期数据，当我们执行del删除redis数据以及expire过期的就属于过期数据，也就是我们现在或将来一定要删除的数据，为了防止CPU压力过大，采取不同的删除策略。
过期数据的删除策略
和ElasticSearch的定期段合并策略相同，Redis处理过期数据也有一套。时效性数据的内存结构如下，有个espires的hash结构来存储数据的内存地址和时间，到时删除
删除策略的目标是在内存占用和CPU性能之间寻找一种平衡，既不能让过期数据过多，又不能让CPU太忙。目前有三种处理策略：
定时删除
创建一个定时器，当key设置有过期时间，且过期时间到达时，由定时器任务立即执行对键的删除操作。
- 优点：节约内存，到时就删除，快速释放掉不必要的内存占用
- 缺点：CPU压力很大，无论CPU此时负载量多高，均占用CPU，会影响redis服务器响应时间和指令吞吐量
总而言之一句话：牺牲时间换取空间
惰性删除
定时删除可能会导致很多过期key到了时间并没有被删除掉。所以就有了惰性删除。过期key，依然停留在内存里，除非访问一下过期key，才会被redis给删除掉。这就是所谓的惰性删除。expireIfNeeded(),检查数据是否过期，执行get的时候调用
- 优点：节约CPU性能，发现必须删除的时候才删除
- 缺点：内存压力很大，出现长期占用内存的数据
总而言之一句话：牺牲空间换时间
定期删除
当然同AOF的几种策略相比，删除策略也不会走极端，于是就有了定期删除的策略：
这几个参数在配置中可以指定：
hz 10
周期性轮询redis库中的时效性数据，采用随机抽取的策略，利用过期数据占比的方式控制删除频度.
- 优点： CPU性能占用设置有峰值，检测频度可自定义设置，内存压力不是很大，长期占用内存的冷数据会被持续清理
总而言之一句话：空间和时间具有平衡性
删除策略对比
三种删除策略及对比如下：
| 删除策略 | 特点 | 执行特点 | 总结 |
| 定时删除 | 节约内存，消耗CPU | 不分时段执行，内存占用低，CPU损耗高 | 牺牲时间换空间 |
| 惰性删除 | 内存占用验证，CPU消耗低 | 延迟执行，内存占用高，CPU损耗低 | 牺牲空间换时间 |
| 定期删除 | 内存定期随机清理 | 每秒花费固定CPU资源维护内存，处理时间久的冷数据 | 定时抽查，重点抽查 |
一般会组合惰性删除和定期删除进行使用。
Redis的逐出算法
当新数据进入redis时，如果内存不足怎么办？Redis使用内存存储数据，在执行每一个命令前，会调用freeMemoryIfNeeded()检测内存是否充足。如果内存不满足新加入数据的最低存储要求，redis要临时删除一些数据为当前指令清理存储空间。清理数据的策略称为逐出算法。
首先需要注意几个参数，这几个参数决定了逐出算法的后续逻辑。
- 最大可使用内存maxmemory：占用物理内存的比例，默认值为0，表示不限制。通常设置在50%以上
- 每次选取待删除数据的个数maxmemory-samples： 选取数据时并不会全库扫描，导致严重的性能损耗，降低读写性能。因此采用随机获取数据的方式作为待检测删除数据
- 删除策略maxmemory-policy ：达到最大内存后，对被选出来的数据进行删除的策略
逐出算法有三大类：
检测易失性数据（可能会过期的数据集server.db[i].expires）
检测易失数据有四种算法：
- volatile-lru --> 从已设置过期时间的数据集中挑选最近最少使用的数据淘汰【时间】
- volatile-lfu–>从已设置过期时间的数据集中挑选最不经常使用的数据淘汰【次数】
- volatile-ttl–>从已设置过期时间的数据集中挑选将要过期的数据淘汰
- volatile-random -->从已设置过期时间的数据集中任意选择数据淘汰
检测全库数据（所有数据集server.db[i].dict）
检测易失数据有三种算法：
- allkeys-lru --> 当内存不足以容纳新写入数据时，在键空间中，移除最近最少使用的key（最常用）
- allkeys-random–>从数据集中任意选择数据淘汰
- allkeys-lfu–>当内存不足以容纳新写入数据时，在键空间中，移除最不经常使用的key
因为不需要关心数据是否过期，所以没有volatile-ttl
放弃数据驱逐
no-eviction–>禁止驱逐数据（redis4.0默认策略），也就是说当内存不足以容纳新写入数据时，新写入操作或报错，回引发OOM（Out of memory）
综合以上，我们推荐使用的策略是：maxmemory-policy volatile-lru
无法逐出的情况
逐出数据的过程不是100%能够清理出足够的可使用的内存空间，如果不成功则反复执行。当对所有数据尝试完毕后，如果不能达到内存清理的要求【均无过期数据，且占用内存已满】，将出现错误信息。
（err）OOM command not allowed when used memory > 'maxmemory'
Cluster集群模式
其实Redis的集群有三种模式，主从复制模式、哨兵模式以及Cluster集群模式。其中Cluster集群模式可以说是当前最成熟的解决方案。所以本节重点介绍这部分内容。
Cluster模式，它一定意义上也是基于主从复制模式的，只不过比主从复制模式更加强大，不仅做到了主从的读写分离包括读的负载均衡，还能进行很好的写的负载均衡：
- 高可扩展， 分散单台服务器的访问压力，实现负载均衡
- 高可扩展，分散单台服务器的存储压力，实现可扩展性
- 高可用， 降低单台服务器宕机带来的业务灾难
具备了以上特点的Cluster架构如下：
Cluster集群结构设计
数据写入时会依据CRC算key，计算结果再对16384个插槽取余，然后放置到指定主服务器。因为初始化机器的时候就会给每个主初始化一定数量的槽，数据放到哪个槽一定是有明确的路由地址的：
如果直接访问一次命中就直接取，否则就由客户端直接去目标机器找，因为存在地址簿，所以最多两次命中：
槽是集群建立之初或集群加减机器时都会动态调整和变化的。每个机器分为若干个槽slot，加机器和减机器都可以通过动态调整槽来实现。
数据存取的方式如下：
[root@192 redis-6.0.8]# redis-cli -h 192.168.5.101 192.168.5.101:6379> set love guochengyu (error) MOVED 16198 192.168.5.103:6379
可以看的出存储到了插槽16198，被路由到了master3上了，所以需要使用重定向的方式插入和获取，只需要直接在客户端命令后边加个-c即可：
[root@192 redis-6.0.8]# redis-cli -h 192.168.5.101 -c 192.168.5.101:6379> set love guochenyubaobei -> Redirected to slot [16198] located at 192.168.5.103:6379 OK 192.168.5.103:6379> get love "guochenyubaobei" 192.168.5.103:6379>
