【烈日炎炎战后端】Redis（6.1万字）

ゞ浴缸里的玫瑰 2022-11-24 11:21 96阅读 0赞

### Redis ###

*   *   *   *  1. Redis是什么?
             *   *   *  \[1\] 什么是远程字典服务器？
                     *  \[2\] 为什么redis是高性能的？
                     *  \[3\] 分布式的含义？
                     *  \[4\] NoSQL的含义
             *  2. Redis有哪五种不同类型的值？应用场景有哪些？
             *   *   *  \[1\] String（字符串）
                     *  \[2\] Hash（哈希）
                     *  \[3\] List（列表）
                     *  \[4\] Set（集合）
                     *  \[5\] zset (sorted set：有序集合)
             *  3. 使用redis可能出现的问题
             *   *   *  \[1\] 缓存雪崩
                     *  \[2\] 缓存击穿
                     *  \[3\] 缓存穿透
                     *  \[4\] 三种问题区别记忆方法
                     *  \[5\] 数据库和缓存的双写一致性问题
             *  5. Redis底层数据结构
             *  6. Redis的过期策略
             *   *   *  \[1\] 定期删除（主动）
                     *  \[2\]惰性删除（被动）
                     *  \[3\] 定期删除+惰性删除
                     *  \[4\] 内存淘汰机制
                     *  \[5\] LRU算法
             *  7. Redis分布式锁
             *   *   *  \[1\] 分布式锁
                     *  \[2\] redis
                     *  \[3\] redisson
                     *  \[4\] 牛客网总结
                     *  \[5\] 常用的分布式锁
                     *  \[6\] 分布式锁三种实现方式
             *  8. Redis的I/O多路复用
             *   *   *  \[1\] 为什么 Redis 中要使用 I/O 多路复用？
                     *  \[2\] I/O 多路复用模型
                     *  \[3\] select
                     *  \[4\] poll
                     *  \[5\] epoll
                     *  \[6\] Reactor 设计模式
                     *  \[7\] 封装 epoll 函数
                     *  \[8\] select & poll & epoll比较
                     *  \[9\] 深入理解select、poll和epoll及区别：杂文
             *  9. 多线程的 Redis
             *  10 Redis持久化方式
             *  11. Redis哨兵（Sentinel）模式

> **为什么有些概念很难理解？**  
> **例1**：redis是一款远程内存数据库.  
> **例2**：熊猫是一种哺乳动物.  
> 很显然，例2的句子比例1句子更容易理解.**例1和例2是同样的语法结构**–主谓宾.**不同的是词语本身的含义**：“哺乳动物”是常见的词语，我们都对它的含义很熟悉；而“远程内存数据库”对于初学者来说却晦涩难懂，并不常见.也就是说，句子结构并不是我们去理解概念的阻碍，而是词语本身的含义.或许在我们理解了“远程内存数据库”中的词语，概念本身就并不陌生.  
> 所以本文对于新事物的学习思想是：**对专业名词追根溯源**.

#### 1. Redis是什么? ####

Redis (REmote DIctionary Server)是一个开源（BSD许可），内存存储的数据结构服务器，可用作数据库，高速缓存和消息队列，是一个高性能的key-value数据库.  
Redis与其他key-value缓存产品有以下三个特点：

*  Redis**支持数据的持久化**，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用.
 *  Redis**不仅支持简单的key-value类型的数据**，同时还提供list，set，zset，hash等数据结构的存储.
 *  Redis**支持数据的备**份，即master-slave模式的数据备份.

**为什么要用Redis**

*  **性能极高** – Redis读的速度是110000次/s，写的速度是81000次/s .
 *  **丰富的数据类型** – Redis支持Strings， Lists， Hashes， Sets 及 Ordered Sets 数据类型操作.
 *  **原子** – Redis的所有操作都是原子性的，即要么成功执行要么失败完全不执行.单个操作是原子性的.多个操作也支持事务，即原子性，通过MULTI和EXEC指令包起来.
 *  **丰富的特性** – Redis还支持publish/subscribe， key过期等特性.

> 总结：
> 
> 1.  Redis（Remote Dictionary Server远程字典服务器)，是一款高性能的分布式数据库，基于**内存**运行并支持**持久化**的**NoSQL**数据库.
> 2.  它可以储存键（key）与**string**、**hash**、**list**、**set**和s**orted se**t五种不同类型的值（value）之间的映射.
> 3.  使用redis可能出现**缓存雪崩**、**缓存击穿**、**缓存穿透**和数据库和缓存的双写一致性问题等问题.

###### \[1\] 什么是远程字典服务器？ ######

**远程**：有远程的就有本地的.本地的意思就是你自己的电脑，远程就是指别人的电脑.  
**字典**：就像普通的字典一样，数据字典就是用来对数据进行定义和解释的.它可以用作：数据库、缓存和消息中间件.  
**服务器**：服务器就是远程电脑，服务器可以在网络中为其它客户机提供计算或者应用服务.

###### \[2\] 为什么redis是高性能的？ ######

1）基于**内存**实现，完全内存计算；  
2）**单线程**操作，避免了**线程上下文切换**操作；  
3）**多路I/O复用**的线程模型，实现了一个线程监控多个IO流，及时响应请求；  
4）Redis对外部的依赖比较少，属于轻量级内存数据库.

> > **内存**: 内存的访问速度是很快的，常用的DDR4内存的读写速度是机械硬盘的30倍；
> 
> > **单线程**：避免了线程上下文切换和竞争条件.也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗
> 
> > **什么是线程上下文切换？**  
> > 即使是单核CPU也支持多线程执行代码，CPU通过给每个线程分配CPU时间片来实现这个机制.时间片是CPU分配给各个线程的时间，因为时间片非常短，所以CPU通过不停地切换线程执行，让我们感觉多个线程时同时执行的，时间片一般是几十毫秒（ms）.**CPU通过时间片分配算法来循环执行任务，当前任务执行一个时间片后会切换到下一个任务.但是，在切换前会保存上一个任务的状态，以便下次切换回这个任务时，可以再次加载这个任务的状态，从任务保存到再加载的过程就是一次上下文切换**.  
> > 这就像我们同时读两本书，当我们在读一本英文的技术书籍时，发现某个单词不认识， 于是便打开中英文词典，但是在放下英文书籍之前，大脑必须先记住这本书读到了多少页的第多少行，等查完单词之后，能够继续读这本书.这样的切换是会影响读书效率的，同样上下文切换也会影响多线程的执行速度.  
> > **I/O多路复用技术是什么，为什么很快？**  
> > 常见的IO模型有阻塞模型、非阻塞模型、IO多路复用，信号驱动IO，异步.
> > 
> > 1.  阻塞式IO: 使用系统调用，并一直阻塞直到内核将数据准备好，之后再由内核缓冲区复制到用户态，在等待内核准备的这段时间什么也干不了.
> > 2.  非阻塞式IO：内核在没有准备好数据的时候会返回错误码，而调用程序不会休眠，而是不断轮询询问内核数据是否准备好
> > 3.  IO多路复用：类似非阻塞，只不过轮询不是由用户线程去执行，而是由内核去轮询，内核监听程序监听到数据准备好后，调用内核函数复制数据到用户空间
> > 4.  信号驱动IO： 应用进程使用 sigaction 系统调用，内核立即返回，应用进程可以继续执行，也就是说等待数据阶段应用进程是非阻塞的.
> > 5.  异步IO：应用进程执行 aio\_read 系统调用会立即返回，应用进程可以继续执行，不会被阻塞，内核会在所有操作完成之后向应用进程发送信号.
> 
> > **不好理解？举例说明:**
> > 
> > 1.  阻塞：比如某个时候你在去餐厅吃饭，又想出去逛街.但是你不知道饭菜什么时候来，只好坐在餐厅里面等.直到做好，然后吃完才离开去做逛街.这就是典型的阻塞.
> > 2.  非阻塞忙轮询：接着上面的例子，如果用忙轮询的方法，出去我们逛一会，然后每分钟回来问一下服务员：“饭菜好了没？”
> > 3.  IO多路复用：餐厅安装了电子屏幕用来显示点餐的状态，逛街一会，回来就不用去询问服务员了，直接看电子屏幕就可以了.
> > 4.  异步IO：不想逛街，餐厅太吵了，回家好好休息一下.于是我们叫外卖，打个电话点餐，然后我可以在家好好休息一下，饭好了送货员送到家里来.这就是典型的异步，只需要打个电话说一下，然后可以做自己的事情，饭好了就送来了.
> > 5.  异步 I/O 与信号驱动 I/O 的区别在于，异步 I/O 的信号是通知应用进程 I/O 完成，而信号驱动 I/O 的信号是通知应用进程可以开始 I/O.
> 
> > **阻塞是个什么概念呢？**  
> > 比如某个时候你在去餐厅吃饭，又想出去逛街.但是你不知道饭菜什么时候来，只好坐在餐厅里面等.直到做好，然后吃完才离开去做逛街.这就是典型的阻塞.

###### \[3\] 分布式的含义？ ######

把一个需要非常巨大的计算能力才能解决的问题分成许多小的部分，然后把这些部分分配给多个计算机进行处理，最后把这些计算结果综合起来得到最终的结果

###### \[4\] NoSQL的含义 ######

**NoSQL**泛指非关系型的数据库.与之对应的是关系数据库SQL.关系数据库的查询经常会关联两个表的数据，而NoSQL不关联表.

#### 2. Redis有哪五种不同类型的值？应用场景有哪些？ ####

1.  **string**：redis 中字符串 value 最大可为512M.可以用来做一些**计数器**（也是实际工作中最常见的）.
2.  **hash**：键值对集合，是一个字符串类型的 Key和 Value 的映射表，也就是说其存储的Value是一个键值对（Key- Value）.可以用来存放一些具有特定结构的信息.用于**用户信息管理**.
3.  **list**：简单的字符串列表，按照插入顺序排序，可以添加一个元素到列表的头部（左边）或者尾部（右边），其底层实现是一个链表.可以**实现一个简单消息队列**功能，做基于redis的分页功能等.
4.  **set**：是一个字符串类型的无序集合.可以用来进行全局去重等.做**用户标签**.
5.  **sorted set**：是一个字符串类型的有序集合，给每一个元素一个固定的分数score来保持顺序.可以用来做**排行榜应用**或者进行范围查找等.

*  补充：用**sorted set**实现一个排行榜需要注意什么问题？答：并列问题.

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-qgsRMNHI-1596677444952)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200629214430327.png)\]

**详细介绍：**

Redis支持五种数据类型：**string**（字符串），**hash**（哈希），**list**（列表），**set**（集合）及**zset**(sorted set：有序集合).每种数据类型有其适合的使用场景，下面具体介绍.

###### \[1\] String（字符串） ######

string 是 redis 最基本的类型，你可以理解成与 **Memcached** 一模一样的类型，一个 key 对应一个 value.string 类型是二进制安全的.意思是 redis 的 string 可以包含任何数据.比如jpg图片或者序列化的对象.string 类型是 Redis 最基本的数据类型，string 类型的值最大能存储 512MB.

**使用方法**

SET key value   设置指定 key 的值  
    GET key     获取指定 key 的值.
    SETEX key seconds value 将值 value 关联到 key ，并将 key 的过期时间设为 seconds (以秒为单位).

**使用场景**

1.  **会话缓存**  
    用户登录系统后，使用Redis保存用户的Session信息，每次用户查询登录信息都直接从Redis中获取.
2.  **计数器**

*  比如登录系统会限制密码错误次数，当一个用户在一定时间内连续输入密码错误，就不能登录，需要一段时间后才能登录，我们可以使用redis，把username作为key，错误的次数作为value，同时设置过期时间即可.
 *  手机验证码限收到短信的次数
 *  统计其他计数

1.  **定时器**  
    redis的key可以设置过期时间，我们基于此特性设置一个定时器.
2.  **对象**  
    我们把对象序列化后，可以使用redis保存该对象，然后在获取对象信息的时候，反序列化value
3.  **分布式锁**  
    redis提供了setnx()方法，即SET IF NOT EXIST，只有在key不存在的时候才能set成功，这就意味着同一时间多个请求只有一个请求能保存成功，这块的可以自行搜索redis的分布式锁

###### \[2\] Hash（哈希） ######

Redis hash 是一个键值(key=>value)对集合，即编程语言中的Map类型.  
Redis hash 是一个 string 类型的 field 和 value 的映射表.

**使用方法**

HSET key field value 
    将哈希表 key 中的字段 field 的值设为 value .
    HGET key field
    获取存储在哈希表中指定字段的值.
    HKEYS key
    获取所有哈希表中的字段
    HMSET key field1 value1 [field2 value2 ]
    同时将多个 field-value (域-值)对设置到哈希表 key 中.

**使用场景**

hash 特别适合用于存储对象，并且可以像数据库中update一个属性一样只修改某一项属性值(Memcached中需要取出整个字符串反序列化成对象修改完再序列化存回去)

###### \[3\] List（列表） ######

Redis 列表是简单的字符串列表，按照插入顺序排序.你可以添加一个元素到列表的头部（左边）或者尾部（右边）.

**使用方法**

LPUSHX key value 
    将一个值插入到已存在的列表头部  
    LPUSH key value1 [value2]
    将一个或多个值插入到列表头部
    LPOP key
    移出并获取列表的第一个元素
    BLPOP key1 [key2 ] timeout
    移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止.

**使用场景**

1.  **消息队列**  
    Redis的lpush+brpop命令组合即可实现阻塞队列，生产者客户端使用lrpush从列表左侧插入元素，多个消费者客户端使用brpop命令阻塞式的"抢"列表尾部的元素，多个客户端保证了消费的负载均衡和高可用性.
2.  **类目/文章/活动等列表**  
    最常见的就是各个系统的首页数据，包括电商系统的商品类目，拼团活动列表，博客园的首页文章列表等
3.  **其他**  
    根据push和pop的方式不同，有以下组合方式

lpush + lpop = Stack(栈) lpush + rpop = Queue(队列) lpush + ltrim = Capped Collection(有限集合) lpush + brpop = Message Queue(消息队列)

\`\`

###### \[4\] Set（集合） ######

Redis 的 Set 是 String 类型的无序集合.集合成员是唯一的，这就意味着集合中不能出现重复的数据.

Redis 中集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是 O(1).

**使用方法**

SADD key member1 [member2] 
    向集合添加一个或多个成员  
    SDIFF key1 [key2]
    返回给定所有集合的差集
    SINTER key1 [key2]
    返回给定所有集合的交集
    SMEMBERS key
    返回集合中的所有成员

**使用场景**

1.  标签(tag)  
    比如在点餐评价系统中，用户给某商家评价，商家会有多个评价标签，但是不会重复的，如果100万人给某商家评价打了标签，如果使用MySQL数据库获取大数据量去重后的评价标签，会影响数据库的性能和系统的并发量.
2.  相同点/异同点  
    利用交集、并集、差集等操作，可以计算两个人的共同喜好，全部的喜好，自己独有的喜好等功能.

###### \[5\] zset (sorted set：有序集合) ######

Redis 有序集合和集合一样也是string类型元素的集合，且不允许重复的成员.

不同的是每个元素都会关联一个double类型的分数.redis正是通过分数来为集合中的成员进行从小到大的排序.

有序集合的成员是唯一的，但分数(score)却可以重复.

**使用方法**

ZADD key score1 member1 [score2 member2] 
    向有序集合添加一个或多个成员，或者更新已存在成员的分数  
    ZCARD key
    获取有序集合的成员数
    ZREM key member [member ...]
    移除有序集合中的一个或多个成员

**使用场景**：

*  **排行榜**：例如博客园需要对用户发表的文章做排行榜，榜单的维度可能是多个方面的：按照时间、按照点赞数、按照热度，浏览数等

#### 3. 使用redis可能出现的问题 ####

**首先我们应该先明确缓存处理的流程**：  
前台请求，后台先从缓存中取数据，取到直接返回结果，取不到时从数据库中取，数据库取到更新缓存，并返回结果，数据库也没取到，那直接返回空结果.

###### \[1\] 缓存雪崩 ######

指**缓存由于某些原因（比如 宕机、cache服务挂了或者不响应）整体失效了，导致大量请求到达后端数据库**，从而导致数据库崩溃，整个系统崩溃，发生灾难.

*  解决办法：可以给缓存设置不同的缓存时间，更新数据使用互斥锁或者通过双缓存在避免缓存雪崩.

###### \[2\] 缓存击穿 ######

缓存击穿是指缓存中没有但数据库中有的数据.如果一些key被高并发访问\*\*，恰好在这个时间点某个key缓存过期，从而导致了大量请求达到数据库\*\*，，这就导致数据库中并发的去执行了很多不必要的查询操作，从而导致巨大冲击和压力.

*  解决办法：使用互斥锁，只让一个请求去load DB，成功之后重新写缓存，其余请求没有获取到互斥锁，可以尝试重新获取缓存中的数据.

###### \[3\] 缓存穿透 ######

是指**缓存和数据库中都没有的数据，而用户不断发起请求，导致请求都打到了数据库**上，导致数据库异常.

*  解决办法：可以使用互斥锁或者无论是否取到结果都将结果存入缓存，还可以使用有效的机制（比如布隆过滤器）来拦截不合法的key值等.

###### \[4\] 三种问题区别记忆方法 ######

在一个高频访问的应用系统中，每次用户的请求需要去DB中获取数据，会对数据库造成很大的压力、容易导致数据库的奔溃.所以才会出现缓存来分担一部分的数据库的压力. 但是使用缓存也带来了一系列问题:

> 打个比方，数据库是人，缓存是防弹衣，子弹是线程，本来防弹衣是防止子弹打到人身上的，但是当防弹衣里面没有防弹的物质时，子弹就会穿过它打到人身上.

1.  **缓存雪崩（被霰弹枪打）**：**"雪崩"对应的是大量请求**，缓存某一时间失效了，然后**大量请求**到达后端数据库，从而导致数据库崩溃.
    
    解决方法：避免同时过期，构建多级缓存，部署多个redis实例。
2.  **缓存击穿（被狙击枪多次打一点）**：**"击穿"对应的是redis某点被（key）击穿**。如果**一些key被高并发访问**，恰好在这个时间点某个key缓存过期，从而导致了大量请求达到数据库，，这就导致数据库中并发的去执行了很多不必要的查询操作，从而导致巨大冲击和压力.
    
    解决方法：加互斥锁，永不过期.
3.  **缓存穿透（被巴雷特打）**：**"穿透"最为严重，指的是缓存和数据库被穿透，都穿的透透的了**。是指缓存和数据库中都没有的数据，而用户不断发起请求，导致**请求都打到了数据库**上，导致数据库异常.这时的用户很可能是攻击者.
    
    解决方法：缓存空值，布隆过滤器.

**缓存雪崩、缓存穿透、缓存击穿的解决方案**

**缓存雪崩**

![image-20200629221647159][]

**缓存击穿**

![image-20200629221419115][image-20200629221647159] ![image-20200719220627332][image-20200629221647159]

###### \[5\] 数据库和缓存的双写一致性问题 ######

> 在**高并发**请求下很容易导致数据不一致的问题，如果你的业务需要保证数据的强一致性，那么建议不要使用缓存.**在数据库中和缓存数据的删除或者写入过程中，如果有失败的情况，会导致数据的不一致**.  
> 解决办法：  
> **双删延时的解决办法.可以先删除缓存数据，然后再更新数据库数据，最后再隔固定的时间再次删除缓存.**  
> 更新数据库产生的binlog订阅（使用canal）.将有变化的key记录下来，并且尝试去不断的去删除缓存（如果上次删除缓存失败）

缓存不一致详解：

**最初级的缓存不一致问题及解决方案**

问题：先修改数据库，再删除缓存.如果删除缓存失败了，那么会导致数据库中是新数据，缓存中是旧数据，数据就出现了不一致.

解决思路：先删除缓存，再修改数据库.如果数据库修改失败了，那么数据库中是旧数据，缓存中是空的，那么数据不会不一致.因为读的时候缓存没有，则读数据库中旧数据，然后更新到缓存中.

**比较复杂的数据不一致问题分析**

数据发生了变更，先删除了缓存，然后要去修改数据库，此时还没修改.一个请求过来，去读缓存，发现缓存空了，去查询数据库，**查到了修改前的旧数据**，放到了缓存中.随后数据变更的程序完成了数据库的修改.完了，数据库和缓存中的数据不一样了…

**为什么上亿流量高并发场景下，缓存会出现这个问题？**

只有在对一个数据在并发的进行读写的时候，才可能会出现这种问题.其实如果说你的并发量很低的话，特别是读并发很低，每天访问量就 1 万次，那么很少的情况下，会出现刚才描述的那种不一致的场景.但是问题是，如果每天的是上亿的流量，每秒并发读是几万，每秒只要有数据更新的请求，就**可能会出现上述的数据库+缓存不一致的情况**.

**解决方案如下：**

更新数据的时候，根据**数据的唯一标识**，将操作路由之后，发送到一个 jvm 内部队列中.读取数据的时候，如果发现数据不在缓存中，那么将重新读取数据+更新缓存的操作，根据唯一标识路由之后，也发送同一个 jvm 内部队列中.

一个队列对应一个工作线程，每个工作线程**串行**拿到对应的操作，然后一条一条的执行.这样的话，一个数据变更的操作，先删除缓存，然后再去更新数据库，但是还没完成更新.此时如果一个读请求过来，读到了空的缓存，那么可以先将缓存更新的请求发送到队列中，此时会在队列中积压，然后同步等待缓存更新完成.

这里有一个**优化点**，一个队列中，其实**多个更新缓存请求串在一起是没意义的**，因此可以做过滤，如果发现队列中已经有一个更新缓存的请求了，那么就不用再放个更新请求操作进去了，直接等待前面的更新操作请求完成即可.

待那个队列对应的工作线程完成了上一个操作的数据库的修改之后，才会去执行下一个操作，也就是缓存更新的操作，此时会从数据库中读取最新的值，然后写入缓存中.

如果请求还在等待时间范围内，不断轮询发现可以取到值了，那么就直接返回；如果请求等待的时间超过一定时长，那么这一次直接从数据库中读取当前的旧值.

高并发的场景下，该解决方案要注意的问题：

*  读请求长时阻塞

由于读请求进行了非常轻度的异步化，所以一定要注意读超时的问题，每个读请求必须在超时时间范围内返回.

该解决方案，最大的风险点在于说，**可能数据更新很频繁**，导致队列中积压了大量更新操作在里面，然后**读请求会发生大量的超时**，最后导致大量的请求直接走数据库.务必通过一些模拟真实的测试，看看更新数据的频率是怎样的.

另外一点，因为一个队列中，可能会积压针对多个数据项的更新操作，因此需要根据自己的业务情况进行测试，可能需要**部署多个服务**，每个服务分摊一些数据的更新操作.如果一个内存队列里居然会挤压 100 个商品的库存修改操作，每隔库存修改操作要耗费 10ms 去完成，那么最后一个商品的读请求，可能等待 10 \* 100 = 1000ms = 1s 后，才能得到数据，这个时候就导致**读请求的长时阻塞**.

一定要做根据实际业务系统的运行情况，去进行一些压力测试，和模拟线上环境，去看看最繁忙的时候，内存队列可能会挤压多少更新操作，可能会导致最后一个更新操作对应的读请求，会 hang 多少时间，如果读请求在 200ms 返回，如果你计算过后，哪怕是最繁忙的时候，积压 10 个更新操作，最多等待 200ms，那还可以的.

**如果一个内存队列中可能积压的更新操作特别多**，那么你就要**加机器**，让每个机器上部署的服务实例处理更少的数据，那么每个内存队列中积压的更新操作就会越少.

其实根据之前的项目经验，一般来说，数据的写频率是很低的，因此实际上正常来说，在队列中积压的更新操作应该是很少的.像这种针对读高并发、读缓存架构的项目，一般来说写请求是非常少的，每秒的 QPS 能到几百就不错了.

我们来**实际粗略测算一下**.

如果一秒有 500 的写操作，如果分成 5 个时间片，每 200ms 就 100 个写操作，放到 20 个内存队列中，每个内存队列，可能就积压 5 个写操作.每个写操作性能测试后，一般是在 20ms 左右就完成，那么针对每个内存队列的数据的读请求，也就最多 hang 一会儿，200ms 以内肯定能返回了.

经过刚才简单的测算，我们知道，单机支撑的写 QPS 在几百是没问题的，如果写 QPS 扩大了 10 倍，那么就扩容机器，扩容 10 倍的机器，每个机器 20 个队列.

*  读请求并发量过高

这里还必须做好压力测试，确保恰巧碰上上述情况的时候，还有一个风险，就是突然间大量读请求会在几十毫秒的延时 hang 在服务上，看服务能不能扛的住，需要多少机器才能扛住最大的极限情况的峰值.

但是因为并不是所有的数据都在同一时间更新，缓存也不会同一时间失效，所以每次可能也就是少数数据的缓存失效了，然后那些数据对应的读请求过来，并发量应该也不会特别大.

*  多服务实例部署的请求路由

可能这个服务部署了多个实例，那么必须**保证**说，执行数据更新操作，以及执行缓存更新操作的请求，都通过 Nginx 服务器**路由到相同的服务实例上**.

比如说，对同一个商品的读写请求，全部路由到同一台机器上.可以自己去做服务间的按照某个请求参数的 hash 路由，也可以用 Nginx 的 hash 路由功能等等.

*  热点商品的路由问题，导致请求的倾斜

万一某个商品的读写请求特别高，全部打到相同的机器的相同的队列里面去了，可能会造成某台机器的压力过大.就是说，因为只有在商品数据更新的时候才会清空缓存，然后才会导致读写并发，所以其实要根据业务系统去看，如果更新频率不是太高的话，这个问题的影响并不是特别大，但是的确可能某些机器的负载会高一些.

#### 5. Redis底层数据结构 ####

//https://stor.51cto.com/art/201910/605032.htm

主要提供了5种数据结构：**字符串**(String)、**哈希**(hash)、**列表**(list)、**集合**(set)、**有序集合**(short set);

**redis底层实现的8种数据结构**

SDS simple synamic string：支持自动动态扩容的字节数组 
    list ：链表 
    dict ：使用双哈希表实现的， 支持平滑扩容的字典 
    zskiplist ：附加了后向指针的跳跃表 
    intset ： 用于存储整数数值集合的自有结构 
    ziplist ：一种实现上类似于TLV， 但比TLV复杂的， 用于存储任意数据的有序序列的数据结构 
    quicklist：一种以ziplist作为结点的双链表结构， 实现的非常不错 
    zipmap ： 一种用于在小规模场合使用的轻量级字典结构

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8]

其中5种存储类型与8种数据结构的桥梁， **是redisObject**;

Redis中的Key与Value在表层都是一个redisObject实例， 所以该结构有所谓的"类型"， 即是ValueType. 对于每一种Value Type类型的redisObject;

其底层至少支持两种不同的底层数据结构来实现. 以应对在不同的应用场景中， Redis的运行效率， 或内存占用等

**底层数据结构分析**

**1、SDS - simple dynamic string**

可以在安装目录的src文件夹下看到sds.c和sds.h的源码文件

typedef char *sds; 
      
    /* Note: sdshdr5 is never used， we just access the flags byte directly. 
     * However is here to document the layout of type 5 SDS strings. */ 
    struct __attribute__ ((__packed__)) sdshdr5 { 
     unsigned char flags; /* 3 lsb of type， and 5 msb of string length */ 
     char buf[]; 
    }; 
    struct __attribute__ ((__packed__)) sdshdr8 { 
     uint8_t len; /* used */ 
     uint8_t alloc; /* excluding the header and null terminator */ 
     unsigned char flags; /* 3 lsb of type， 5 unused bits */ 
     char buf[]; 
    }; 
    struct __attribute__ ((__packed__)) sdshdr16 { 
     uint16_t len; /* used */ 
     uint16_t alloc; /* excluding the header and null terminator */ 
     unsigned char flags; /* 3 lsb of type， 5 unused bits */ 
     char buf[]; 
    }; 
    struct __attribute__ ((__packed__)) sdshdr32 { 
     uint32_t len; /* used */ 
     uint32_t alloc; /* excluding the header and null terminator */ 
     unsigned char flags; /* 3 lsb of type， 5 unused bits */ 
     char buf[]; 
    }; 
    struct __attribute__ ((__packed__)) sdshdr64 { 
     uint64_t len; /* used */ 
     uint64_t alloc; /* excluding the header and null terminator */ 
     unsigned char flags; /* 3 lsb of type， 5 unused bits */ 
     char buf[]; 
    };

sdshdr的存储结构

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FrS1rhmO-1596677444956)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200709224512039.png)\]

sdshdr是头部， buf是真实存储用户数据的地方.(buf=“数据” + “\\0” );sds有四种不同的头部. sdshdr5未 使用，未显示

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8 1]

en分别以uint8， uint16， uint32， uint64表示用户数据的长度(不包括末尾的\\0)

alloc分别以uint8， uint16， uint32， uint64表示整个SDS， 除过头部与末尾的\\0， 剩余的字节数.

flag始终为一字节， 以低三位标示着头部的类型， 高5位未使用.

创建一个SDS实例的三个接口

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8 2]

**2、list**

链表实现， 链表结点不直接持有数据， 而是通过void \*指针来间接的指向数据. 其实现位于 src/adlist.h与src/adlist.c中，

**内存布局**

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Jo1XFWTE-1596677444961)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200709224607207.png)\]

list在Redis除了作为一些Value Type的底层实现外， 还广泛用于Redis的其它功能实现中， 作为一种数据结构工具使用.

在list的实现中， 除了基本的链表定义外， 还额外增加了:迭代器listIter的定义， 与相关接口的实现.

由于list中的链表结点本身并不直接持有数据， 而是通过value字段， 以void \*指针的形式间接持有， 所以数据的生命周期并不完全与链表及其结点一致. 这给了list的使用者相当大的灵活性. 比如可以多个结点持有同一份数据的地址. 但与此同时， 在对链表进行销毁， 结点复制以及查找匹配时， 就需要list的使用者将相关的函数指针赋值于list.dup， list.free， list.match字段.

**3、dict**

dict类似于C++标准库中的std::unordered\_map， 其实现位于 src/dict.h 与 src/dict.c中

dict中存储的键值对， 是通过dictEntry这个结构间接持有的， k通过指针间接持有键， v通过指针间接持有值. 注意， 若值是整数值的话， 是直接存储在v字段中的， 而不是间接持有. 同时next指针用于指向， 在bucket索引值冲突时， 以链式方式解决冲突， 指向同索引的下一个dictEntry结构.

dict即为字典. 其中type字段中存储的是本字典使用到的各种函数指针， 包括散列函数， 键与值的复制函数， 释放函数， 以及键的比较函数. privdata是用于存储用户自定义数据. 这样， 字典的使用者可以最大化的自定义字典的实现， 通过自定义各种函数实现， 以及可以附带私有数据， 保证了字典有很大的调优空间.

字典为了支持平滑扩容， 定义了ht\[2\]这个数组字段. 其用意是这样的:

一般情况下， 字典dict仅持有一个哈希表dictht的实例， 即整个字典由一个bucket实现.

随着插入操作， bucket中出现冲突的概率会越来越大， 当字典中存储的结点数目， 与bucket数组长度的比值达到一个阈值(1:1)时， 字典为了缓解性能下降， 就需要扩容

扩容的操作是平滑的， 即在扩容时， 字典会持有两个dictht的实例， ht\[0\]指向旧哈希表， ht\[1\]指向扩容后的新哈希表.

**内存布局**

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8 3]

**4、zskiplist**

zskiplist是Redis实现的一种特殊的跳跃表. 跳跃表是一种基于线性表实现简单的搜索结构， 其最大的特点就是: 实现简单， 性能能逼近各种搜索树结构.

**zskiplist的核心设计要点:**

头结点不持有任何数据， 且其level\[\]的长度为32

每个结点， 除了持有数据的ele字段， 还有一个字段score， 其标示着结点的得分， 结点之间凭借得分来判断先后顺序， 跳跃表中的结点按结点的得分升序排列.

每个结点持有一个backward指针， 这是原版跳跃表中所没有的. 该指针指向结点的前一个紧邻结点.

每个结点中最多持有32个zskiplistLevel结构. 实际数量在结点创建时， 按幂次定律随机生成(不超过32). 每个zskiplistLevel中有两个字段.

**内存布局**

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8 4]

**5、intset**

用于存储在序的整数的数据结构， 也底层数据结构中最简单的一个， 其定义与实现在src/intest.h与src/intset.c中

inset结构中的encoding的取值有三个， 分别是宏INTSET\_ENC\_INT16， INTSET\_ENC\_INT32， INTSET\_ENC\_INT64. length代表其中存储的整数的个数， contents指向实际存储数值的连续内存区域

**内存布局**

![åŽŸåˆ›ï¼šredisä¸8ç§æ•°æ®ç»“æž„çš„åº•å±‚æ•°æ®ç»“æž„æºç è¯¦è§£][redis_8 3]

intset中各字段， 包括contents中存储的数值， 都是以主机序(小端字节序)存储的. 这意味着Redis若运行在PPC这样的大端字节序的机器上时， 存取数据都会有额外的字节序转换开销

当encoding == INTSET\_ENC\_INT16时， contents中以int16\_t的形式存储着数值. 类似的， 当encoding == INTSET\_ENC\_INT32时， contents中以int32\_t的形式存储着数值.

但凡有一个数值元素的值超过了int32\_t的取值范围， 整个intset都要进行升级， 即所有的数值都需要以int64\_t的形式存储. 显然升级的开销是很大的.

intset中的数值是以升序排列存储的， 插入与删除的复杂度均为O(n). 查找使用二分法， 复杂度为O(log\_2(n))

intset的代码实现中， 不预留空间， 即每一次插入操作都会调用zrealloc接口重新分配内存. 每一次删除也会调用zrealloc接口缩减占用的内存. 省是省了， 但内存操作的时间开销上升了.

intset的编码方式一经升级， 不会再降级.

总之， intset适合于如下数据的存储:

所有数据都位于一个稳定的取值范围中. 比如均位于int16\_t或int32\_t的取值范围中

数据稳定， 插入删除操作不频繁. 能接受O(lgn)级别的查找开销

**6、ziplist**

ziplist是Redis底层数据结构中， 最苟的一个结构. 它的设计宗旨就是: 省内存， 从牙缝里省内存. 设计思路和TLV一致， 但为了从牙缝里节省内存， 做了很多额外工作.

ziplist的内存布局与intset一样: 就是一块连续的内存空间. 但区域划分比较复杂

![åŽŸåˆ›ï¼šredisä¸8ç§æ•°æ®ç»“æž„çš„åº•å±‚æ•°æ®ç»“æž„æºç è¯¦è§£][redis_8 4]

和intset一样， ziplist中的所有值都是以小端序存储的

zlbytes字段的类型是uint32\_t， 这个字段中存储的是整个ziplist所占用的内存的字节数

zltail字段的类型是uint32\_t， 它指的是ziplist中最后一个entry的偏移量. 用于快速定位最后一个entry， 以快速完成pop等操作

zllen字段的类型是uint16\_t， 它指的是整个ziplit中entry的数量. 这个值只占16位， 所以蛋疼的地方就来了: 如果ziplist中entry的数目小于65535， 那么该字段中存储的就是实际entry的值. 若等于或超过65535， 那么该字段的值固定为65535， 但实际数量需要一个个entry的去遍历所有entry才能得到.

zlend是一个终止字节， 其值为全F， 即0xff. ziplist保证任何情况下， 一个entry的首字节都不会是255

在画图展示entry的内存布局之前， 先讲一下entry中都存储了哪些信息:

每个entry中存储了它前一个entry所占用的字节数. 这样支持ziplist反向遍历.

每个entry用单独的一块区域， 存储着当前结点的类型: 所谓的类型， 包括当前结点存储的数据是什么(二进制， 还是数值)， 如何编码(如果是数值， 数值如何存储， 如果是二进制数据， 二进制数据的长度)最后就是真实的数据了

**7、quicklist**

如果说ziplist是整个Redis中为了节省内存， 而写的最苟的数据结构， 那么称quicklist就是在最苟的基础上， 再苟了一层. 这个结构是Redis在3.2版本后新加的， 在3.2版本之前， 我们可以讲， dict是最复杂的底层数据结构， ziplist是最苟的底层数据结构. 在3.2版本之后， 这两个记录被双双刷新了.

这是一种， 以ziplist为结点的， 双端链表结构. 宏观上， quicklist是一个链表， 微观上， 链表中的每个结点都是一个ziplist.

它的定义与实现分别在src/quicklist.h与src/quicklist.c中

**这里定义了五个结构体:**

quicklistNode， 宏观上， quicklist是一个链表， 这个结构描述的就是链表中的结点. 它通过zl字段持有底层的ziplist. 简单来讲， 它描述了一个ziplist实例

quicklistLZF， ziplist是一段连续的内存， 用LZ4算法压缩后， 就可以包装成一个quicklistLZF结构. 是否压缩quicklist中的每个ziplist实例是一个可配置项. 若这个配置项是开启的， 那么quicklistNode.zl字段指向的就不是一个ziplist实例， 而是一个压缩后的quicklistLZF实例

quicklist. 这就是一个双链表的定义. head， tail分别指向头尾指针. len代表链表中的结点. count指的是整个quicklist中的所有ziplist中的entry的数目. fill字段影响着每个链表结点中ziplist的最大占用空间， compress影响着是否要对每个ziplist以LZ4算法进行进一步压缩以更节省内存空间.

**quicklistIter是一个迭代器**

quicklistEntry是对ziplist中的entry概念的封装. quicklist作为一个封装良好的数据结构， 不希望使用者感知到其内部的实现， 所以需要把ziplist.entry的概念重新包装一下.

quicklist的内存布局图如下所示:

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8 5]

下面是有关quicklist的更多额外信息:

quicklist.fill的值影响着每个链表结点中， ziplist的长度.

当数值为负数时， 代表以字节数限制单个ziplist的最大长度. 具体为:

\-1 不超过4kb

\-2 不超过 8kb

\-3 不超过 16kb

\-4 不超过 32kb

\-5 不超过 64kb

当数值为正数时， 代表以entry数目限制单个ziplist的长度. 值即为数目. 由于该字段仅占16位， 所以以entry数目限制ziplist的容量时， 最大值为2^15个

quicklist.compress的值影响着quicklistNode.zl字段指向的是原生的ziplist， 还是经过压缩包装后的quicklistLZF

0 表示不压缩， zl字段直接指向ziplist

1 表示quicklist的链表头尾结点不压缩， 其余结点的zl字段指向的是经过压缩后的quicklistLZF

2 表示quicklist的链表头两个， 与末两个结点不压缩， 其余结点的zl字段指向的是经过压缩后的quicklistLZF

以此类推， 最大值为2^16

quicklistNode.encoding字段， 以指示本链表结点所持有的ziplist是否经过了压缩. 1代表未压缩， 持有的是原生的ziplist， 2代表压缩过

quicklistNode.container字段指示的是每个链表结点所持有的数据类型是什么. 默认的实现是ziplist， 对应的该字段的值是2， 目前Redis没有提供其它实现. 所以实际上， 该字段的值恒为2

quicklistNode.recompress字段指示的是当前结点所持有的ziplist是否经过了解压. 如果该字段为1即代表之前被解压过， 且需要在下一次操作时重新压缩.

quicklist的具体实现代码篇幅很长， 这里就不贴代码片断了， 从内存布局上也能看出来， 由于每个结点持有的ziplist是有上限长度的， 所以在与操作时要考虑的分支情况比较多. 想想都蛋疼.

quicklist有自己的优点， 也有缺点， 对于使用者来说， 其使用体验类似于线性数据结构， list作为最传统的双链表， 结点通过指针持有数据， 指针字段会耗费大量内存. ziplist解决了耗费内存这个问题. 但引入了新的问题: 每次写操作整个ziplist的内存都需要重分配. quicklist在两者之间做了一个平衡. 并且使用者可以通过自定义quicklist.fill， 根据实际业务情况， 经验主义调参.

**8、zipmap**

dict作为字典结构， 优点很多， 扩展性强悍， 支持平滑扩容等等， 但对于字典中的键值均为二进制数据， 且长度都很小时， dict的中的一坨指针会浪费不少内存， 因此Redis又实现了一个轻量级的字典， 即为zipmap.

zipmap适合使用的场合是:

键值对量不大， 单个键， 单个值长度小

键值均是二进制数据， 而不是复合结构或复杂结构. dict支持各种嵌套， 字典本身并不持有数据， 而仅持有数据的指针. 但zipmap是直接持有数据的.

zipmap的定义与实现在src/zipmap.h与src/zipmap.c两个文件中， 其定义与实现均未定义任何struct结构体， 因为zipmap的内存布局就是一块连续的内存空间. 其内存布局如下所示:

![原创：redis中8种数据结构的底层数据结构源码详解][redis_8 6]

zipmap起始的第一个字节存储的是zipmap中键值对的个数. 如果键值对的个数大于254的话， 那么这个字节的值就是固定值254， 真实的键值对个数需要遍历才能获得.

**zipmap的最后一个字节是固定值0xFF**

zipmap中的每一个键值对， 称为一个entry， 其内存占用如上图， 分别六部分:

len\_of\_key， 一字节或五字节. 存储的是键的二进制长度. 如果长度小于254， 则用1字节存储， 否则用五个字节存储， 第一个字节的值固定为0xFE， 后四个字节以小端序uint32\_t类型存储着键的二进制长度.

**key\_data为键的数据**

len\_of\_val， 一字节或五字节， 存储的是值的二进制长度. 编码方式同len\_of\_key

len\_of\_free， 固定值1字节， 存储的是entry中未使用的空间的字节数. 未使用的空间即为图中的free， 它一般是由于键值对中的值被替换发生的. 比如， 键值对hello <-> word被修改为hello <-> w后， 就空了四个字节的闲置空间

**val\_data， 为值的数据**

free， 为闲置空间. 由于len\_of\_free的值最大只能是254， 所以如果值的变更导致闲置空间大于254的话， zipmap就会回收内存空间.

#### 6. Redis的过期策略 ####

Redis keys过期有两种方式：定期扫描和惰性删除.

###### \[1\] 定期删除（主动） ######

**定时删除，用一个定时器来负责监视所有key，当key过期则自动删除key**.  
虽然内存及时释放，但是十分消耗CPU资源.在大并发请求下，会影响redis的性能.

###### \[2\]惰性删除（被动） ######

**客户端尝试访问key时，被访问的key会被发现并主动的过期**

缺陷：有些过期的keys，永远不会访问他们，那么他们就永远不会被删除，而占用内存.

###### \[3\] 定期删除+惰性删除 ######

\*\*redis默认每100ms主动检查一次，如果有过期的key则删除，当满足1/4的keys过期，则重复之前步骤.\*\*具体步骤为：

1.  redis默认每100ms随机抽取20个keys进行过期检查；
2.  同时删除已经过期的keys，如果有多于25%的keys过期，则重复抽取.
3.  直到过期的keys的百分比低于25%，
4.  这意味着，在任何给定的时刻，最多会清除1/4的过期keys.

###### \[4\] 内存淘汰机制 ######

采用定期删除+惰性删除就能保证过期的key会全部删除掉么?

如果定期删除没删除key.然后也没去请求key，也就是说惰性删除也没生效.这样，redis的内存会越来越高.那么就应该采用内存淘汰机制.  
在redis.conf中有配置

> maxmemory-policy volatile-lru

内存淘汰策略如下:

*  noeviction：当内存不足以容纳新写入数据时，新写入操作会报错，不建议使用
 *  allkeys-lru：当内存不足以容纳新写入数据时，在键空间中，移除最近最少使用的key.推荐使用
 *  allkeys-random：当内存不足以容纳新写入数据时，在键空间中，随机移除某个key
 *  volatile-lru：当内存不足以容纳新写入数据时，在设置了过期时间的键空间中，移除最近最少使用的key
 *  volatile-random：当内存不足以容纳新写入数据时，在设置了过期时间的键空间中，随机移除某个key
 *  volatile-ttl：当内存不足以容纳新写入数据时，在设置了过期时间的键空间中，有更早过期时间的key优先移除

###### \[5\] LRU算法 ######

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-OfcRLkDx-1596677444965)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200719222247853.png)\]

#### 7. Redis分布式锁 ####

> https://www.jianshu.com/p/47fd7f86c848

在Java中，关于锁我想大家都很熟悉.在并发编程中，我们通过锁，来避免由于竞争而造成的数据不一致问题.通常，我们以`synchronized 、Lock`来使用它.但是Java中的锁，只能保证在同一个JVM进程内中执行.如果在分布式集群环境下呢？

###### \[1\] 分布式锁 ######

分布式锁，是一种思想，它的实现方式有很多.比如，我们将沙滩当做分布式锁的组件，那么它看起来应该是这样的：

*  **加锁**：在沙滩上踩一脚，留下自己的脚印，就对应了加锁操作.其他进程或者线程，看到沙滩上已经有脚印，证明锁已被别人持有，则等待.
 *  **解锁**：把脚印从沙滩上抹去，就是解锁的过程.
 *  **锁超时**：为了避免死锁，我们可以设置一阵风，在单位时间后刮起，将脚印自动抹去.

分布式锁的实现有很多，比如基于数据库、memcached、Redis、系统文件、zookeeper等.它们的核心的理念跟上面的过程大致相同.

###### \[2\] redis ######

我们先来看如何**通过单节点Redis实现一个简单的分布式锁**.

**1、加锁**

加锁实际上就是在redis中，给Key键设置一个值，为避免死锁，并给定一个过期时间.

SET lock_key random_value NX PX 5000

值得注意的是：  
`random_value` 是客户端生成的唯一的字符串.  
`NX` 代表只在键不存在时，才对键进行设置操作.  
`PX 5000` 设置键的过期时间为5000毫秒.

这样，如果上面的命令执行成功，则证明客户端获取到了锁.

**2、解锁**

解锁的过程就是将Key键删除.但也不能乱删，不能说客户端1的请求将客户端2的锁给删除掉.这时候`random_value`的作用就体现出来.

为了保证解锁操作的原子性，我们用LUA脚本完成这一操作.先判断当前锁的字符串是否与传入的值相等，是的话就删除Key，解锁成功.

if redis.call('get'，KEYS[1]) == ARGV[1] then 
       return redis.call('del'，KEYS[1]) 
    else
       return 0 
    end

**3、实现**

首先，我们在pom文件中，引入Jedis.在这里，笔者用的是最新版本，注意由于版本的不同，API可能有所差异.

<dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>3.0.1</version>
    </dependency>

加锁的过程很简单，就是通过SET指令来设置值，成功则返回；否则就循环等待，在timeout时间内仍未获取到锁，则获取失败.

@Service
    public class RedisLock { 
    
        Logger logger = LoggerFactory.getLogger(this.getClass());
    
        private String lock_key = "redis_lock"; //锁键
    
        protected long internalLockLeaseTime = 30000;//锁过期时间
    
        private long timeout = 999999; //获取锁的超时时间
    
        
        //SET命令的参数 
        SetParams params = SetParams.setParams().nx().px(internalLockLeaseTime);
    
        @Autowired
        JedisPool jedisPool;
    
        
        /** * 加锁 * @param id * @return */
        public boolean lock(String id){ 
            Jedis jedis = jedisPool.getResource();
            Long start = System.currentTimeMillis();
            try{ 
                for(;;){ 
                    //SET命令返回OK ，则证明获取锁成功
                    String lock = jedis.set(lock_key， id， params);
                    if("OK".equals(lock)){ 
                        return true;
                    }
                    //否则循环等待，在timeout时间内仍未获取到锁，则获取失败
                    long l = System.currentTimeMillis() - start;
                    if (l>=timeout) { 
                        return false;
                    }
                    try { 
                        Thread.sleep(100);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }finally { 
                jedis.close();
            }
        }
    }

解锁我们通过`jedis.eval`来执行一段LUA就可以.将锁的Key键和生成的字符串当做参数传进来.

/**
         * 解锁
         * @param id
         * @return
         */
        public boolean unlock(String id){
            Jedis jedis = jedisPool.getResource();
            String script =
                    "if redis.call('get'，KEYS[1]) == ARGV[1] then" +
                            "   return redis.call('del'，KEYS[1]) " +
                            "else" +
                            "   return 0 " +
                            "end";
            try {
                Object result = jedis.eval(script， Collections.singletonList(lock_key)， 
                                        Collections.singletonList(id));
                if("1".equals(result.toString())){
                    return true;
                }
                return false;
            }finally {
                jedis.close();
            }
        }

最后，我们可以在多线程环境下测试一下.我们开启1000个线程，对count进行累加.调用的时候，关键是唯一字符串的生成.这里，笔者使用的是`Snowflake`算法.

@Controller
    public class IndexController { 
    
        @Autowired
        RedisLock redisLock;
        
        int count = 0;
        
        @RequestMapping("/index")
        @ResponseBody
        public String index() throws InterruptedException { 
    
            int clientcount =1000;
            CountDownLatch countDownLatch = new CountDownLatch(clientcount);
    
            ExecutorService executorService = Executors.newFixedThreadPool(clientcount);
            long start = System.currentTimeMillis();
            for (int i = 0;i<clientcount;i++){ 
                executorService.execute(() -> { 
                
                    //通过Snowflake算法获取唯一的ID字符串
                    String id = IdUtil.getId();
                    try { 
                        redisLock.lock(id);
                        count++;
                    }finally { 
                        redisLock.unlock(id);
                    }
                    countDownLatch.countDown();
                });
            }
            countDownLatch.await();
            long end = System.currentTimeMillis();
            logger.info("执行线程数:{}，总耗时:{}，count数为:{}"，clientcount，end-start，count);
            return "Hello";
        }
    }

至此，单节点Redis的分布式锁的实现就已经完成了.比较简单，但是问题也比较大，最重要的一点是，锁不具有可重入性.

###### \[3\] redisson ######

> [Redisson][]是架设在[Redis][]基础上的一个Java驻内存数据网格（In-Memory Data Grid）.充分的利用了Redis键值数据库提供的一系列优势，基于Java实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类.使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度.同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作.

相对于Jedis而言，Redisson强大的一批.当然了，随之而来的就是它的复杂性.它里面也实现了分布式锁，而且包含多种类型的锁，更多请参阅[分布式锁和同步器][Link 1]

**1、可重入锁**

上面我们自己实现的Redis分布式锁，其实不具有可重入性.那么下面我们先来看看Redisson中如何调用可重入锁.

在这里，笔者使用的是它的最新版本，3.10.1.

<dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson</artifactId>
        <version>3.10.1</version>
    </dependency>

首先，通过配置获取RedissonClient客户端的实例，然后`getLock`获取锁的实例，进行操作即可.

public static void main(String[] args) { 
    
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379");
        config.useSingleServer().setPassword("redis1234");
        
        final RedissonClient client = Redisson.create(config);  
        RLock lock = client.getLock("lock1");
        
        try{ 
            lock.lock();
        }finally{ 
            lock.unlock();
        }
    }

**2、获取锁实例**

我们先来看`RLock lock = client.getLock("lock1");` 这句代码就是为了获取锁的实例，然后我们可以看到它返回的是一个`RedissonLock`对象.

public RLock getLock(String name) { 
        return new RedissonLock(connectionManager.getCommandExecutor()， name);
    }

在`RedissonLock`构造方法中，主要初始化一些属性

public RedissonLock(CommandAsyncExecutor commandExecutor， String name) { 
        super(commandExecutor， name);
        //命令执行器
        this.commandExecutor = commandExecutor;
        //UUID字符串
        this.id = commandExecutor.getConnectionManager().getId();
        //内部锁过期时间
        this.internalLockLeaseTime = commandExecutor.
                    getConnectionManager().getCfg().getLockWatchdogTimeout();
        this.entryName = id + ":" + name;
    }

**3、加锁**

当我们调用`lock`方法，定位到`lockInterruptibly`.在这里，完成了加锁的逻辑.

public void lockInterruptibly(long leaseTime， TimeUnit unit) throws InterruptedException { 
        
        //当前线程ID
        long threadId = Thread.currentThread().getId();
        //尝试获取锁
        Long ttl = tryAcquire(leaseTime， unit， threadId);
        // 如果ttl为空，则证明获取锁成功
        if (ttl == null) { 
            return;
        }
        //如果获取锁失败，则订阅到对应这个锁的channel
        RFuture<RedissonLockEntry> future = subscribe(threadId);
        commandExecutor.syncSubscription(future);
    
        try { 
            while (true) { 
                //再次尝试获取锁
                ttl = tryAcquire(leaseTime， unit， threadId);
                //ttl为空，说明成功获取锁，返回
                if (ttl == null) { 
                    break;
                }
                //ttl大于0 则等待ttl时间后继续尝试获取
                if (ttl >= 0) { 
                    getEntry(threadId).getLatch().tryAcquire(ttl， TimeUnit.MILLISECONDS);
                } else { 
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally { 
            //取消对channel的订阅
            unsubscribe(future， threadId);
        }
        //get(lockAsync(leaseTime， unit));
    }

如上代码，就是加锁的全过程.先调用`tryAcquire`来获取锁，如果返回值ttl为空，则证明加锁成功，返回；如果不为空，则证明加锁失败.这时候，它会订阅这个锁的Channel，等待锁释放的消息，然后重新尝试获取锁.流程如下：

![img][]

**获取锁**

获取锁的过程是怎样的呢？接下来就要看`tryAcquire`方法.在这里，它有两种处理方式，一种是带有过期时间的锁，一种是不带过期时间的锁.

private <T> RFuture<Long> tryAcquireAsync(long leaseTime， TimeUnit unit， final long threadId) { 
    
        //如果带有过期时间，则按照普通方式获取锁
        if (leaseTime != -1) { 
            return tryLockInnerAsync(leaseTime， unit， threadId， RedisCommands.EVAL_LONG);
        }
        
        //先按照30秒的过期时间来执行获取锁的方法
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(
            commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()，
            TimeUnit.MILLISECONDS， threadId， RedisCommands.EVAL_LONG);
            
        //如果还持有这个锁，则开启定时任务不断刷新该锁的过期时间
        ttlRemainingFuture.addListener(new FutureListener<Long>() { 
            @Override
            public void operationComplete(Future<Long> future) throws Exception { 
                if (!future.isSuccess()) { 
                    return;
                }
    
                Long ttlRemaining = future.getNow();
                // lock acquired
                if (ttlRemaining == null) { 
                    scheduleExpirationRenewal(threadId);
                }
            }
        });
        return ttlRemainingFuture;
    }

接着往下看，`tryLockInnerAsync`方法是真正执行获取锁的逻辑，它是一段LUA脚本代码.在这里，它使用的是hash数据结构.

<T> RFuture<T> tryLockInnerAsync(long leaseTime， TimeUnit unit，     
                                long threadId， RedisStrictCommand<T> command) { 
    
            //过期时间
            internalLockLeaseTime = unit.toMillis(leaseTime);
    
            return commandExecutor.evalWriteAsync(getName()， LongCodec.INSTANCE， command，
                      //如果锁不存在，则通过hset设置它的值，并设置过期时间
                      "if (redis.call('exists'， KEYS[1]) == 0) then " +
                          "redis.call('hset'， KEYS[1]， ARGV[2]， 1); " +
                          "redis.call('pexpire'， KEYS[1]， ARGV[1]); " +
                          "return nil; " +
                      "end; " +
                      //如果锁已存在，并且锁的是当前线程，则通过hincrby给数值递增1
                      "if (redis.call('hexists'， KEYS[1]， ARGV[2]) == 1) then " +
                          "redis.call('hincrby'， KEYS[1]， ARGV[2]， 1); " +
                          "redis.call('pexpire'， KEYS[1]， ARGV[1]); " +
                          "return nil; " +
                      "end; " +
                      //如果锁已存在，但并非本线程，则返回过期时间ttl
                      "return redis.call('pttl'， KEYS[1]);"，
            Collections.<Object>singletonList(getName())， 
                    internalLockLeaseTime， getLockName(threadId));
        }

这段LUA代码看起来并不复杂，有三个判断：

*  **通过exists判断，如果锁不存在，则设置值和过期时间，加锁成功**
 *  **通过hexists判断，如果锁已存在，并且锁的是当前线程，则证明是重入锁，加锁成功**
 *  **如果锁已存在，但锁的不是当前线程，则证明有其他线程持有锁.返回当前锁的过期时间，加锁失败**

![img][img 1]

加锁成功后，在redis的内存数据中，就有一条hash结构的数据.Key为锁的名称；field为随机字符串+线程ID；值为1.如果同一线程多次调用`lock`方法，值递增1.

127.0.0.1:6379> hgetall lock1
    1) "b5ae0be4-5623-45a5-8faa-ab7eb167ce87:1"
    2) "1"

**解锁**

我们通过调用`unlock`方法来解锁.

public RFuture<Void> unlockAsync(final long threadId) { 
        final RPromise<Void> result = new RedissonPromise<Void>();
        
        //解锁方法
        RFuture<Boolean> future = unlockInnerAsync(threadId);
    
        future.addListener(new FutureListener<Boolean>() { 
            @Override
            public void operationComplete(Future<Boolean> future) throws Exception { 
                if (!future.isSuccess()) { 
                    cancelExpirationRenewal(threadId);
                    result.tryFailure(future.cause());
                    return;
                }
                //获取返回值
                Boolean opStatus = future.getNow();
                //如果返回空，则证明解锁的线程和当前锁不是同一个线程，抛出异常
                if (opStatus == null) { 
                    IllegalMonitorStateException cause = 
                        new IllegalMonitorStateException(" attempt to unlock lock， not locked by current thread by node id: "
                            + id + " thread-id: " + threadId);
                    result.tryFailure(cause);
                    return;
                }
                //解锁成功，取消刷新过期时间的那个定时任务
                if (opStatus) { 
                    cancelExpirationRenewal(null);
                }
                result.trySuccess(null);
            }
        });
    
        return result;
    }

然后我们再看`unlockInnerAsync`方法.这里也是一段LUA脚本代码.

protected RFuture<Boolean> unlockInnerAsync(long threadId) { 
        return commandExecutor.evalWriteAsync(getName()， LongCodec.INSTANCE， EVAL，
        
                //如果锁已经不存在， 发布锁释放的消息
                "if (redis.call('exists'， KEYS[1]) == 0) then " +
                    "redis.call('publish'， KEYS[2]， ARGV[1]); " +
                    "return 1; " +
                "end;" +
                //如果释放锁的线程和已存在锁的线程不是同一个线程，返回null
                "if (redis.call('hexists'， KEYS[1]， ARGV[3]) == 0) then " +
                    "return nil;" +
                "end; " +
                //通过hincrby递减1的方式，释放一次锁
                //若剩余次数大于0 ，则刷新过期时间
                "local counter = redis.call('hincrby'， KEYS[1]， ARGV[3]， -1); " +
                "if (counter > 0) then " +
                    "redis.call('pexpire'， KEYS[1]， ARGV[2]); " +
                    "return 0; " +
                //否则证明锁已经释放，删除key并发布锁释放的消息
                "else " +
                    "redis.call('del'， KEYS[1]); " +
                    "redis.call('publish'， KEYS[2]， ARGV[1]); " +
                    "return 1; "+
                "end; " +
                "return nil;"，
        Arrays.<Object>asList(getName()， getChannelName())， 
            LockPubSub.unlockMessage， internalLockLeaseTime， getLockName(threadId));
    
    }

如上代码，就是释放锁的逻辑.同样的，它也是有三个判断：

*  **如果锁已经不存在，通过publish发布锁释放的消息，解锁成功**
 *  **如果解锁的线程和当前锁的线程不是同一个，解锁失败，抛出异常**
 *  **通过hincrby递减1，先释放一次锁.若剩余次数还大于0，则证明当前锁是重入锁，刷新过期时间；若剩余次数小于0，删除key并发布锁释放的消息，解锁成功**

![img][img 2]

至此，Redisson中的可重入锁的逻辑，就分析完了.但值得注意的是，上面的两种实现方式都是针对单机Redis实例而进行的.如果我们有多个Redis实例，请参阅**Redlock算法**.该算法的具体内容，请参考http://redis.cn/topics/distlock.html

https://blog.csdn.net/weixin\_43258908/article/details/89199088

https://blog.csdn.net/liubenlong007/article/details/53690312?utm\_medium=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.compare&depth\_1-utm\_source=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.compare

https://blog.csdn.net/ai\_goodStudent/article/details/86520018

*  redis 服务器设置了一个 AOF 重写缓冲区.这个缓冲区在创建子进程后开始使用，当redis服务器执行一个客户端的写请求命令，之后将这个写命令也发送到 AOF 重写缓冲区.
 *  当子进程完成 AOF 日志重写之后，给父进程发送信号，父进程接收此信号后，将 AOF 重写缓冲区的内容写到新的 AOF 文件中，保持数据的一致性.

###### \[4\] 牛客网总结 ######

![image-20200629222006147][image-20200629221647159] ![image-20200629222115786][image-20200629221647159] ![image-20200629222136887][image-20200629221647159] ![image-20200629222204906][image-20200629221647159]

###### \[5\] 常用的分布式锁 ######

https://www.jb51.net/article/184718.html

https://www.cnblogs.com/wlwl/p/11651409.html

https://www.cnblogs.com/owenma/p/12355262.html

**一、基于数据库实现分布式锁**

**1. 悲观锁**

利用select … where … for update 排他锁

注意: 其他附加功能与实现一基本一致，这里需要注意的是“where name=lock ”，name字段必须要走索引，否则会锁表.有些情况下，比如表不大，mysql优化器会不走这个索引，导致锁表问题.

**2. 乐观锁**

所谓乐观锁与前边最大区别在于基于CAS思想，是不具有互斥性，不会产生锁等待而消耗资源，操作过程中认为不存在并发冲突，只有update version失败后才能觉察到.我们的抢购、秒杀就是用了这种实现以防止超卖.  
通过增加递增的版本号字段实现乐观锁

**二、基于jdk的实现方式**

**思路**：另启一个服务，利用jdk并发工具来控制唯一资源，如在服务中维护一个concurrentHashMap，其他服务对某个key请求锁时，通过该服务暴露的端口，以网络通信的方式发送消息，服务端解析这个消息，将concurrentHashMap中的key对应值设为true，分布式锁请求成功，可以采用基于netty通信调用，当然你想用java的bio、nio或者整合dubbo、spring cloud feign来实现通信也没问题  
**缺点**：这种方式的分布式锁看似简单，但是要考虑可用性、可靠性、效率、扩展性的话，编码难度会比较高.

**三、基于缓存（Redis等）实现分布式锁**

1、官方叫做 RedLock 算法，是 redis 官方支持的分布式锁算法.

这个分布式锁有 3 个重要的考量点：

*  1.互斥（只能有一个客户端获取锁）
 *  2.不能死锁
 *  3.容错（只要大部分 redis 节点创建了这把锁就可以）

2、下面是redis分布式锁的各种实现方式和缺点，按照时间的发展排序

*  1、直接setnx  
    直接利用setnx，执行完业务逻辑后调用del释放锁，简单粗暴  
    **缺点**：如果setnx成功，还没来得及释放，服务挂了，那么这个key永远都不会被获取到
 *  2、setnx设置一个过期时间  
    为了改正第一个方法的缺陷，我们用setnx获取锁，然后用expire对其设置一个过期时间，如果服务挂了，过期时间一到自动释放  
    **缺点**：setnx和expire是两个方法，不能保证原子性，如果在setnx之后，还没来得及expire，服务挂了，还是会出现锁不释放的问题
 *  3、set nx px  
    redis官方为了解决第二种方式存在的缺点，在2.8版本为set指令添加了扩展参数nx和ex，保证了setnx+expire的原子性，使用方法：  
    `set key value ex 5 nx`  
    **缺点**：  
    ①如果在过期时间内，事务还没有执行完，锁提前被自动释放，其他的线程还是可以拿到锁  
    ②上面所说的那个缺点还会导致当前的线程释放其他线程占有的锁
 *  4、加一个事务id  
    上面所说的第一个缺点，没有特别好的解决方法，只能把过期时间尽量设置的长一点，并且最好不要执行耗时任务  
    第二个缺点，可以理解为当前线程有可能会释放其他线程的锁，那么问题就转换为保证线程只能释放当前线程持有的锁，即setnx的时候将value设为任务的唯一id，释放的时候先get key比较一下value是否与当前的id相同，是则释放，否则抛异常回滚，其实也是变相地解决了第一个问题  
    **缺点**：get key和将value与id比较是两个步骤，不能保证原子性
 *  5、set nx px + 事务id + lua  
    我们可以用lua来写一个getkey并比较的脚本，jedis/luttce/redisson对lua脚本都有很好的支持  
    **缺点**：集群环境下，对master节点申请了分布式锁，由于redis的主从同步是异步进行的，master在内存中写入了nx之后直接返回，客户端获取锁成功，此时master节点挂了，并且数据还没来得及同步，另一个节点被升级为master，这样其他的线程依然可以获取锁
 *  6、redlock  
    为了解决上面提到的redis集群中的分布式锁问题，redis的作者antirez的提出了red lock的概念，假设集群中所有的n个master节点完全独立，并且没有主从同步，此时对所有的节点都去setnx，并且设置一个请求过期时间re和锁的过期时间le，同时re必须小于le（可以理解，不然请求3秒才拿到锁，而锁的过期时间只有1秒也太蠢了），此时如果有n / 2 + 1个节点成功拿到锁，此次分布式锁就算申请成功  
    **缺点**：可靠性还没有被广泛验证，并且严重依赖时间，好的分布式系统应该是异步的，并不能以时间为担保，程序暂停、系统延迟等都可能会导致时间错误（网上还有很多人都对这个方法提出了质疑，比如full gc发生的锁的正确性问题，但是antirez都一一作出了解答，感兴趣的同学可以参考一下[这位同学的文章][Link 2]）

**四、基于zookeeper实现的分布式锁**

1.  实现方式

ZooKeeper是一个为分布式应用提供一致性服务的开源组件，它内部是一个分层的文件系统目录树结构，规定同一个目录下只能有一个唯一文件名.基于ZooKeeper实现分布式锁的步骤如下：

（1）创建一个目录mylock；  
（2）线程A想获取锁就在mylock目录下创建临时顺序节点；  
（3）获取mylock目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁；  
（4）线程B获取所有节点，判断自己不是最小节点，设置监听比自己次小的节点；  
（5）线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是不是最小的节点，如果是则获得锁.

这里推荐一个Apache的开源库Curator，它是一个ZooKeeper客户端，Curator提供的InterProcessMutex是分布式锁的实现，acquire方法用于获取锁，release方法用于释放锁.

**优点**：具备高可用、可重入、阻塞锁特性，可解决失效死锁问题.

**缺点**：因为需要频繁的创建和删除节点，性能上不如Redis方式.

1.  两种利用特性实现原理：

*  **1、利用临时节点特性**  
    zookeeper的临时节点有两个特性，一是节点名称不能重复，二是会随着客户端退出而销毁，因此直接将key作为节点名称，能够成功创建的客户端则获取成功，失败的客户端监听成功的节点的删除事件  
    **缺点**：所有客户端监听同一个节点，但是同时只有一个节点的事件触发是有效的，造成资源的无效调度
 *  **2、利用顺序临时节点特性**  
    zookeeper的顺序临时节点拥有临时节点的特性，同时，在一个父节点下创建创建的子临时顺序节点，会根据节点创建的先后顺序，用一个32位的数字作为后缀，我们可以用key创建一个根节点，然后每次申请锁的时候在其下创建顺序节点，接着获取根节点下所有的顺序节点并排序，获取顺序最小的节点，如果该节点的名称与当前添加的名称相同，则表示能够获取锁，否则监听根节点下面的处于当前节点之前的节点的删除事件，如果监听生效，则回到上一步重新判断顺序，直到获取锁.

###### \[6\] 分布式锁三种实现方式 ######

\\1. 基于数据库实现分布式锁；

\\2. 基于缓存（Redis等）实现分布式锁；

\\3. 基于Zookeeper实现分布式锁；

**一， 基于数据库实现分布式锁**

\\1. 悲观锁

利用select … where … for update 排他锁

\\2. 乐观锁

所谓乐观锁与前边最大区别在于基于CAS思想，是不具有互斥性，不会产生锁等待而消耗资源，操作过程中认为不存在并发冲突，只有update version失败后才能觉察到.我们的抢购、秒杀就是用了这种实现以防止超卖.

通过增加递增的版本号字段实现乐观锁

**二， 基于缓存（Redis等）实现分布式锁**

\\1. 使用命令介绍：

（1）SETNX

SETNX key val：当且仅当key不存在时，set一个key为val的字符串，返回1；若key存在，则什么都不做，返回0.

（2）expire

expire key timeout：为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁.

（3）delete

delete key：删除key

在使用Redis实现分布式锁的时候，主要就会使用到这三个命令.

\\2. 实现思想：

（1）获取锁的时候，使用setnx加锁，并使用expire命令为锁添加一个超时时间，超过该时间则自动释放锁，锁的value值为一个随机生成的UUID，通过此在释放锁的时候进行判断.

（2）获取锁的时候还设置一个获取的超时时间，若超过这个时间则放弃获取锁.

（3）释放锁的时候，通过UUID判断是不是该锁，若是该锁，则执行delete进行锁释放.

**三， 基于Zookeeper实现分布式锁**

> （1）创建一个目录mylock；  
> （2）线程A想获取锁就在mylock目录下创建临时顺序节点；  
> （3）获取mylock目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁；  
> （4）线程B获取所有节点，判断自己不是最小节点，设置监听比自己次小的节点；  
> （5）线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是不是最小的节点，如果是则获得锁.

这里推荐一个Apache的开源库Curator，它是一个ZooKeeper客户端，Curator提供的InterProcessMutex是分布式锁的实现，acquire方法用于获取锁，release方法用于释放锁.

优点：具备高可用、可重入、阻塞锁特性，可解决失效死锁问题.

缺点：因为需要频繁的创建和删除节点，性能上不如Redis方式.

**四，对比**

数据库分布式锁实现

缺点：

1.db操作性能较差，并且有锁表的风险

2.非阻塞操作失败后，需要轮询，占用cpu资源;

3.长时间不commit或者长时间轮询，可能会占用较多连接资源

**Redis(缓存)分布式锁实现**

缺点：

1.锁删除失败 过期时间不好控制

2.非阻塞，操作失败后，需要轮询，占用cpu资源;

**ZK分布式锁实现**

缺点：性能不如redis实现，主要原因是写操作（获取锁释放锁）都需要在Leader上执行，然后同步到follower.

总之：ZooKeeper有较好的性能和可靠性.

从理解的难易程度角度（从低到高）数据库 > 缓存 > Zookeeper

从实现的复杂性角度（从低到高）Zookeeper >= 缓存 > 数据库

从性能角度（从高到低）缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）Zookeeper > 缓存 > 数据库

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家.

总结

**基于数据库分布式锁实现**

**优点**：直接使用数据库，实现方式简单.  
**缺点**：

1.  db操作性能较差，并且有锁表的风险
2.  非阻塞操作失败后，需要轮询，占用cpu资源;
3.  长时间不commit或者长时间轮询，可能会占用较多连接资源

**基于jdk的并发工具自己实现的锁**

**优点**：不需要引入中间件，架构简单  
**缺点**：编写一个可靠、高可用、高效率的分布式锁服务，难度较大

**基于redis缓存**

\\1. redis set px nx + 唯一id + lua脚本  
**优点**：redis本身的读写性能很高，因此基于redis的分布式锁效率比较高  
**缺点**：依赖中间件，分布式环境下可能会有节点数据同步问题，可靠性有一定的影响，如果发生则需要人工介入

\\2. 基于redis的redlock  
**优点**：可以解决redis集群的同步可用性问题  
**缺点**：

1.  依赖中间件，并没有被广泛验证，维护成本高，需要多个独立的master节点；需要同时对多个节点申请锁，降低了一些效率
2.  锁删除失败 过期时间不好控制
3.  非阻塞，操作失败后，需要轮询，占用cpu资源;

基于zookeeper的分布式锁

**优点**：不存在redis的超时、数据同步（zookeeper是同步完以后才返回）、主从切换（zookeeper主从切换的过程中服务是不可用的）的问题，可靠性很高

**缺点**：依赖中间件，保证了可靠性的同时牺牲了一部分效率（但是依然很高）.性能不如redis.

jdk的方式不太推荐.

1.  从理解的难易程度角度（从低到高）数据库 > 缓存 > Zookeeper
2.  从实现的复杂性角度（从低到高）Zookeeper >= 缓存 > 数据库
3.  从性能角度（从高到低）缓存 > Zookeeper >= 数据库
4.  从可靠性角度（从高到低）Zookeeper > 缓存 > 数据库

没有绝对完美的实现方式，具体要选择哪一种分布式锁，需要结合每一种锁的优缺点和业务特点而定

#### 8. Redis的I/O多路复用 ####

https://www.jianshu.com/p/311f9d276b2a

###### \[1\] 为什么 Redis 中要使用 I/O 多路复用？ ######

首先，Redis 是跑在单线程中的，所有的操作都是按照顺序线性执行的，但是由于读写操作等待用户输入或输出都是阻塞的，所以 I/O 操作在一般情况下往往不能直接返回，这会导致某一文件的 I/O 阻塞导致整个进程无法对其它客户提供服务，而 I/O 多路复用就是为了解决这个问题而出现的.

**阻塞I/O：**

先来看一下传统的阻塞 I/O 模型到底是如何工作的：当使用 read 或者 write 对某一个文件描述符（File Descriptor 以下简称 FD)进行读写时，如果当前 FD 不可读或不可写，整个 Redis 服务就不会对其它的操作作出响应，导致整个服务不可用.

![img][img 3]

###### \[2\] I/O 多路复用模型 ######

https://blog.csdn.net/wuyangyang555/article/details/82146831

https://blog.csdn.net/wsx199397/article/details/38533239?utm\_medium=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.compare&depth\_1-utm\_source=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.compare

阻塞式的 I/O 模型并不能满足这里的需求，我们需要一种效率更高的 I/O 模型来支撑 Redis 的多个客户（redis-cli），这里涉及的就是 I/O 多路复用模型了：

**在 I/O 多路复用模型中，最重要的函数调用就是 select，该方法的能够同时监控多个文件描述符的可读可写情况，当其中的某些文件描述符可读或者可写时，select 方法就会返回可读以及可写的文件描述符个数**.

> 与此同时也有其它的 I/O 多路复用函数 epoll/kqueue/evport，它们相比 select 性能更优秀，同时也能支撑更多的服务.
> 
> **文件描述符**：linux下，所有的操作都是对文件进行操作，而对文件的操作是利用[文件描述符][Link 3](file descriptor)来实现的.\*\*每个文件进程控制块中都有一份文件描述符表（可以把它看成是一个数组，里面的元素是指向file结构体指针类型），这个数组的下标就是文件描述符.\*\*在[源代码][Link 4]中，一般用fd作为[文件描述符][Link 3]的标识.
> 
> **套接字**：套接字是一种通信机制，凭借这种机制，客户/服务器系统的开发工作既可以在本地单机上进行，也可以跨网络进行，Linux所提供的功能（如打印服 务，ftp等）通常都是通过套接字来进行通信的，套接字的创建和使用与管道是有区别的，因为套接字明确地将客户和服务器区分出来，套接字可以实现将多个客 户连接到一个服务器.

![img][img 4]

在UNIX系统上，一切皆文件套接字也不例外，每一个套接字都有对应的fd（即文件描述符）我们简单看看这几个系统调用的原型.

###### \[3\] select ######

\*\*select(int nfds， fd\_set \*r， fd\_set \*w，fd\_set *e， struct timeval timeout)*

对于select()，我们需要传3个集合，r（读），w（写）和e其中，r表示我们对哪些fd的可读事件感兴趣，w表示我们对哪些fd的可写事件感兴趣**每个集合其实是一个bitmap，通过0/1表示我们感兴趣的fd**.例如，

如：我们对于fd为6的可读事件感兴趣，那么r集合的第6个bit需要被设置为1这个系统调用会阻塞，直到我们感兴趣的事件（至少一个）发生调用返回时，内核同样使用这3个集合来存放fd实际发生的事件信息也就是说，调用前这3个集合表示我们感兴趣的事件，调用后这3个集合表示实际发生的事件.

select为最早期的UNIX系统调用，它存在4个问题：

1）**这3个bitmap有大小限制**（FD\_SETSIZE，通常为1024）；

2）**由于这3个集合在返回时会被内核修改，因此我们每次调用时都需要重新设置**；

3）**我们在调用完成后需要扫描这3个集合才能知道哪些fd的读/写事件发生了，一般情况下全量集合比较大而实际发生读/写事件的fd比较少，效率比较低下；**

4）**内核在每次调用都需要扫描这3个fd集合，然后查看哪些fd的事件实际发生，在读/写比较稀疏的情况下同样存在效率问题.**

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-gxhKvu5M-1596677444968)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200720092539779.png)\]

###### \[4\] poll ######

由于存在这些问题，于是人们对select进行了改进，从而有了poll

poll(struct pollfd *fds， int nfds， inttimeout)
    
    struct pollfd { int fd;short events;short revents;}

poll本质上和select没有区别，它将用户传入的数组拷贝到内核空间，然后查询每个fd对应的设备状态， 但是它没有最大连接数的限制，原因是它是基于链表来存储的.

poll调用需要传递的是一个**pollfd结构的数组**，调用返回时结果信息也存放在这个数组里面pollfd的结构中存放着fd我们对该fd感兴趣的事件(events)以及该fd实际发生的事件(revents)**，poll传递的不是固定大小的bitmap，因此select的问题1解决了**；**poll将感兴趣事件和实际发生事件分开了，因此select的问题2也解决了**但select的问题3和问题4仍然没有解决.

select问题3比较容易解决，只要系统调用返回的是实际发生相应事件的fd集合，我们便不需要扫描全量的fd集合.对于select的问题4，我们为什么需要每次调用都传递全量的fd呢？内核可不可以在第一次调用的时候记录这些fd，然后我们在以后的调用中不需要再传这些fd呢？问题的关键在于**无状态**对于每一次系统调用，内核不会记录下任何信息，所以每次调用都需要重复传递相同信息.

###### \[5\] epoll ######

上帝说要有状态，所以我们有了**epoll**和kqueue

int epoll_create(int size);
    
    int epoll_ctl(int epfd， int op， int fd，struct epoll_event *event);
    
    int epoll_wait(int epfd， struct epoll_event*events， int maxevents， int timeout);

**epoll\_create**的作用是创建一个context，这个context相当于状态保存者的概念

**epoll\_ctl**的作用是，当你对一个新的fd的读/写事件感兴趣时，通过该调用将fd与相应的感兴趣事件更新到context中

**epoll\_wait**的作用是，等待context中fd的事件发生

**epoll的解决方案不像select或poll一样每次都把current（现时发生的）轮流加入fd对应的设备等待队列中，而只在epoll\_ctl时把current挂一遍（这一遍必不可少）并为每个fd指定一个回调函数，当设备就绪，唤醒等待队列上的等待者时，就会调用这个回调函数，而这个回调函数会把就绪的fd加入一个就绪链表**）.epoll\_wait的工作实际上就是在这个就绪链表中查看有没有就绪的fd.

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QmXGO7Gs-1596677444969)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200720094515533.png)\]

**epoll的两种工作方式：1.水平触发（LT）2.边缘触发（ET）**

*  **LT模式**：若就绪的事件一次没有处理完要做的事件，就会一直去处理.即就会将没有处理完的事件继续放回到就绪队列之中（即那个内核中的链表），一直进行处理.
 *  **ET模式**：就绪的事件只能处理一次，若没有处理完会在下次的其它事件就绪时再进行处理.而若以后再也没有就绪的事件，那么剩余的那部分数据也会随之而丢失.  
    由此可见：ET模式的效率比LT模式的效率要高很多.只是如果使用ET模式，就要保证每次进行数据处理时，要将其处理完，不能造成数据丢失，这样对编写代码的人要求就比较高.  
    注意：ET模式只支持非阻塞的读写：为了保证数据的完整性.

###### \[6\] Reactor 设计模式 ######

Redis 服务采用 Reactor 的方式来实现文件事件处理器（每一个网络连接其实都对应一个文件描述符）

![img][img 5]

![img][img 6]

**文件事件处理器使用 I/O 多路复用模块同时监听多个 FD**，当 accept、read、write 和 close 文件事件产生时，文件事件处理器就会回调 FD 绑定的事件处理器.

虽然整个文件事件处理器是在单线程上运行的，但是通过 I/O 多路复用模块的引入，实现了同时对多个 FD 读写的监控，提高了网络通信模型的性能，同时也可以保证整个 Redis 服务实现的简单.

**I/O 多路复用模块**

I/O 多路复用模块封装了底层的 select、epoll、avport 以及 kqueue 这些 I/O 多路复用函数，为上层提供了相同的接口.

![img][img 7]

在这里我们简单介绍 Redis 是如何包装 select 和 epoll 的，简要了解该模块的功能，整个 I/O 多路复用模块抹平了不同平台上 I/O 多路复用函数的差异性，提供了相同的接口：

*  static int aeApiCreate(aeEventLoop \*eventLoop)
 *  static int aeApiResize(aeEventLoop \*eventLoop， int setsize)
 *  static void aeApiFree(aeEventLoop \*eventLoop)
 *  static int aeApiAddEvent(aeEventLoop \*eventLoop， int fd， int mask)
 *  static void aeApiDelEvent(aeEventLoop \*eventLoop， int fd， int mask)
 *  static int aeApiPoll(aeEventLoop \*eventLoop， struct timeval \*tvp)

同时，因为各个函数所需要的参数不同，我们在每一个子模块内部通过一个 aeApiState 来存储需要的上下文信息：

// select
    typedef struct aeApiState { 
        fd_set rfds， wfds;
        fd_set _rfds， _wfds;
    } aeApiState;
    
    // epoll
    typedef struct aeApiState { 
        int epfd;
        struct epoll_event *events;
    } aeApiState;

这些上下文信息会存储在 eventLoop 的 void \*state 中，不会暴露到上层，只在当前子模块中使用.

封装 select 函数

> select 可以监控 FD 的可读、可写以及出现错误的情况.

在介绍 I/O 多路复用模块如何对 select 函数封装之前，先来看一下 select 函数使用的大致流程：

int fd = /* file descriptor */
    
    fd_set rfds;
    FD_ZERO(&rfds);
    FD_SET(fd， &rfds)
    
    for ( ; ; ) { 
        select(fd+1， &rfds， NULL， NULL， NULL);
        if (FD_ISSET(fd， &rfds)) { 
            /* file descriptor `fd` becomes readable */
        }
    }

1.  初始化一个可读的 fd\_set 集合，保存需要监控可读性的 FD；
2.  使用 FD\_SET 将 fd 加入 rfds；
3.  调用 select 方法监控 rfds 中的 FD 是否可读；
4.  当 select 返回时，检查 FD 的状态并完成对应的操作.

而在 Redis 的 ae\_select 文件中代码的组织顺序也是差不多的，首先在 aeApiCreate 函数中初始化 rfds 和 wfds：

static int aeApiCreate(aeEventLoop *eventLoop) { 
        aeApiState *state = zmalloc(sizeof(aeApiState));
        if (!state) return -1;
        FD_ZERO(&state->rfds);
        FD_ZERO(&state->wfds);
        eventLoop->apidata = state;
        return 0;
    }

而 aeApiAddEvent 和 aeApiDelEvent 会通过 FD\_SET 和 FD\_CLR 修改 fd\_set 中对应 FD 的标志位：

static int aeApiAddEvent(aeEventLoop *eventLoop， int fd， int mask) { 
        aeApiState *state = eventLoop->apidata;
        if (mask & AE_READABLE) FD_SET(fd，&state->rfds);
        if (mask & AE_WRITABLE) FD_SET(fd，&state->wfds);
        return 0;
    }

整个 ae\_select 子模块中最重要的函数就是 aeApiPoll，它是实际调用 select 函数的部分，其作用就是在 I/O 多路复用函数返回时，将对应的 FD 加入 aeEventLoop 的 fired 数组中，并返回事件的个数：

static int aeApiPoll(aeEventLoop *eventLoop， struct timeval *tvp) { 
        aeApiState *state = eventLoop->apidata;
        int retval， j， numevents = 0;
    
        memcpy(&state->_rfds，&state->rfds，sizeof(fd_set));
        memcpy(&state->_wfds，&state->wfds，sizeof(fd_set));
    
        retval = select(eventLoop->maxfd+1，
                    &state->_rfds，&state->_wfds，NULL，tvp);
        if (retval > 0) { 
            for (j = 0; j <= eventLoop->maxfd; j++) { 
                int mask = 0;
                aeFileEvent *fe = &eventLoop->events[j];
    
                if (fe->mask == AE_NONE) continue;
                if (fe->mask & AE_READABLE && FD_ISSET(j，&state->_rfds))
                    mask |= AE_READABLE;
                if (fe->mask & AE_WRITABLE && FD_ISSET(j，&state->_wfds))
                    mask |= AE_WRITABLE;
                eventLoop->fired[numevents].fd = j;
                eventLoop->fired[numevents].mask = mask;
                numevents++;
            }
        }
        return numevents;
    }

###### \[7\] 封装 epoll 函数 ######

Redis 对 epoll 的封装其实也是类似的，使用 epoll\_create 创建 epoll 中使用的 epfd：

static int aeApiCreate(aeEventLoop *eventLoop) { 
        aeApiState *state = zmalloc(sizeof(aeApiState));
    
        if (!state) return -1;
        state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
        if (!state->events) { 
            zfree(state);
            return -1;
        }
        state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
        if (state->epfd == -1) { 
            zfree(state->events);
            zfree(state);
            return -1;
        }
        eventLoop->apidata = state;
        return 0;
    }

在 aeApiAddEvent 中使用 epoll\_ctl 向 epfd 中添加需要监控的 FD 以及监听的事件：

static int aeApiAddEvent(aeEventLoop *eventLoop， int fd， int mask) { 
        aeApiState *state = eventLoop->apidata;
        struct epoll_event ee = { 0}; /* avoid valgrind warning */
        /* If the fd was already monitored for some event， we need a MOD * operation. Otherwise we need an ADD operation. */
        int op = eventLoop->events[fd].mask == AE_NONE ?
                EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    
        ee.events = 0;
        mask |= eventLoop->events[fd].mask; /* Merge old events */
        if (mask & AE_READABLE) ee.events |= EPOLLIN;
        if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
        ee.data.fd = fd;
        if (epoll_ctl(state->epfd，op，fd，&ee) == -1) return -1;
        return 0;
    }

由于 epoll 相比 select 机制略有不同，**在 epoll\_wait 函数返回时并不需要遍历所有的 FD 查看读写情况；在 epoll\_wait 函数返回时会提供一个 epoll\_event 数组**：

typedef union epoll_data { 
        void    *ptr;
        int      fd; /* 文件描述符 */
        uint32_t u32;
        uint64_t u64;} epoll_data_t;
    
    struct epoll_event { 
        uint32_t     events; /* Epoll 事件 */
        epoll_data_t data;
    };

> 其中保存了发生的 epoll 事件（EPOLLIN、EPOLLOUT、EPOLLERR 和 EPOLLHUP）以及发生该事件的 FD.

aeApiPoll 函数只需要将 epoll\_event 数组中存储的信息加入 eventLoop 的 fired 数组中，将信息传递给上层模块：

static int aeApiPoll(aeEventLoop *eventLoop， struct timeval *tvp) { 
        aeApiState *state = eventLoop->apidata;
        int retval， numevents = 0;
    
        retval = epoll_wait(state->epfd，state->events，eventLoop->setsize，
                tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
        if (retval > 0) { 
            int j;
    
            numevents = retval;
            for (j = 0; j < numevents; j++) { 
                int mask = 0;
                struct epoll_event *e = state->events+j;
    
                if (e->events & EPOLLIN) mask |= AE_READABLE;
                if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
                if (e->events & EPOLLERR) mask |= AE_WRITABLE;
                if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
                eventLoop->fired[j].fd = e->data.fd;
                eventLoop->fired[j].mask = mask;
            }
        }
        return numevents;
    }

子模块的选择

因为 Redis 需要在多个平台上运行，同时为了最大化执行的效率与性能，所以会根据编译平台的不同选择不同的 I/O 多路复用函数作为子模块，提供给上层统一的接口；在 Redis 中，我们通过宏定义的使用，合理的选择不同的子模块：

#ifdef HAVE_EVPORT#include "ae_evport.c"#else
        #ifdef HAVE_EPOLL
        #include "ae_epoll.c"
        #else
            #ifdef HAVE_KQUEUE
            #include "ae_kqueue.c"
            #else
            #include "ae_select.c"
            #endif
        #endif
    #endif

因为 select 函数是作为 POSIX 标准中的系统调用，在不同版本的操作系统上都会实现，所以将其作为保底方案：

![img][img 8]

Redis 会优先选择时间复杂度为 O(1)的 I/O 多路复用函数作为底层实现，包括 Solaries 10 中的 evport、Linux 中的 epoll 和 macOS/FreeBSD 中的 kqueue，上述的这些函数都使用了内核内部的结构，并且能够服务几十万的文件描述符.

但是如果当前编译环境没有上述函数，就会选择 select 作为备选方案，由于其在使用时会扫描全部监听的描述符，所以其时间复杂度较差 O(n)，并且只能同时服务 1024 个文件描述符，所以一般并不会以 select 作为第一方案使用.

###### \[8\] select & poll & epoll比较 ######

https://www.jianshu.com/p/c8f462827499

https://www.bilibili.com/video/BV1qJ411w7du

**select**： 单个进程所能打开的最大连接数有FD\_SETSIZE宏定义， 其大小为1024或者2048； FD数目剧增后， 会带来性能问题；消息传递从内核到与到用户空间，需要copy数据；  
（1）每次调用select，都需要把fd集合从用户态拷贝到内核态，这个开销在fd很多时会很大  
（2）同时每次调用select都需要在内核遍历传递进来的所有fd，这个开销在fd很多时也很大

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0rhdEQ7B-1596677444973)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200711143703434.png)\]

**poll**： 基本上与select一样， 不同点在于没有FD数目的限制， 因为底层实现不是一个数组， 而是链表；

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-s8E6yE3Y-1596677444973)(X:\\Users\\xu\\AppData\\Roaming\\Typora\\typora-user-images\\image-20200711144411018.png)\]

**epoll**： FD连接数虽然有限制， 但是很大几乎可以认为无限制；epoll内核中实现是根据每个fd上的callback函数来实现的，只有活跃的socket才会主动调用callback，所以在活跃socket较少的情况下，使用epoll没有前面两者的线性下降的性能问题； 内核和用户通过共享内存来传递消息；

> **什么是socket？**
> 
> 我们都知道unix(like)世界里，一切皆文件，而文件是什么呢？文件就是一二进制流而已，不管socket，还是FIFO、管道、终端，对我们来说，一切都是文件，一切都是流.在信息 交换的过程中，我们都是对这些流进行数据的收发操作，简称为I/O操作(input and output)，往流中读出数据，系统调用read，写入数据，系统调用write.不过话说回来了 ，计算机里有这么多的流，我怎么知道要操作哪个流呢？对，就是**文件描述符**，即通常所说的fd，一个fd就是一个整数，所以，对这个整数的操作，就是对这个文件（流）的操作.我们创建一个socket，通过系统调用会返回一个文件描述符，那么剩下对socket的操作就会转化为对这个描述符的操作.不能不说这又是一种分层和抽象的思想.
> 
> > *socket*一般指套接字.所谓套接字(*Socket*)，就是对网络中不同主机上的应用进程之间进行双向通信的端点的抽象.

linux提供了select、poll、epoll接口来实现IO复用，三者的原型如下所示，本文从参数、实现、性能等方面对三者进行对比.

int select(int nfds， fd_set *readfds， fd_set *writefds， fd_set *exceptfds， struct timeval *timeout);
    int poll(struct pollfd *fds， nfds_t nfds， int timeout);
    int epoll_wait(int epfd， struct epoll_event *events， int maxevents， int timeou);

**select、poll、epoll\_wait参数及实现对比**

1.  **int nfds**：select的第一个参数nfds为fdset集合中最大描述符值加1，fdset是一个位数组，其大小限制为\_\_FD\_SETSIZE（1024），位数组的每一位代表其对应的描述符是否需要被检查.
    
    \*\*fd\_set \*readfds， fd\_set \*writefds， fd\_set \*exceptfds：\*\*select的第二三四个参数表示需要关注读、写、错误事件的文件描述符位数组，这些参数既是输入参数也是输出参数，可能会被内核修改用于标示哪些描述符上发生了关注的事件.所以每次调用select前都需要重新初始化fdset.
    
    \**struct timeval timeout*：timeout参数为超时时间，该结构会被内核修改，其值为超时剩余的时间.
    
    select对应于内核中的sys\_select调用，sys\_select首先将第二三四个参数指向的fd\_set拷贝到内核，然后对每个被SET的描述符调用进行poll，并记录在临时结果中（fdset），如果有事件发生，select会将临时结果写到用户空间并返回；当轮询一遍后没有任何事件发生时，如果指定了超时时间，则select会睡眠到超时，睡眠结束后再进行一次轮询，并将临时结果写到用户空间，然后返回.
    
    **select返回后**，需要逐一检查关注的描述符是否被SET（事件是否发生）.
2.  \*\*struct pollfd \*fds：\*\*poll与select不同，通过一个pollfd数组向内核传递需要关注的事件，故没有描述符个数的限制，pollfd中的events字段和revents分别用于标示关注的事件和发生的事件，故pollfd数组只需要被初始化一次.
    
    poll的实现机制与select类似，其对应内核中的sys\_poll，只不过poll向内核传递pollfd数组，然后对pollfd中的每个描述符进行poll，相比处理fdset来说，poll效率更高.poll返回后，需要对pollfd中的每个元素检查其revents值，来得指事件是否发生.
3.  epoll通过epoll\_create创建一个用于epoll轮询的描述符，通过epoll\_ctl添加/修改/删除事件，通过epoll\_wait检查事件，epoll\_wait的第二个参数用于存放结果.
    
    epoll与select、poll不同，首先，其不用每次调用都向内核拷贝事件描述信息，在第一次调用后，事件信息就会与对应的epoll描述符关联起来.**另外epoll不是通过轮询，而是通过在等待的描述符上注册回调函数，当事件发生时，回调函数负责把发生的事件存储在就绪事件链表中，最后写到用户空间.**
    
    epoll返回后，该参数指向的缓冲区中即为发生的事件，对缓冲区中每个元素进行处理即可，而不需要像poll、select那样进行轮询检查.
    
    **select、poll、epoll\_wait性能对比**  
    select、poll的内部实现机制相似，性能差别主要在于向内核传递参数以及对fdset的位操作上，另外，select存在描述符数的硬限制，不能处理很大的描述符集合.这里主要考察poll与epoll在不同大小描述符集合的情况  
    原文链接：https://blog.csdn.net/leafrenchleaf/article/details/84159301

###### \[9\] 深入理解select、poll和epoll及区别：杂文 ######

https://blog.csdn.net/wteruiycbqqvwt/article/details/90299610

https://blog.csdn.net/nanxiaotao/article/details/90612404?utm\_medium=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.compare&depth\_1-utm\_source=distribute.pc\_relevant.none-task-blog-BlogCommendFromMachineLearnPai2-1.compare

原文链接：https://blog.csdn.net/nanxiaotao/article/details/90612404

#### 9. 多线程的 Redis ####

[link][]

相信redis6.0以前一直都是单线程，到了6的版本才加入了多线程.

**一、问题概述**

Redis 6.0 之后的版本抛弃了单线程模型这一设计，原本使用单线程运行的 Redis 也开始选择性使用多线程模型，乍一看Redis的作者这么牛，也逃不过“真香定律”，

这个问题其实可以拆分，拆分为两个主要的问题：

（1）为什么 Redis 一开始选择单线程模型（单线程的好处）？

（2）为什么 Redis 在 6.0 之后加入了多线程（在某些情况下，单线程出现了缺点，多线程可以解决）？

随着时间的推移，单线程出现的问题也越来越多，原来的设计肯定就有些不合时宜，该做出改变就做出改变.技术并不是一成不变的.

**二、为什么Redis一开始使用单线程**

不管是单线程或者是多线程都是为了提升Redis的开发效率，因为Redis是一个基于内存的数据库，还要处理大量的外部的网络请求，这就不可避免的要进行多次IO.好在Redis使用了很多优秀的机制来保证了它的高效率.那么为什么Redis要设计成单线程模式的呢？可以总结如下：

（1）**IO多路复用**

我们来看一下Redis顶层设计.

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-avriBcJ9-1596677444973)(https://pics7.baidu.com/feed/dcc451da81cb39dba486f96646c28e22aa183092.jpeg?token=b7e96b5d677dac17de14489f884c271b)\]

FD是一个文件描述符，意思是表示当前文件处于可读、可写还是异常状态.使用 I/O 多路复用机制同时监听多个文件描述符的可读和可写状态.**你可以理解为具有了多线程的特点.**

一旦受到网络请求就会在内存中快速处理，由于绝大多数的操作都是纯内存的，所以处理的速度会非常地快.**也就是说在单线程模式下，即使连接的网络处理很多，因为有IO多路复用，依然可以在高速的内存处理中得到忽略.**

（2）**可维护性高**

多线程模型虽然在某些方面表现优异，但是它却引入了程序执行顺序的不确定性，带来了并发读写的一系列问题.单线程模式下，可以方便地进行调试和测试.

（3）**基于内存，单线程状态下效率依然高**

多线程能够充分利用CPU的资源，但对于Redis来说，由于基于内存速度那是相当的高，能达到在一秒内处理10万个用户请求，如果一秒十万还不能满足，那我们就可以使用Redis分片的技术来交给不同的Redis服务器.这样的做饭避免了在同一个 Redis 服务中引入大量的多线程操作.

而且基于内存，除非是要进行AOF备份，否则基本上不会涉及任何的 I/O 操作.这些数据的读写由于只发生在内存中，所以处理速度是非常快的；用多线程模型处理全部的外部请求可能不是一个好的方案.

**现在我们知道了基本上可以总结成两句话，基于内存而且使用多路复用技术，单线程速度很快，又保证了多线程的特点.因为没有必要使用多线程.**

**三、为什么引入多线程？**

刚刚说了一堆使用单线程的好处，现在话锋一转，又要说为什么要引入多线程，别不适应.引入多线程说明Redis在有些方面，单线程已经不具有优势了.

因为读写网络的read/write系统调用在Redis执行期间占用了大部分CPU时间，如果把网络读写做成多线程的方式对性能会有很大提升.

Redis 的多线程部分只是用来处理网络数据的读写和协议解析，执行命令仍然是单线程.之所以这么设计是不想 Redis 因为多线程而变得复杂，需要去控制 key、lua、事务，LPUSH/LPOP 等等的并发问题.

Redis 在最新的几个版本中加入了一些可以被其他线程异步处理的删除操作，也就是我们在上面提到的 UNLINK、FLUSHALL ASYNC 和 FLUSHDB ASYNC，我们为什么会需要这些删除操作，而它们为什么需要通过多线程的方式异步处理？

我们知道Redis可以使用del命令删除一个元素，如果这个元素非常大，可能占据了几十兆或者是几百兆，那么在短时间内是不能完成的，这样一来就需要多线程的异步支持.

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-ZlIeOVLo-1596677444974)(https://pics4.baidu.com/feed/bd3eb13533fa828bbd4ad2d869cbc632960a5afe.jpeg?token=c7bd1ec4f81b30ba2d0916d7191e765c)\]

现在删除工作可以在后台进行.

**四、总结**

Redis 选择使用单线程模型处理客户端的请求主要还是因为 CPU 不是 Redis 服务器的瓶颈，所以使用多线程模型带来的性能提升并不能抵消它带来的开发成本和维护成本，系统的性能瓶颈也主要在网络 I/O 操作上；而 Redis 引入多线程操作也是出于性能上的考虑，对于一些大键值对的删除操作，通过多线程非阻塞地释放内存空间也能减少对 Redis 主线程阻塞的时间，提高执行的效率.

**一句话讲完：之前用单线程是因为基于内存速度快，而且多路复用有多路复用的作用，也就是足够了，现在引入是因为在某些操作要优化，比如删除操作，因此引入了多线程.**

#### 10 Redis持久化方式 ####

> 已订正

**1. Redis 提供了不同级别的持久化方式:**

*  **RDB(Redis DataBase)持久化**：Redis 可以通过创建快照来存储内存里面的数据在某个时间点上的副本到本地数据库中，如果系统，Redis或者硬件其中之一崩溃，那么Redis将丢失最近一次创建快照之后写入的所有数据.
 *  **AOF(Append Only File)持久化**：AOF持久化会将被执行的写命令写到AOF文件的末尾，以此来记录数据发生的变化.它可以做到每秒一次的频率对AOF文件进行同步.
 *  **不持久化**：如果你只希望数据在服务器运行的时候存在，你也可以不使用任何持久化方式
 *  **RDB+AOF模式** ：在这种情况下， 当Redis重启的时候会优先载入AOF文件来恢复原始的数据，因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整.

**2. RDB的优点，和缺点**

**优点**：RDB是一个非常紧凑的文件，它保存了某个时间点的数据集，非常适用于数据集的备份，比如你可以在每个小时报保存一下过去24小时内的数据，同时每天保存过去30天的数据，这样即使出了问题你也可以根据需求恢复到不同版本的数据集.与AOF相比，**在恢复大的数据集的时候，RDB方式会更快一些**.

**缺点**：如果你希望在redis意外停止工作（例如电源中断）的情况下丢失的数据最少的话，那么RDB不适合你.**因为RDB更容易丢失数据**.虽然你可以配置不同的save时间点(例如每隔5分钟并且对数据集有100个写的操作)，是Redis要完整的保存整个数据集是一个比较繁重的工作，你通常会每隔5分钟或者更久做一次完整的保存，万一在Redis意外宕机，你可能会丢失几分钟的数据.RDB 需要经常fork子进程来保存数据集到硬盘上，当数据集比较大的时候，fork的过程是非常耗时的，可能会导致Redis在一些毫秒级内不能响应客户端的请求.

**3. AOF优点和缺点**

**优点**：使用AOF 会让你的Redis更加耐久: 你可以使用不同的fsync策略：无fsync，每秒fsync，每次写的时候fsync.使用默认的每秒fsync策略，Redis的性能依然很好(fsync是由后台线程进行处理的，主线程会尽力处理客户端请求)，**一旦出现故障，你最多丢失1秒的数据**.

**缺点**：对于相同的数据集来说，**AOF 文件的体积通常要大于 RDB 文件的体积**.根据所使用的 fsync 策略，**AOF 的速度可能会慢于 RDB .**

**4. 如何选择持久化方式**

**RDB 持久化恢复数据集的速度要比AOF快一点，但是容易丢失更多的数据.AOF可以在每秒进行一次记录，如果发生系统崩溃最多丢失1秒的数据，更加可靠.但是它AOF文件比较大，恢复速度要慢一点.**

如果你非常关心你的数据， 但仍然可以承受数分钟以内的数据丢失， 那么你可以只使用 RDB 持久化.有很多用户都只使用 AOF 持久化，但我们并不推荐这种方式： 因为定时生成 RDB 快照（snapshot）非常便于进行数据库备份， 并且 RDB 恢复数据集的速度也要比 AOF 恢复的速度要快一般来说，如果想达到足以媲美 PostgreSQL 的数据安全性， 你应该同时使用两种持久化功能.

> **1. 快照与备份有什么区别？[link][link 1]**
> 
> **一句话答案：快照记录逻辑地址和物理地址的对应关系；备份则是数据存储的某一个时刻的副本.这是两种完全不同的概念.**
> 
> **2. 内部存储与外部存储的区别：[link][link 2]**
> 
> **内部存储：**
> 
> 内部存储不是内存，而是一个位于系统中很特殊的一个位置.放入内部存储中的数据一般都只能被你的应用访问到，且一个应用所创建的所有文件都在应用包名相同的目录下，即/data/data/packagename.创建于内部存储的文件，是与这个应用关联起来的.当一个应用被卸载后，内部存储中的这些数据也被删除.
> 
> **外部存储：**
> 
> 最容易混淆的是外部存储，如果说pc上也要区分出外部存储和内部存储的话，那么自带的硬盘算是内部存储，U盘或者移动硬盘算是外部存储，因此我们很容易带着这样的理解去看待安卓手机，认为机身固有存储是内部存储，而扩展的T卡是外部存储.

#### 11. Redis哨兵（Sentinel）模式 ####

> 链接：https://www.jianshu.com/p/06ab9daf921d

**主从切换技术的方法是：当主服务器宕机后，需要手动把一台从服务器切换为主服务器，这就需要人工干预，费事费力，还会造成一段时间内服务不可用。这不是一种推荐的方式，更多时候，我们优先考虑哨兵模式**。

**一、哨兵模式概述**

哨兵模式是一种特殊的模式，首先Redis提供了哨兵的命令，哨兵是一个独立的进程。其原理是**哨兵通过发送命令，等待Redis服务器响应，从而监控运行的多个Redis实例。**

![img][img 9]

这里的哨兵有两个作用

*  通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器。
 *  当哨兵监测到master宕机，会自动将slave切换成master，然后通过**发布订阅模式**通知其他的从服务器，修改配置文件，让它们切换主机。

然而一个哨兵进程对Redis服务器进行监控，可能会出现问题，为此，我们可以使用多个哨兵进行监控。各个哨兵之间还会进行监控，这样就形成了多哨兵模式。

用文字描述一下**故障切换（failover）的过程。假设主服务器宕机，哨兵1先检测到这个结果，系统并不会马上进行failover过程，仅仅是哨兵1主观的认为主服务器不可用，这个现象成为主观下线**。当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵之间就会进行一次投票，投票的结果由一个哨兵发起，进行failover操作。切换成功后，就会通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程称为**客观下线**。这样对于客户端而言，一切都是透明的。

**二、Redis配置哨兵模式**

配置3个哨兵和1主2从的Redis服务器来演示这个过程。

<table> 
 <thead> 
  <tr> 
   <th>服务类型</th> 
   <th>是否是主服务器</th> 
   <th>IP地址</th> 
   <th>端口</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Redis</td> 
   <td>是</td> 
   <td>192.168.11.128</td> 
   <td>6379</td> 
  </tr> 
  <tr> 
   <td>Redis</td> 
   <td>否</td> 
   <td>192.168.11.129</td> 
   <td>6379</td> 
  </tr> 
  <tr> 
   <td>Redis</td> 
   <td>否</td> 
   <td>192.168.11.130</td> 
   <td>6379</td> 
  </tr> 
  <tr> 
   <td>Sentinel</td> 
   <td>-</td> 
   <td>192.168.11.128</td> 
   <td>26379</td> 
  </tr> 
  <tr> 
   <td>Sentinel</td> 
   <td>-</td> 
   <td>192.168.11.129</td> 
   <td>26379</td> 
  </tr> 
  <tr> 
   <td>Sentinel</td> 
   <td>-</td> 
   <td>192.168.11.130</td> 
   <td>26379</td> 
  </tr> 
 </tbody> 
</table>

![img][img 10]

多哨兵监控Redis

首先配置Redis的主从服务器，修改redis.conf文件如下

# 使得Redis服务器可以跨网络访问
    bind 0.0.0.0
    # 设置密码
    requirepass "123456"
    # 指定主服务器，注意：有关slaveof的配置只是配置从服务器，主服务器不需要配置
    slaveof 192.168.11.128 6379
    # 主服务器密码，注意：有关slaveof的配置只是配置从服务器，主服务器不需要配置
    masterauth 123456

上述内容主要是配置Redis服务器，从服务器比主服务器多一个slaveof的配置和密码。

配置3个哨兵，每个哨兵的配置都是一样的。在Redis安装目录下有一个sentinel.conf文件，copy一份进行修改

# 禁止保护模式
    protected-mode no
    # 配置监听的主服务器，这里sentinel monitor代表监控，mymaster代表服务器的名称，可以自定义，192.168.11.128代表监控的主服务器，6379代表端口，2代表只有两个或两个以上的哨兵认为主服务器不可用的时候，才会进行failover操作。
    sentinel monitor mymaster 192.168.11.128 6379 2
    # sentinel author-pass定义服务的密码，mymaster是服务名称，123456是Redis服务器密码
    # sentinel auth-pass <master-name> <password>
    sentinel auth-pass mymaster 123456

上述关闭了保护模式，便于测试。

有了上述的修改，我们可以进入Redis的安装目录的src目录，通过下面的命令启动服务器和哨兵

# 启动Redis服务器进程
    ./redis-server ../redis.conf
    # 启动哨兵进程
    ./redis-sentinel ../sentinel.conf

注意启动的顺序。**首先是主机（192.168.11.128）的Redis服务进程，然后启动从机的服务进程，最后启动3个哨兵的服务进程。**

**三、Java中使用哨兵模式**

/** * 测试Redis哨兵模式 * @author liu */
    public class TestSentinels { 
        @SuppressWarnings("resource")
        @Test
        public void testSentinel() { 
            JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
            jedisPoolConfig.setMaxTotal(10);
            jedisPoolConfig.setMaxIdle(5);
            jedisPoolConfig.setMinIdle(5);
            // 哨兵信息
            Set<String> sentinels = new HashSet<>(Arrays.asList("192.168.11.128:26379",
                    "192.168.11.129:26379","192.168.11.130:26379"));
            // 创建连接池
            JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels,jedisPoolConfig,"123456");
            // 获取客户端
            Jedis jedis = pool.getResource();
            // 执行两个命令
            jedis.set("mykey", "myvalue");
            String value = jedis.get("mykey");
            System.out.println(value);
        }
    }

上面是通过Jedis进行使用的，同样也可以使用Spring进行配置RedisTemplate使用。

**四、哨兵模式的其他配置项**

<table> 
 <thead> 
  <tr> 
   <th>配置项</th> 
   <th>参数类型</th> 
   <th>作用</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>port</td> 
   <td>整数</td> 
   <td>启动哨兵进程端口</td> 
  </tr> 
  <tr> 
   <td>dir</td> 
   <td>文件夹目录</td> 
   <td>哨兵进程服务临时文件夹，默认为/tmp，要保证有可写入的权限</td> 
  </tr> 
  <tr> 
   <td>sentinel down-after-milliseconds</td> 
   <td>&lt;服务名称&gt;&lt;毫秒数（整数）&gt;</td> 
   <td>指定哨兵在监控Redis服务时，当Redis服务在一个默认毫秒数内都无法回答时，单个哨兵认为的主观下线时间，默认为30000（30秒）</td> 
  </tr> 
  <tr> 
   <td>sentinel parallel-syncs</td> 
   <td>&lt;服务名称&gt;&lt;服务器数（整数）&gt;</td> 
   <td>指定可以有多少个Redis服务同步新的主机，一般而言，这个数字越小同步时间越长，而越大，则对网络资源要求越高</td> 
  </tr> 
  <tr> 
   <td>sentinel failover-timeout</td> 
   <td>&lt;服务名称&gt;&lt;毫秒数（整数）&gt;</td> 
   <td>指定故障切换允许的毫秒数，超过这个时间，就认为故障切换失败，默认为3分钟</td> 
  </tr> 
  <tr> 
   <td>sentinel notification-script</td> 
   <td>&lt;服务名称&gt;&lt;脚本路径&gt;</td> 
   <td>指定sentinel检测到该监控的redis实例指向的实例异常时，调用的报警脚本。该配置项可选，比较常用</td> 
  </tr> 
 </tbody> 
</table>

sentinel down-after-milliseconds配置项只是一个哨兵在超过规定时间依旧没有得到响应后，会自己认为主机不可用。对于其他哨兵而言，并不是这样认为。哨兵会记录这个消息，当拥有认为主观下线的哨兵达到sentinel monitor所配置的数量时，就会发起一次投票，进行failover，此时哨兵会重写Redis的哨兵配置文件，以适应新场景的需要。

[image-20200629221647159]: 
[redis_8]: /images/20221124/be328f1eee754e0daaa74babf0403af2.png
[redis_8 1]: /images/20221124/c1377d4dce4441069a5a8f56dddd187e.png
[redis_8 2]: /images/20221124/8c80dc5a413a4f03888da47f0e41f414.png
[redis_8 3]: /images/20221124/9f8fe44f3e2a45339c3fa9d7567e91aa.png
[redis_8 4]: /images/20221124/f0186fb79f0c4af89b996642acda1047.png
[redis_8 5]: /images/20221124/51da537aca124196af0afbf57ece930e.png
[redis_8 6]: /images/20221124/38432c44c52941b6b6447f9ee9cb3f8c.png
[Redisson]: https://redisson.org/
[Redis]: http://redis.cn/
[Link 1]: https://github.com/redisson/redisson/wiki/8.-%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%E5%92%8C%E5%90%8C%E6%AD%A5%E5%99%A8
[img]: /images/20221124/b906ac1adb76493998656da4c81720d4.png
[img 1]: /images/20221124/afb3050aca8948be96621eef002226ee.png
[img 2]: /images/20221124/69bdf633ac314e7582a1acb34b2c3fca.png
[Link 2]: https://links.jianshu.com/go?to=https%3A%2F%2Fwww.xilidou.com%2F2017%2F10%2F23%2FRedis%E5%AE%9E%E7%8E%B0%E5%88%86%E5%B8%83%E5%BC%8F%E9%94%81%2F
[img 3]: /images/20221124/c2eb31b1bd714f5e8d3d37d616f8eeb5.png
[Link 3]: https://baike.baidu.com/item/%E6%96%87%E4%BB%B6%E6%8F%8F%E8%BF%B0%E7%AC%A6
[Link 4]: https://baike.baidu.com/item/%E6%BA%90%E4%BB%A3%E7%A0%81
[img 4]: /images/20221124/1c050686bfe940d0afc2f50c706883da.png
[img 5]: /images/20221124/8f7c5bee827142ad904d58b3a72aea25.png
[img 6]: /images/20221124/2c97c1e13e064b1e99992d7bf59bae4e.png
[img 7]: /images/20221124/04426a382df1461490725c7fd002e8fa.png
[img 8]: https:upload-images.jianshu.io/upload_images/13880925-4f157293d864b83a.png?imageMogr2/auto-orient/strip|imageView2/2/w/1000/format/webp
[link]: https://baijiahao.baidu.com/s?id=1664285811566919896&wfr=spider&for=pc
[link 1]: https://www.zhihu.com/question/20374919
[link 2]: https://blog.csdn.net/zhujunzhujun/article/details/78345554
[img 9]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xMTMyMDAzOS01N2E3N2NhMjc1N2QwOTI0LnBuZw?x-oss-process=image/format,png
[img 10]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8xMTMyMDAzOS0zZjQwYjE3YzA0MTIxMTZjLnBuZw?x-oss-process=image/format,png