Redis: 本项目旨在学习Redis，有 Redis基础篇：常用数据结构，常用指令，快速入门案例。 Redis实战篇：各种业务场景下的应用 Redis高级篇：Redis部署和配置相关内容，lua语法入门 Redis原理篇：从源码看Redis底层结构，了解网络模型与数据结构仅用于学习，非商用！

# Redis

** ：重点知识点

*** ： 重点应用

## 基础篇

### 认识NoSQL

**非关系型数据库**

| 区别         | SQL                                                     | NoSQL                                                        |
| ------------ | ------------------------------------------------------- | ------------------------------------------------------------ |
| **数据结构** | 结构化                                                  | 非结构化                                                     |
| **数据关联** | 关联的，多表中的外键约束                                | 无关联的                                                     |
| **SQL查询**  | 使用固定的（sql）语句查询，通用                         | 相对简单，不统一                                             |
| **事务特性** | ACID：原子性，一致性，隔离性，持久性                    | BASE（基本可用，允许中间态，最终一致性                       |
| 存储方式     | 磁盘                                                    | 内存                                                         |
|              | 垂直                                                    | 水平                                                         |
| **使用场景** | 1. 数据结构固定  2.相关业务对数据安全性、一致性要求较高 | 1. 数据结构不固定  2. 对一致性、安全性要求不高  3. 对性能要求 |

### 认识Redis

**特征**

- 键值(key-value)型，value支持多种不同数据结构，功能丰富
- 单线程，每个命令具备原子性。6.0 网络请求处理为多线程
- **低延迟，速度快（基于*内存*、IO多路复用、良好的编码）。**
- 支持数据持久化
- 支持主从集群、分片集群
- 支持多语言客户端

### Redis命令

#### Redis 数据结构介绍

Redis是一个key-value的数据库，key一般是String类型，不过value的类型多种多样：

#### Redis通用命令

通用指令是部分数据类型的，都可以使用的指令，常见的有：help commands——帮助文档

1. KEYS 
 - **查看符合模板的所有key**,不建议在生产环境设备上使用
 - ![image-20230116215246866](笔记/Redis/图片/image-20230116215246866.png) 
2. DEL
 - 删除一个指定的key
3. EXISTS
 - 判断 key 是否存在
4. EXPIRE
 - 给一个 key 设置有效期，有效期到期时该 key 会被自动删除
5. TTL
 - 查看一个 key 的剩余有效期
 - -1：永久有效； -2：过期

#### string类型

String类型，也就是字符串类型，是Redis中最简单的存储类型。

其value是字符串，不过根据字符串的格式不同，又可以分为3类：

- string:普通字符串
- int:整数类型，可以做自增、自减操作
- float:浮点类型，可以做自增、自减操作

不管是哪种格式，底层都是字节数组形式存储，只不过是编码方式不同。字符串类型的最大空间不能超过512m

**常见命令**

- SET：添加或者修改已经存在的一个String类型的键值对
- GET:根据key获取String类型的value
- **MSET**:批量添加多个String类型的键值对
- **MGET**:根据多个key获取多个String类型的value
- *INCR*:让一个整型的 key 自增1
- INCRBY:让一个整型的key自增并指定步长，例如：incrby num 2让 num 值自增2
- *INCRBYFLOAT*:让一个浮点类型的数字自增并指定步长
- **SETNX**:添加一个String类型的键值对，前提是这个key不存在，否则不执行
- **SETEX**:添加一个String类型的键值对，并且指定有效期

#### key的结构

Redis的key允许有多个单词形成层级结构，多个单词之间用'：'隔开，格式如下：
								**项目名：业务名：类型：id**
这个格式并非固定，也可以根据自己的需求来删除或添加词条。

#### Hash类型

Hash类型，也叫散列，其value是一个无序字典，类似于引ava中的HashMap结构。

**常见命令**

- **HSET** key field value:添加或者修改hash类型key的field的值
- HGET key field:获取一个hash类型key的field的值
- HMSET:批量添加多个hash类型key的field的值
- HMGET:批量获取多个hash类型key的field的值
- **HGETALL**:获取一个hash类型的key中的所有的field和value（entrySet
- HKEYS:获取一个hash类型的key中的所有的field（keySet
- HVALS:获取一个hash类型的key中的所有的value（values
- HINCRBY:让一个hash类型key的字段值自增并指定步长
- **HSETNX**:添加一个hash类型的key的field值，*前提是这个**field**不存在，否则不执行*

#### List类型

Redis中的List类型与ava中的LinkedList类似，可以看做是一个双向链表结构。既可以支持正向检索和也可以支持反向检索。

**特征**

- 有序
- 元素可以重复
- 插入和删除快
- 查询速度一般

**常见命令**

- **LPUSH** key element..:向列表左侧插入一个或多个元素
- LPOP key:移除并返回列表左侧的第一个元素，没有则返回nil
- RPUSH key element..:向列表右侧插入一个或多个元素
- RPOP key:移除并返回列表右侧的第一个元素
- **LRANGE** key star end:返回一段角标范围内的所有元素
- *BLPOP和BRPOP*:与LPOP和RPOP类似，只不过在没有元素时**等待指定时间**，而不是直接返回nil

**问题**

1. 模拟一个栈
   1. 入口和出口在同一边
   2. LPUSH 和 LPOP / RPUSH 和 RPOP
2. 模拟一个队列
   1. 入口和出口在不同边
   2. LPUSH 和 RPOP
3. 模拟一个阻塞队列
   1. 入口和出口在不同边
   2. 出队时采用 BLPOP 或 BRPOP

#### set类型

Redis的Set结构与java中的HashSet:类似，可以看做是一个value为null的HashMap。因为也是一个hash表，因此具备与HashSet类似的特征：

- 无序
- 元素不可重复
- 查找快
- 支持交集、并集、差集等功能

**常见命令**

- 单集合操作

- SADD key member..:向set中添加一个或多个元素

- SREM key member.:移除set中的指定元素

- **SCARD** key:返回set中元素的个数

- SISMEMBER key member:判断一个元素是否存在于set中

- SMEMBERS:获取set中的所有元素

- 多集合操作
 - SINTER key1key2..:求key1与key2的 交集
 - SDIFF key1key2..:求key1与key2的 差集
 - SUNION key1key2.:求key1和key2的 并集

#### SortedSet类型

Redist的SortedSet是一个可排序的set集合，与java中的TreeSet有些类似，但底层数据结构却差别很大。SortedSet中的每一个元素都带有一个score属性，可以***基于score属性对元素排序***，底层的实现是一个跳表（SkipList)加hash表。

**特性**

- 可排序
- 元素不重复
- 查询速度快

因为SortedSet的可排序特性，经常被用来实现 **排行榜** 这样的功能。

**常见命令**

- ZADD key **score** member:添加一个或多个元素到sorted set,如果已经存在则更新其score值
- ZREM key member:删除sorted set中的一个指定元素
- **ZSCORE** key member:获取sorted set中的**指定元素的score值**
- **ZRANK** key member:获取sorted set中的指定元素的**排名**
- ZCARD key:获取sorted set中的元素个数
- **ZCOUNT** key min max:统计**score值在给定范围内**的所有元素的**个数**
- ZINCRBY key increment member:让sorted set中的指定元素自增，步长为指定的increment值
- **ZRANGE** key min max :按照score排序后，获取**指定排名范围内的元素**（按角标 0 ~ n）
- **ZRANGEBYSCORE** key min max [WITHSCORES] [LIMIT offset count]:按照score:排序后，获取指定score范围内的元素（按分数查，offset：第一次参数的偏移量，count：查几条）
  - 滚动查询：
  - ZRANGEBYSCORE z1 0 10000 WITHSCORES LIMIT **0** 3
  -  ZRANGEBYSCORE z1 **2** 10000 WITHSCORES LIMIT 1 3

- ZDIFF、ZINTER、ZUNION:求差集、交集、并集

**注意**：所有的排名**默认都是升序**，如果要降序则在命令的 ***Z 后面添加 REV 即可***

### Redis的java客户端

### Jedis

#### Jedisi连接池

Jedis.本身是线程不安全的，并且频繁的创建和销毁连接会有性能损耗，因此我们推荐大家使用 jedis连接池 代替 jedis 的直连方式

### SpringData

SpringData:是Spring中数据操作的模块，包含对各种数据库的集成，其中对Redis的集成模块就叫做SpringDataRedis，官网地址：https:/spring.io/projects/spring-data-redis

- 提供了对不同Redis客户端的整合(Lettuce 和 jedis)
- 提供了RedisTemplate:统一API来操作 Redis
- 支持 Redis 的发布订阅模型
- 支持 Redis 哨兵和 Redis 集群
- 支持基于 Lettuce 的响应式编程
- 支持基于刊 JDK、JS0N、字符串、Spring对象的数据序列化及反序列化
- 支持基于 Redis 的 JDKCollection:实现

#### **快速入门**

2. 配置文件

```yaml
spring:
  redis:
    host: 192.168.153.128
    port: 6379
    password:
    lettuce:
      pool:
        max-active: 8
        max-idle: 8
        min-idle: 0
        max-wait: 100
```

#### 序列化方式

RedisTemplate可以接收任意Object作为值写入Redis,只不过写入前会把Object/序列化为字节形式，默认是采用JDK序列化，得到的结果是这样的：

![image-20230117160739894]( 笔记/Redis/图片/image-20230117160739894.png)

- 可读性差
- 内存占用较大

##### **自定义序列化方式**

```java
@Configuration
public class RedisConfig {

@Bean
 public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
 // 创建 RedisTemplate 对象
 RedisTemplate<String, Object> template = new RedisTemplate<>();
 // 设置连接工厂
 template.setConnectionFactory(connectionFactory);
 // 创建 JSON 序列化工具
 GenericJackson2JsonRedisSerializer jackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
 // 设置 key 序列化
 template.setKeySerializer(RedisSerializer.string());
 template.setHashKeySerializer(RedisSerializer.string());
 // 设置 value 序列化
 template.setValueSerializer(RedisSerializer.json());
 template.setValueSerializer(RedisSerializer.json());
 //返回
 return template;
 }
}
```

##### **存在的问题**

为了在 反序列化 时知道对象的类型，JSON序列化器会将类的class类型写入json结果中，存入Redis,会带来额外的内存开销。

##### 解决措施

为了节省内存空间，我们并不会使用 JSON 序列化器来处理 value ,而是**统一使用String序列化器**，要求只能存储 String 类型的key和value。当需要存储 Java 对象时，**手动完成**对象的序列化和反序列化。

##### StringRedisTemplate

Spring默认提供了一个StringRedisTemplate:类，它的key和value的序列化方式默认就是String方式。省去了我们自定义RedisTemplate的过程：

```java
@Resource
private StringRedisTemplate stringRedisTemplate;

@Test
void testSaveUser() {
    User user = new User("yoshino", 22);
    // 手动序列化
    Gson gson = new Gson();
    String json = gson.toJson(user);

stringRedisTemplate.opsForValue().set("user:100", json);
    String jsonUser = stringRedisTemplate.opsForValue().get("user:100");
    // 手动反序列化
    User user1 = gson.fromJson(jsonUser, User.class);
    System.out.println(json);
    System.out.println(user1);
}
```

## 实战篇

### 短信登录

#### 导入

**数据库**

**后端直接导入，修改 yml 文件**

输入：[localhost:8081/shop-type/list](http://localhost:8081/shop-type/list) 测试

**前端导入**

#### 基于 Session 实现登录

##### 发送短信验证码

##### 短信验证码登录

##### 登录验证功能

##### 隐藏用户敏感信息

```java
//需要引入
<dependency>
 <groupId>cn.hutool</groupId>
 <artifactId>hutool-all</artifactId>
 <version>5.7.17</version>
</dependency>

session.setAttribute("user", BeanUtil.copyProperties(user, UserDTO.class));
```

#### 集群的 session 共享问题

**session共享问题**：多台Tomcat并不共享session存储空间，当请求切换到不同tomcat服务时导致数据丢失的问题。

**方案应该满足：**

- 数据共享
- 内存存储
- key、value:结构

#### 基于 Redis 实现共享 Session 登录

```java
// 7.2将 User 对象转为 Hash 存储
UserDTO userDTO = BeanUtil.copyProperties(user, UserDTO.class);
Map<String, Object> userMap = BeanUtil.beanToMap(userDTO, new HashMap<>(),
 CopyOptions.create()
 .setIgnoreNullValue(true)
 .setFieldValueEditor((fieldName, fieldValue) -> fieldValue.toString()));
```

**Redis代替session需要考虑的问题：**

- 选择合适的数据结构
- 选择合适的key
- 选择合适的存储粒度
- ***注意过期时间***

#### 登录拦截器的优化

**问题**

### 商品查询缓存

#### 什么是缓存

**缓存**就是数据交换的缓冲区（称作 Cache),是存贮数据的临时地方，一般读写性能较高

#### 添加 redis 缓存

**给店铺类型查询业务添加缓存**

#### 缓存更新策略

**业务场景**

- 低一致性需求：使用内存淘汰机制。例如店铺类型的查询缓存
- 高一致性需求：主动更新，并以超时剔除作为兜底方案。例如店铺详情查询的缓存

**主动更新策略**

3：停电数据丢失，数据库的数据非准确的

**选择 1 策略**

#### 给查询商铺的缓存添加超时剔除和主动更新的策略

修改ShopController中的业务逻辑，满足下面的需求：

- 根据d查询店铺时，如果缓存未命中，则查询数据库，将数据库结果写入缓存，并设置超时时间
- 根据id修改店铺时，先修改数据库，再删除缓存

### **缓存穿透

**缓存穿透** 是指客户端请求的数据在缓存中和数据库中都不存在，这样缓存永远不会生效，这些请求都会打到数据库。

**解决方案**

- 缓存空对象
  - 优点：实现简单，维护方便
  - 缺点
    - 额外的内存消耗（设置 TTL）
    - 可能造成短期的不一致
- 布隆过滤
  - 优点：内存占用较少，没有多余 key
  - 缺点
    - 实现复杂
    - 存在误判可能

**缓存穿透产生的原因是什么？**

- 用户请求的数据在缓存中和数据库中**都不存在**，不断发起这样的请求，给数据库带来巨大压力

**缓存穿透的解决方案有哪些？**

- 缓存 null 值
- 布隆过滤
- 增强 id 的复杂度，避免被猜测 id 规律
- **做好数据的基础格式校验**（在访问前就将不符合格式的 id 查询拦截
- 加强用户权限校验
- 做好热点参数的限流

### **缓存雪崩

**缓存雪崩 **是指在同一时段大量的缓存 key 同时失效或者 Redis 服务宕机，导致大量请求到达数据库，带来巨大压力。

**解决方案**

- 给不同的 Key 的 TTL 添加随机值
- 利用 Redis 集群提高服务的可用性
- 给缓存业务添加降级限流策略
- 给业务添加多级缓存

### **缓存击穿（热点 key）

缓存击穿问题也叫**热点 Key 问题**，就是一个被**高并发访问**并且**缓存重建业务较复杂**的 key 突然失效了，无数的请求访问会在瞬间给数据库带来巨大的冲击。

**解决方案**

- 互斥锁
- 逻辑过期

### 商品查询缓存

#### 基于互斥锁方式解决缓存击穿问题

需求：修改根据d查询商铺的业务，基于互斥锁方式来解决缓存击穿问题

#### 基于逻辑过期方式解决缓存击穿问题

需求：修改根据d查询商铺的业务，基于逻辑过期方式来解决缓存击穿问题

**活动开始前，就已经存好（已经预热过的，在活动中的热点店铺）**

### ***缓存工具封装(CacheClient)

基于StringRedisTemplate:封装一个缓存工具类，满足下列需求：

- 方法1：将任意 Java 对象序列化为 json 并存储在 string 类型的 key 中，并且可以设置 TTL 过期时间
- 方法2：将任意 Java 对象序列化为 json 并存储在 string 类型的 key 中，并且可以设置逻辑过期时间，用于处理缓存击穿问题
- 方法3：根据指定的 key 查询缓存，并反序列化为指定类型，利用缓存空值的方式解决缓存穿透问题
- 方法4：根据指定的 key 查询缓存，并反序列化为指定类型，需要利用逻辑过期解决缓存击穿问题

### ***优惠券秒杀

#### 全局唯一 ID

##### 全局ID生成器

当用户抢购时，就会生成订单并保存到tb_voucher_order这张表中，而订单表如果使用数据库自增ID就存在一些问题：

- id的规律性太明显
- 受单表数据量的限制

全局引D生成器，是一种在分布式系统下用来**生成全局唯一 ID 的工具**，一般要满足下列特性：

为了增加 ID 的安全性，我们可以不直接使用Redis自增的数值，而是拼接一些其它信息：

**ID的组成部分：**

- 符号位：1bit, 永远为0
- 时间戳：31bit,以秒为单位，可以使用69年
- 序列号：32bit,秒内的计数器，支持每秒产生2^32个不同 ID

##### 总结

**全局唯一 ID 生成策略**

- UUID
- Redis 自增
- snowflake 算法
- 数据库自增

**Redis 自增 ID 策略**

- 每天一个 key，方便统计订单量
- ID 构造是 时间戳 + 计数器

#### 实现添加优惠券秒杀下单

每个店铺都可以发布优惠券，分为平价券和特价券。平价券可以任意购买，而特价券需要秒杀抢购：

表关系如下：

- tb_voucher:优惠券的基本信息，优惠金额、使用规则等
- tb_seckill_voucher:优惠券的库存、开始抢购时间，结束抢购时间。特价优惠券才需要填写这些信息

#### 实现秒杀下单

下单时需要判断两点：

- 秒杀是否开始或结束，如果尚未开始或已经结束则无法下单
- 库存是否充足，不足则无法下单

#### 超卖问题

**测试**

使用 JMeter 测试

**错误流程图**

超卖问题是典型的多线程安全问题，针对这一问题的常见解决方案就是 **加锁**：

##### **乐观锁**

乐观锁的关键是判断之前查询得到的数据是否有被修改过，常见的方式有两种：

- 版本号

- CAS 法

> stock = 原来的，改为 stock > 1

**超卖这样的线程安全问题，解决方案有哪些**？

1. 悲观锁：添加同步锁，让线程串行执行
   - 优点：简单粗暴
   - 缺点：性能一般
2. 乐观锁：不加锁，在更新时判断是否有其它线程在修改
   - 优点：性能好
   - 缺点：存在成功率低的问题

### 一人一单

需求：修改秒杀业务，要求同一个优惠券，一个用户只能下一单

### 集群模式下线程并发问题

#### 一人一单的并发安全问题

通过加锁可以解决在单机情况下的一人一单安全问题，但是在集群模式下就不行了。

> Ctrl + D 实现

![image-20230124225931057]( 笔记/Redis/图片/image-20230124225931057.png)

**出现一人多单的问题**

### 分布式锁

#### 工作原理

**什么是分布式锁**

分布式锁：满足分布式系统或集群模式下 **多进程可见** 并且 **互斥** 的锁。

#### 基于Redis的分布式锁

实现分布式锁时需要实现的两个基本方法：

- 获取锁

- 互斥：确保只能有一个线程获取锁
 - <img src=" 笔记/Redis/图片/image-20230125105149519.png" alt="image-20230125105149519" style="zoom:67%;" /> 
 - <img src=" 笔记/Redis/图片/image-20230125105424096.png" alt="image-20230125105424096" style="zoom:80%;" />

- 释放锁

- 手动释放
  - 超时释放：获取锁时添加一个超时时间
  - DEL  lock

#### 基于Redis:实现分布式锁初级版本

需求：定义一个类，实现下面接口，利用Redis.实现分布式锁功能。

```java
public interface ILock {
    /**
     * 尝试获取锁
     * @param timeoutSec 锁持有的超时时间，过期后自动释放
     * @return true代表获取锁成功；false代表获取锁失败
     */
    boolean tryLock(long timeoutSec);

/**
     * 释放锁
     */
    void unlock();
}
```

<img src=" 笔记/Redis/图片/image-20230125113106401.png" alt="image-20230125113106401" style="zoom: 67%;" /> 单台 jvm 适用

**误删问题**

#### 改进Redis的分布式锁：对value值进行验证

需求：修改之前的分布式锁实现，满足：

1. 在获取锁时存入线程标示（可以用UUD表示）
2. 在释放锁时先获取锁中的线程标示，判断是否与当前线程标示一致
   - 如果一致则释放锁
   - 如果不一致则不释放锁

**误删问题**

#### ***Redis的Lua脚本

Redis提供了Lua脚本功能，在一个脚本中编写多条Redist命令，确保多条命令执行时的原子性。Lua是一种编程语言，它的基本语法大家可以参考网站：https:/www.runoob.com/lua/lua-tutorial.html

写好脚本以后，需要用Redis命令来调用脚本，调用脚本的常见命令如下：

**释放锁的业务流程**是这样的：

1. 获取锁中的线程标示
2. 判断是否与指定的标示（当前线程标示)一致
3. 如果一致则释放锁（删除）
4. 如果不一致则什么都不做

**Lua 脚本表示：**

```lua
-- 获取锁中的线程标示 get key
local id = redis.call('get',  KEYS[1])
-- 比较线程标示与锁中的标示是否一致
if (id == ARGV[1]) then
    -- 释放锁
    return redis.call('del',  KEYS[1])
end
return 0
```

#### 再次改进 Redis 的分布式：使用 lua 脚本

**实现**

**基于setnx实现的分布式锁存在下面的问题：**

### **Redisson

Redisson:是一个在**Redis的基础上**实现的 **java 驻内存数据网格**(In-Memory Data Grid)。它不仅提供了一系列的分布式的 java 常用对象，还提供了许多分布式服务，其中就包含了各种分布式锁的实现。

官网地址：https://redisson.org
GitHub地址：https://github.com/redisson/redisson

#### Redisson 入门

1. 引入依赖：

```xml

<dependency>
 <groupId>org.redisson</groupId>
 <artifactId>redisson</artifactId>
 <version>3.17.5</version>
</dependency>
```

2. 配置Redisson客户端：

```java
/**
 * @author yoshino
 **/
@Configuration
@ConfigurationProperties(prefix = "spring.redis")
@Data
public class RedissonConfig {

private String host;
    private String port;
    @Bean
    public RedissonClient redissonClient(){
        // 配置
        Config config = new Config();
        String redisAddress = String.format("redis://%s:%s", host, port);
        config.useSingleServer().setAddress(redisAddress);
        // 创建 RedissonClient 对象
        return Redisson.create(config);
    }
}
```

3. 使用 Redisson 的分布式锁：

#### Redisson 可重入锁原理:使用 hash 结构

##### **获取锁的 lua 脚本(Redisson 获取锁的 lua 源码)**

```lua
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1];-- 线程唯一标识
local releaseTime = ARGV[2];-- 锁的自动释放时间
-- 判断惭是否存往
if (redis.call('exists', key) == 0) then
    -- 不存在，获取锁
    redis.call('hincrby', key, threadId, '1');
    -- 设置有效期
    redis.call('expire', key, releaseTime);
    return nil; -- 返回结果
end;
-- 锁已经存在，判断threadId是否是自己
if (redis.call('hexists', key, threadId) == 1)
then
    -- 存在，获取锁，重入次数 + 1
    redis.call('hincrby', key, threadId, '1');
    -- 设置有效期
    redis.call('expire', key, releaseTime);
    return nil;--返回结果
end ;
return redis.call('pttl', key);--代码走到这里，说明获取锁的不是自己，获取锁失败，返回锁的剩余有效期
```

##### **释放锁的 lua 脚本(Redisson 释放锁的 lua 源码)**

```lua
local key = KEYS[1]; -- 锁的key
local threadId = ARGV[1];-- 线程唯一标识
local releaseTime = ARGV[2];-- 锁的自动释放时间
-- 判断当前锁是否还是被自己持有
if (redis.call('HEXISTS', key, threadId) == 0)
then
    return nil;-- 如果已经不是自己，则直接返回
end
--是自己的锁，则重入次数-1
local count = redis.call('HINCRBY', key, threadId, -1);
-- 判断是否重入次数是否已经为0
if (count > 0) then
    -- 大于0说明不能释放锁，
    -- 重置有效期然后返回
    redis.call('EXPIRE', key, releaseTime);
    return 0;
else
    -- 等于0说明可以释放锁直接删除
    redis.call('DEL', key);
    rediscall('publish',KEYS[2],threadId); -- 发布消息通知
    return 1;
end
return nil;
```

#### 基于 Redisson 的分布式锁的优化：可重试和超时续约

优化 setnx 实现的分布式锁带来的问题

**trylock 携带 等待时间时 (锁重试)： **利用 消息队列 和 信号量 方式

**trylock 不携带 释放时间时(-1) (锁重试)：**看门狗机制 每 10s 更新一次 ttl

**如何取消（释放锁时）**

##### 获取锁

##### 释放锁

#### Redisson分布式锁主从一致性问题：RedissonMultiLock

`lock = redissonClient.getMultiLock(lock1, lock2, lock3);`

**必须获取所有 Node 的锁才算成功 获取，否则算失败**（联锁）

#### **Redisson分布式锁原理

- **可重入：**利用 **hash** 结构记录线程 id 和重入次数
- **可重试：**利用 信号量 和 Pub-Sub 功能实现等待、唤醒、获取锁失败的重试机制（有等待时间）
- **超时续约：**利用 watchDog，每隔一段时间（releaseTime / 3），重置超时时间（释放时间参数必须为 -1，默认提供的释放时间 30s）

#### 锁总结

1. 不可重入Redis分布式锁：
   - 原理：利用setnx的互斥性；利用eX避免死锁；释放锁时判断线程标示
   - 缺陷：不可重入、无法重试、锁超时失效
2. 可重入的Redis:分布式锁：
   - 原理：利用hash结构，记录线程标示和重入次数；利用watchDog延续锁时间；利用信号量控制锁重试等待
   - 缺陷：redis宕机引起锁失效问题

3. Redisson的multiLock:

- 原理：多个独立的Redis节点，**必须在所有节点都获取重入锁**，才算获取锁成功

- 缺陷：运维成本高、实现复杂

### Redis 优化秒杀

上图，为串联的执行顺序，效率低

**解决方案**

**redis 的数据结构**

**redis 中的操作（lua 脚本）**

**tomcat 中的操作**

#### 改进秒杀业务，提高并发性能

**需求：**

1. 新增秒杀优惠券的同时，将优惠券信息保存到Redis中
2. 基于Lua脚本，判断秒杀库存、一人一单，决定用户是否抢购成功
3. 如果抢购成功，将优惠券id和用户id封装后存入阻塞队列
4. 开启线程任务，不断从阻塞队列中获取信息，实现异步下单功能

#### 秒杀业务的优化思路是什么？

1. 先利用Redis完成库存余量、一人一单判断，完成抢单业务
2. 再将下单业务放入阻塞队列，利用独立线程异步下单

#### 基于阻塞队列的异步秒杀存在哪些问题？

- 内存限制问题（BlockQuery 使用 JDK 的内存，需要加以限制）
- 数据安全问题（数据库操作失败后，Redis 的数据 和 数据库不一致 问题）

### Redis消息队列实现异步秒杀

消息队列（Message Queue)，字面意思就是存放消息的队列。最简单的消息队列模型包括3个角色：

- 消息队列：存储和管理消息，也被称为消息代理(Message Broker)
- 生产者：发送消息到消息队列
- 消费者：从消息队列获取消息并处理消息

**Redis提供了三种不同的方式来实现消息队列：**

- list结构：基于List结构模拟消息队列
- PubSub:基本的点对点消息模型
- Stream:比较完善的消息队列模型

#### 基于 List 结构模拟消息队列

消息队列（Message Queue),字面意思就是存放消息的队列。而Redis的list数据结构是一个双向链表，很容易模拟出队列效果。

队列是**入口和出口不在一边**，因此我们可以利用：LPUSH结合RPOP、或者RPUSH结合LPOP来实现。

不过要注意的是，当队列中没有消息时RPOP或LPOP操作会返回 null,并不像 JVM 的阻塞队列那样会阻塞并等待消息。

因此这里应该使用**BRPOP或者BLPOP**来实现阻塞效果。

**基于Lst的消息队列有哪些优缺点？**

优点：

- 利用Redis存储，不受限于 JVM 内存上限
- 基于Redis的持久化机制，数据安全性有保证
- 可以满足消息有序性

缺点：

- 无法避免消息丢失（拿到了就 remove 了，不能确保一定完成业务）
- 只支持单消费者

#### 基于 PubSub 的消息队列

**PubSub（发布订阅)**是Redis2.0版本引入的消息传递模型。顾名思义，消费者可以订阅一个或多个channel,生产者向对应channel发送消息后，所有订阅者都能收到相关消息。

- SUBSCRIBE channel[channel]:订阅一个或多个频道
- PUBLISH channel msg:向一个频道发送消息
- PSUBSCRIBE pattern[pattern]:订阅与pattern格式匹配的所有频道

**基于PubSub的消息队列有哪些优缺点？**

优点：

- 采用发布订阅模型，支持多生产、多消费

缺点：

- 不支持数据持久化（没有消费者时，消息丢失）
- 无法避免消息丢失
- 消息堆积有上限，超出时数据丢失（仅在消费者有缓存窗口）

#### ***基于Stream的消息队列

Stream是Redis5.0引入的一种新数据类型，可以实现一个功能非常完善的消息队列。

**发送消息的命令**

##### **读取消息的方式之一： XREAD**

> 阻塞时间 ：0 表示等待时间无限长

**！注意**

当我们指定起始 ID 为$时，代表读取最新的消息，如果我们处理一条消息的过程中又有超过1条以上的消息到达队列，则下次获取时也只能获取到最新的一条，会出现 **漏读消息** 的问题。

**STREAM:类型消息队列的XREAD命令特点：**

- 消息可回溯
- 一个消息可以被多个消费者读取
- 可以阻塞读取
- 有消息漏读的风险

##### 基于Stream的消息队列-消费者组

消费者组（**Consumer Group**):将多个消费者划分到一个组中，监听同一个队列。具备下列特点：

**创建消费者组**

**从消费者组读取消息**

**确认消息**

![image-20230129213033101]( 笔记/Redis/图片/image-20230129213033101.png)

**读取pending-list中的消息（消费但未确认）**

![image-20230129213329487]( 笔记/Redis/图片/image-20230129213329487.png)

`XPENDING s1 g1 - + 10`：读10条，- + 全部

**STREAM类型消息队列的XREADGROUP命令特点：**

- 消息可回溯
- 可以多消费者争抢消息，加快消费速度
- 可以阻塞读取
- 没有消息漏读的风险
- 有消息确认机制，保证消息至少被消费一次

#### 基于Redis的Stream结构作为消息队列，实现异步秒杀下单

**需求：**

1. 创建一个Stream类型的消息队列，名为stream.orders `127.0.0.1:6379> XGROUP CREATE stream.orders g1 0 MKSTREAM`
2. 修改之前的秒杀下单Lua脚本，在认定有抢购资格后，直接向stream.orders中添加消息，内容包含voucherld、userld、orderld
3. 项目启动时，开启一个线程任务，尝试获取stream.orders中的消息，完成下单

### 达人探店

探店笔记类似点评网站的评价，往往是图文结合。对应的表有两个：

- tb_bog:探店笔记表，包含笔记中的标题、文字、图片等
- tb_blog_comments:其他用户对探店笔记的评价

#### 发布探店笔记

点击首页最下方菜单栏中的+按钮，即可发布探店图文：

#### 实现查看发布探店笔记的接口

需求：点击首页的探店笔记，会进入详情页面，实现该页面的查询接口：

#### 点赞

在首页的探店笔记排行榜和探店图文详情页面都有点赞的功能：

**需求**

- 同一个用户只能点赞一次，再次点击则取消点赞
- 如果当前用户已经点赞，则点赞按钮高亮显示（前端已实现，判断字段 Blog 类的 isLike 属性）

#### 点赞排行榜

在探店笔记的详情页面，应该把给该笔记点赞的人显示出来，比如最早点赞的TOP5,形成点赞排行榜：

**实现查询点赞排行榜的接口**
需求：按照点赞时间先后排序，返回Top5的用户

**问题**

数据库查询时，是使用 in 来获取，导致点赞顺序的排序问题

解决：`WHERE id IN ( 1010 , 1 ) ORDER BY FIELD(id, 1010, 1)`

拼接：`String idStr = StrUtil.join(",", ids);`

查询语句：`query().in("id", ids).last("ORDER BY FIELD(id," + idStr + ")").list()`

### 好友关注

#### 关注和取关

需求：基于该表数据结构，实现两个接口：

1. 关注和取关接口
2. 判断是否关注的接口

关注是User之间的关系，是博主与粉丝的关系，数据库中有一张tb follow表来标示：

#### 共同关注

需求：利用Rdis中恰当的数据结构，实现共同关注功能。在博主个人页面展示出当前用户与博主的共同好友。

#### 关注推送

关注推送也叫做Feed流，直译为 **投喂** 。为用户持续的提供“沉浸式”的体验，通过无限下拉刷新获取新的信息。

Feed流产品有两种常见模式：

- **Timeline**:不做内容筛选，简单的按照内容发布时间排序，常用于好友或关注。例如朋友圈

- 优点：信息全面，不会有缺失。并且实现也相对简单

- 缺点：信息噪音较多，用户不一定感兴趣，内容获取效率低

- **智能排序**：利用智能算法屏蔽掉违规的、用户不感兴趣的内容。推送用户感兴趣信息来吸引用户
  - 优点：投喂用户感兴趣信息，用户粘度很高，容易沉迷
  - 缺点：如果算法不精准，可能起到反作用

本例中的个人页面，是基于关注的好友来做Feed流，因此采用 Timeline 的模式。该模式的实现方案有三种：

- 拉模式：也叫做读扩散

- 推模式：也叫做写扩散

- 推拉结合：也叫做读写混合，兼具推和拉两种模式的优点

#### **基于推模式实现关注推送功能**

**需求**：

1. 修改新增探店笔记的业务，在保存bog到数据库的同时，推送到粉丝的收件箱
2. 收件箱满足可以根据时间戳排序，必须用Redis的数据结构实现
3. 查询收件箱数据时，可以实现分页查询

**Feed流的分页问题**

Feed流中的数据会不断更新，所以数据的角标也在变化，因此不能采用传统的分页模式。

**Feed 流的滚动分页**

#### 实现关注推送页面的分页查询

**需求**：在个人主页的“关注”卡片中，查询并展示推送的blog信息：

![image-20230206111728701]( 笔记/Redis/图片/image-20230206111728701.png)

### 附近商户

#### GEO数据结构

GEO就是Geolocation的简写形式，代表地理坐标。Redis在3.2版本中加入了对GEO的支持，允许存储地理坐标信息，帮助我们根据经纬度来检索数据。常见的命令有：

#### 附近商户搜索

在首页中点击某个频道，即可看到频道下的商户：

按照**商户类型**做分组，类型相同的商户作为同一组，以 typeld 为 key 存入同一个GEO集合中即可

SpringDataRedisl的2.3.9版本并不支持Redis6.2提供的GEOSEARCH命令，因此我们需要提示其版本，修改自己的POM文件，内容如下：

### 用户签到

#### BitMap用法

把每一个bit位对应当月的每一天，形成了映射关系。用0和1标示业务状态，这种思路就称为**位图(BitMap)**

Redis中是利用**string**类型数据结构实现BitMap,因此最大上限是**512M**,转换为bit则是2^32个bit位

#### 签到功能

需求：实现签到接口，将当前用户当天签到信息保存到Redis中

#### 签到统计

#### 实现签到统计功能

### UV 统计

#### HyperLogLog 用法

首先我们搞懂两个概念：

**UV**:全称 Unique Visitor,也叫*独立访客量*，是指通过互联网访问、浏览这个网页的自然人。1天内同一个用户多次访问该网站，只记录1次。

**PV**：全称Page View,也叫*页面访问量或点击量*，用户每访问网站的一个页面，记录1次PV,用户多次打开页面，则记录多次PV。往往用来衡量网站的流量。

#### 实现 UV 统计

## 高级篇

### 分布式缓存

#### 单点 redis 的问题

数据丢失问题：Reds是内存存储，服务重启可能会丢失数据

并发能力问题：单节点Redis并发能力虽然不错，但也无法满足如618这样的高并发场景

故障恢复问题：如果Redis:宕机，则服务不可用，需要一种自动的故障恢复手段

存储能力问题：Redis基于内存，单节点能存储的数据量难以满足海量数据需求

**方法：**

实现 Redis 数据持久化

搭建主从集群，实现读写分离

利用 Redis 哨兵，实现健康检测和自动恢复

搭建分片集群，利用插槽机制实现动态扩容

#### Redis 持久化

##### RDB持久化

RDB全称Redis Database Backup file（Redis数据备份文件)，也被叫做***Redis数据快照***。简单来说就是把内存中的所有数据都记录到磁盘中。当Redis实例故障重启后，从磁盘读取快照文件，恢复数据。

快照文件称为RDB文件，默认是保存在当前运行目录。

Redis**停机时**会执行一次RDB。保存到当前运行的目录下

**配置**

**原理**

bgsave开始时会 fork 主进程得到子进程，子进程**共享**主进程的内存数据。完成 fork 后读取内存数据并写入RDB文件。

fork采用的是copy-on-write技术：

- 当主进程执行读操作时，访问共享内存；
- 当主进程执行写操作时，则会拷贝一份数据，执行写操作。

**总结**

RDB方式bgsave的基本流程？

- fork 主进程得到一个子进程，共享内存空间
- 子进程读取内存数据并写入新的 RDB 文件
- 用新RDB文件 **替换** 旧的 RDB 文件。

RDB会在什么时候执行？Save 60 1000代表什么含义？

- 默认是服务停止时。
- 代表60秒内至少执行1000次修改则触发RDB

RDB的缺点？

- RDB执行间隔时间长，两次RDB之间写入数据有丢失的风险
- fork子进程、压缩、写出RDB文件都比较耗时

##### AOF特久化

AOF全称为Append Only File（追加文件)。Redis处理的每一个写命令都会记录在AOF文件，可以看做是命令日志文件。

**配置**

AOF默认是关闭的，需要修改redis.conf配置文件来开启AOF:

AOF的命令记录的频率也可以通过redis.conf文件来配：

因为是记录命令，AOF文件会比RDB文件大的多。而且AOF会记录对同一个key的多次写操作，但只有最后一次写操作才有意义。通过执行 ***bgrewriteaof*** 命令，可以让AOF文件执行重写功能，用最少的命令达到相同效果。
<img src=" 笔记/Redis/图片/image-20230209161620900.png" alt="image-20230209161620900" style="zoom:50%;" />

Redis也会在触发阈值时**自动**去重写AOF文件。阈值也可以在redis.conf中配置：

##### 总结

### Redis 主从集群

#### 搭建主从架构

单节点Redis的并发能力是有上限的，要进一步提高Redist的并发能力，就需要搭建主从集群，实现读写分离。

#### 搭建主从集群

视频 P102

假设有A、B两个Redis:实例，如何让B作为A的slave节点？

- 在B节点执行命令：slaveof A的IP A的port

#### 数据同步原理

master如何判断slave是不是第一次来同步数据？这里会用到两个很重要的概念：

- **Replication Id**:简称replid,是数据集的标记，id一致则说明是同一数据集。每一个master都有唯一的replid,slave 则会继承master节点的replid
- **offset**:偏移量，随着记录在repl_baklog中的数据增多而逐渐增大。slave 完成同步时也会记录当前同步的offset。如果slave的offset小于master的offset,说明slave数据落后于master,需要更新。

因此 slave 做数据同步，必须**向 master** 声明自己的**replication id和offset**,master才可以判断到底需要同步哪些数据

<img src=" 笔记/Redis/图片/image-20230209211041597.png" alt="image-20230209211041597" style="zoom:50%;" /> slave 第一次同步前，自身就是 master，replid 记录为本身

**简述全量同步的流程？**

- slave 节点请求增量同步
- master 节点判断 replid ,发现不一致，拒绝增量同步
- master 将完整内存数据生成 RDB,发送RDB到slave
- slave 清空本地数据，加载 master 的RDB
- master 将RDB期间的命令记录在 repl_baklog ,并持续将log中的命令发送给 slave
- slave 执行接收到的命令，保持与 master 之间的同步

#### 总结

简述全量同步和增量同步区别？

- 全量同步：master将完整内存数据生成RDB,发送RDB到slave。后续命令则记录在repl_baklog,逐个发送给slave。
- 增量同步：slave提交自己的 offset 到master,master获取repl_baklog中从offset之后的命令给slave

什么时候执行全量同步？

- slave节点**第一次**连接masteri节点时
- slave节点**断开时间太久**，repl_baklog中的 offset 已经被覆盖时

什么时候执行增量同步？

- slave节点**断开又恢复**，并且在 repl_baklog 中**能找到offse**t时

### Redis 哨兵

slave节点宕机恢复后可以找master节点同步数据，那master节点宕机怎么办？

#### 哨兵的作用和原理

Redis:提供了哨兵(Sentinel)机制来实现主从集群的自动故障恢复。哨兵的结构和作用如下：

**服务状态监控**

#### 总结

Sentinel的三个作用是什么？

- 监控
- 故障转移
- 通知

Sentinel如何判断一个redis实例是否健康？

- 每隔 1s 发送一次 ping 命令，如果超过一定时间没有相向，则认为是 **主观下线**
- 如果 大多数 sentinel 都认为实例主观下线，则判断 服务下线

故障转移步骤有哪些？

- 首先选定一个slave作为新的master,执行slaveof no
- 然后让所有节点都执行slaveof新master
- 修改故障节点配置，添加slaveof新master

#### 搭建哨兵集群

视频 p106

#### RedisTemplate 的哨兵模式

在Sentinel集群监管下的Redis主从集群，其节点会因为自动故障转移而发生变化，Redis的客户端必须感知这种变化，及时更新连接信息。Spring的RedisTemplate底层利用lettuce实现了节点的感知和自动切换。

#### Sentinel之间互相监控

Sentinel是特殊的Redis节点，也能发布订阅；

Sentinel没有主从之分；

Sentinel**订阅所有Redis节点的_sentinel_:hello频道**，并在**上线时**给所有Redis节点的_sentinel_:hello**频道发送消息**，包括自己的host、进程ID（runid）、以及Master配置，让**其他Sentinel感知**自己，**更新存储的Sentinel列表**（如果是新的host、新的进程号，则进行添加；如果已经有host相同，但进程ID不同的Sentinel，代表原有的Sentinel重启了，会进行替换），如果Master配置不同，代表其他的Sentinel的Master配置的版本低，也会进行替换。

Sentinel上线后会每隔1秒给其他Sentinel发送消息。

### Redis 分片集群

#### 搭建分片集群

##### 分片集群结构

主从和哨兵可以解决高可用、高并发读的问题。但是依然有两个问题没有解决：

- 海量数据存储问题
- 高并发写的问题

使用分片集群可以解决上述问题，分片集群特征：

- 集群中有多个master,每个master保存不同数据
- 每个master都可以有多个slave节点
- master之间通过ping监测彼此健康状态
- 客户端请求可以访问集群任意节点，最终都会被转发到正确节点

![image-20230210131625239]( 笔记/Redis/图片/image-20230210131625239.png)

##### **搭建**

p108

#### 散列插槽

Redis会把每一个 master 节点映射到 0~16383 共16384个插槽(hash slot)上，查看集群信息时就能看到：

数据**key**不是与节点绑定，而是**与插槽绑定**。redis会根据key的有效部分计算插槽值，分两种情况：

- key 中包含"{}"，且“{}”中至少包含1个字符，“{}” 中的部分是有效部分
- key 中不包含“{}”，整个 key 都是有效部分

**例如**：key是num,那么就根据num计算，如果是{itcast}num,则根据itcasti计算。计算方式是利用CRC16算法得到一个hash值，然后对16384取余，得到的结果就是slot值。

##### 总结

Redis如何判断某个key应该在哪个实例？

- 将16384个插槽分配到不同的实例
- 根据key的有效部分计算哈希值，对16384取余
- 余数作为插槽，寻找插槽所在实例即可

如何将**同一类数据固定**的**保存在同一个Redis实例**？

- 这一类数据使用相同的有效部分，例如key都以 {typeld} 为前缀

#### 集群伸缩

##### 添加一个节点到集群

**向集群中添加一个新的masteri节点，并向其中存储num=10**

需求：

- 启动一个新的redis.实例，端口为7004
 - <img src=" 笔记/Redis/图片/image-20230210143105670.png" alt="image-20230210143105670" style="zoom:67%;" /> 
- 添加7004到之前的集群，并作为一个master节点
 - <img src=" 笔记/Redis/图片/image-20230210143136833.png" alt="image-20230210143136833" style="zoom:67%;" /> 
- 给7004节点分配插槽，使得num这个key可以存储到7004实例
 - <img src=" 笔记/Redis/图片/image-20230210143302108.png" alt="image-20230210143302108" style="zoom:67%;" /> 
 - <img src=" 笔记/Redis/图片/image-20230210143452900.png" alt="image-20230210143452900" style="zoom:67%;" /> 
 - <img src=" 笔记/Redis/图片/image-20230210143523623.png" alt="image-20230210143523623" style="zoom:67%;" />

#### 故障转移

在7002这个slave节点执行**手动故障转移**，重新夺回master:地位
步骤如下：

1. 利用redis-cli连接7002这个节点
2. 执行cluster failovert命令

#### RedisTemplate访问分片集群

### 多级缓存

#### 传统缓存的问题

传统的缓存策略一般是请求到达 Tomcat 后，先查询 Redis ,如果未命中则查询数据库，存在下面的问题：

- 请求要经过Tomcat处理，Tomcat的性能成为整个系统的瓶颈
- Redis缓存失效时，会对数据库产生冲击

#### 多级缓存方案

多级缓存就是充分利用请求处理的每个环节，分别添加缓存，减轻Tomcat压力，提升服务性能：

用作缓存的Nginx是业务Nginx,需要部署为集群，再有专门的Nginx用来做反向代理：

#### ***JVM进程缓存

##### 导入商品案例 P114

安装 mysql ，导入工程

##### **初识Caffeine

**本地进程缓存**

缓存在日常开发中启动至关重要的作用，由于是存储在内存中，数据的读取速度是非常快的，能大量减少对数据库的访问，减少数据库的压力。我们把缓存分为两类：

- 分布式缓存，例如Redis:
  - 优点：存储容量更大、可靠性更好、可以在 **集群间共享**
  - 缺点：访问缓存有 **网络开销**
  - 场景：缓存数据量较大、可靠性要求较高、需要在集群间共享
- 进程本地缓存，例如HashMap、GuavaCache:
  - 优点：读取本地内存，没有网络开销，**速度更快**
  - 缺点：存储容量有限、可靠性较低、**无法共享**
  - 场景：性能要求较高，缓存数据量较小

**Caffeine**是一个基于刊 java8 开发的，提供了近乎最佳命中率的高性能的本地缓存库。目前Spring内部的缓存使用的就是Caffeine。GitHub地址：https://github.com/ben-manes/caffeine

```xml

<dependency>
 <groupId>com.github.ben-manes.caffeine</groupId>
 <artifactId>caffeine</artifactId>
 <version>2.9.3</version>
</dependency>
```

> For Java 11 or above, use `3.x` otherwise use `2.x`.

**Caffeine 的驱逐策略**

##### 实现进程缓存

**实现商品的查询的本地进程缓存**

利用**Caffeine**实现下列需求：

- 给根据id查询商品的业务添加缓存，缓存未命中时查询数据库
- 给根据id查询商品库存的业务添加缓存，缓存未命中时查询数据库
- 缓存初始大小为100
- 缓存上限为10000

**配置 Caffeine**

<img src=" 笔记/Redis/图片/image-20230211112057661.png" alt="image-20230211112057661" style="zoom:50%;" /> 通过注入，使用 get 方法即可

#### Lua语法入门

##### 初识Lua

Lua是一种轻量小巧的脚本语言，用标准C语言编写并以源代码形式开放，其设计目的是为了嵌入应用程序中，从而为应用程序提供灵话的扩展和定制功能。官网：https://www.lua.org/

##### 变量和循环

##### 条件控制、函数

```lua
function printArr(arr)
    for index,value in ipairs(arr)do
    	print(value)
    end
end
```

#### 多级缓存

##### 安装OpenResty P121

OpenResty是一个基于Nginx的高性能Web平台，用于方便地搭建能够处理超高并发、扩展性极高的动态Web应用、Web服务和动态网关。具备下列特点：

- 具备Nginx的完整功能
- 基于Lua语言进行扩展，集成了大量精良的Lua库、第三方模块
- 允许使用Lua自定义业务逻辑、自定义库

官方网站：https://openresty.org/cn/

##### **OpenRest快速入门

**实现商品详情页数据查询**

**步骤一**：修改nginx.conf文件

**步骤二**：编写item.lua文件

##### 请求参数处理

**获取请求路径中的商品id信息，拼接到json结果中返回**

```lua
--获取路径参数
local id = ngx.var[1]
--返回结果
ngx.say ('{"id":'.. id ..'}')
```

##### ***查询Tomcat

**获取请求路径中的商品id信息，根据id向Tomcat查询商品信息**

这里要修改item.lua,满足下面的需求：

1. 获取请求参数中的id
2. 根据id向Tomcat服务发送请求，查询商品信息
3. 根据id向Tomcat服务发送请求，查询库存信息
4. 组装商品信息、库存信息，序列化为SON格式并返回

**nginx内部发送Http请求**

nginx提供了内部API用以发送http请求：

**返回的响应内容包括**：

- resp.status:响应状态码
- resp.header:响应头，是一个table
- resp.body:响应体，就是响应数据

**注意**：这里的path是路径，并**不包含lP和端口**。这个请求会被nginx内部的server监听并处理。

**封装http查询的函数**

我们可以把http查询的请求封装为一个函数，放到OpenResty函数库中，方便后期使用。

**JSON结果处理**

OpenResty 提供了一个 cjson 的模块用来处理 JSON 的序列化和反序列化。

官方地址：https://github.com/openresty/lua-cjson/

**使用Http函数查询数据**

> item.lua

```lua
--导入common函数库
local common = require('common')
local read_http = common.read_http
--导入 cjson 库
local cjson = require('cjson')

--获取路径参数
local id = ngx.var[1]

--查询商品信息
local itemJSON = read_http("/item/" .. id, nil)
--查询库存信息
local stockJSON = read_http("/item/stock/" .. id, nil)

-- JSON 转为 lua 的 table
local item = cjosn.decode(itemJSON)
local stock = cjson.decoode(stockJSON)
--组合数据
item.stock = stock.stock
item.sold = stock.sold

-- 把 item 序列化为 JSON 返回结果
ngx.say(cjson.encode(item))
```

**Tomcat:集群的负载均衡**

> hash ：表示会根据请求路径url 进行hash运算，使得同一个访问地址指向同一个 tomcat 服务器

##### **Redis缓存预热

**冷启动与缓存预热**

**冷启动**：服务刚刚启动时，Redis中并没有缓存，如果所有商品数据都在第一次查询时添加缓存，可能会给数据库带来较大压力。

**缓存预热**：在实际开发中，我们可以利用大数据统计用户访问的热点数据，在项目启动时将这些热点数据提前查询并保存到Redis中。

> 我们数据量较少，可以在启动时将所有数据都放入缓存中。

```java
private static final ObjectMapper MAPPER = new objectMapper();

@Override
public void afterPropertiesSet()throws Exception {
 //初始化缓存
 //1.查询商品信息
 List<Item> itemList = itemService.list();
 //2.放入缓存
 for (Item item : itemList) {
 //2.1.item 序列化为切 JSON
 String json = MAPPER.writeValueAsString(item);
 //2.2.存入redis
 redisTemplate.opsForValue().set("item:id:" + item.getId(),json);
 }
}

```

> **注意** ：需要实现 InitializingBean 接口，重写 afterPropertiesSet() 方法，会在 RedisHandler 创建后，成员变量完成初始化后 执行

##### ***查询Redis缓存

**OpenResty的 Redis 模块**

> common.lua

```lua
--导入redis
local redis = require('resty.redis')
--初始化redis
Local red = redis:new()
red:set_timeouts(1000, 1000, 1000)

--关闭redis连接的工具方法，其实是放入连接池
local function close_redis(red)
    local pool_max_idLe_time = l0000 --连接的空闲时间，单位是毫秒
    local poo1_size = 100 --连接池大小
    Local ok,err = red:set_keepalive(pool_max_idle_time, pool_size)
    if not ok then
    	ngx.log(ngx.ERR,"放入redis连接池失败："，err)
    end
end

--查询redis的方法ip和port是redis,地址，key是查询的key
local function read_redis(ip,port,key)
    --获取一个连接
    local ok,err = red:connect(ip,port)
    if not ok then
        ngx.log(ngx.ERR,"连接redis:失败："，err)
        return nil
    end
    --查询redis
    Local resp,err = red:get(key)
    --查询失败处理
    if not resp then
    	ngx.log(ngx.ERR,"查询Redis失败："，err,",key=",key)
    end
    --得到的数据为空处理
    if resp == ngx.null then
    	resp = nil
    	ngx.log(ngx.ERR,"查询Redis数据为空，key=",key)
    end
    close_redis(red)
    return resp
end

--封装函数，发送http请求，并解析响应
Local function = read_http(path,params)
    local resp = ngx.location.capture(path,{
    method = ngx.HTTP_GET,
    args = params,
    })
    if not resp then
        --记录错误信息，返回404
        ngx.log(ngx.ERR,"http查询失败，path:",path,",args:",args)
        ngx.exit(404)
    end
    return resp.body
end
--一将方法导出
local _M = {
    read_http = read_http,
    read_redis = read_redis
}
return _M
```

**查询商品时，优先Redis缓存查询**

**需求**：

修改item.lua,封装一个函数read data,实现先查询Redis,如果未命中，再查询tomcat

修改item.lua,查询商品和库存时都调用read data这个函数

##### **Nginx本地缓存

OpenResty为 Nginx 提供了 **shard dict** 的功能，可以在 nginx 的多个worker之间共享数据，实现缓存功能。

**在查询商品时，优先查询OpenResty的本地缓存**

需求：

- 修改item.lua中的read_data函数，**优先查询本地缓存**，未命中时再查询Redis、Tomcat
- 查询 Redis 或 Tomcat 成功后，将数据写入本地缓存，并**设置有效期**
- 商品基本信息，有效期30分钟
- 库存信息，有效期1分钟

#### **缓存同步策略

##### 数据同步策略

缓存数据同步的常见方式有三种：

- **设置有效期**：给缓存设置有效期，到期后自动删除。再次查询时更新
  - 优势：简单、方便
  - 缺点：**时效性差**，缓存过期之前可能不一致
  - 场景：**更新频率较低**，时效性要求低的业务
- **同步双写**：在修改数据库的同时，直接修改缓存
  - 优势：时效性强，缓存与数据库强一致
  - 缺点：有代码侵入，耦合度高，
  - 场景：对一致性、时效性要求较高的缓存数据
- **异步通知**：修改数据库时发送事件通知，相关服务监听到通知后修改缓存数据
  - 优势：低耦合，可以同时通知多个缓存服务
  - 缺点：时效性一般，可能存在中间不一致状态
  - 场景：时效性要求一般，有多个服务需要同步

##### 安装Canal

**初识Canal**

Canal,译意为水道/管道/沟渠，canal是阿里巴巴旗下的一款开源项目，基于刊 java 开发。基于数据库增量日志解析，提供增量数据订阅&消费。GitHub的地址：https://github.com/alibaba/canal

Canal就是把自己 **伪装成MySQL的一个slave节点** ，从而**监听masterl的binary log变化**。再把得到的变化信息通知给Canal的客户端，进而完成对其它数据库的同步。

**安装和配置Canal** P131

1. 开启 MySQL 主从
   1. 开启bin_log
   2. 设置用户权限
2. 安装 Canal
   1. 创建网络
   2. 安装 Canal

##### 监听Canal

**Canal客户端**

Canal 提供了各种语言的客户端，当Canall监听到 bin_log 变化时，会通知 Canal 的客户端。

不过这里我们会使用GitHub上的第三方开源的canal-starter。地址：https://github.com/NormanGyllenhaal/canal-client

**编写监听器，监听 Canal 消息：**

Canal推送给canal-client的是被修改的这一行数据(row）,而我们引入的canal-client则会帮我们把行数据封装到 Item 实体类中。这个过程中需要知道数据库与实体的映射关系，要用到到PA的几个注解：

### **总结

![image-20230213192944134]( 笔记/Redis/图片/image-20230213192944134.png)

## 实践篇

### Redis键值设计

#### 优雅的key结构

Redis的Key虽然可以自定义，但最好遵循下面的几个最佳实践约定：

- 遵循基本格式：[业务名称]：[数据名]：[id]
- 长度不超过44字节
- 不包含特殊字符

#### 拒绝BigKey

BigKey通常以**Key的大小和Key中成员的数量**来综合判定，例如：

- Key本身的数据量过大：一个String类型的Key,它的值为5MB。
- Key中的成员数过多：一个ZSET类型的Key,它的成员数量为10,000个。
- Key中成员的数据量过大：一个Hash类型的Key,它的成员数量虽然只有1,000个,但这些成员的Value(值)总大小为100MB。

**查询一个 key 的准确大小**

**看元素长度**

**Big Key的危害**

- 网络阻塞

对BigKey执行读请求时，少量的QPS就可能导致带宽使用率被占满，导致Redis实例，乃至所在物理机变慢

- 数据倾斜

BigKey所在的Redis实例内存使用率远超其他实例，无法使数据分片的内存资源达到均衡

- Redis阻塞

对元素较多的hash、list、Zset等做运算会耗时较旧，使主线程被阻塞

- CPU压力

对BigKey的数据序列化和反序列化会导致CPU的使用率飙升，影响Redis实例和本机其它应用

**如何发现BigKey**

- redis-cli --bigkeys

利用redis-cli提供的-bigkeys参数，可以遍历分析所有key,并返回Key的整体统计信息与**每个数据的Top1的big key**

- scan扫描

自己编程，利用scan扫描Redist中的所有key,利用strlen、hlen等命令判断key的长度（此处不建议使用MEMORY USAGE)

- 第三方工具

利用第三方工具，如 Redis-Rdb-Tools 分析RDB快照文件，全面分析内存使用情况

- 网络监控

自定义工具，监控进出Redis的网络数据，超出预警值时主动告警

**如何删除 BigKey**

#### 恰当的数据类型

#### 总结

Key的最佳实践：

- 固定格式：[业务名]:[数据名]:[id]
- 足够简短：不超过44字节
- 不包含特殊字符

Value的最佳实践：

- 合理的拆分数据，拒绝BigKey
- 选择合适数据结构
- Hash 结构的 entry 数量不要超过1000(**默认配置为 500**)
- 设置合理的超时时间

### 批处理优化

#### Pipeline

> 结论：批量执行的效率大大提高

**MSET**

Redis提供了很多MXX这样的命令，可以实现批量插入数据，例如：

- mset
- hmset

> 不要在一次批处理中传输太多命令，否则单次命令**占用宽带过多，会导致网络阻塞**

> 不同
>
> Pipeline没有限制，可以**任何数据结构做组合**
>
> mset 是 redis 原生操作，具有原子性，会一次执行完毕。

**总结**

批量处理的方案：

1. 原生的M操作
2. Pipeline批处理

**注意事项：**

1. 批处理时不建议一次携带太多命令
2. **Pipeline的多个命令之间不具备原子性**

#### 集群下的批处理

如 MSET 或 Pipeline 这样的批处理需要在一次请求中携带多条命令，而此时如果 **Redis 是一个集群**，那批处理命令的**多个key必须落在一个插槽**中，否则就会导致执行失败。

> **Spring 已经提供 并行slot 的方式实现批处理，在集群模式下**
>
> StringRedisTemplate中，multiSet 方法，已实现并行slot方法
>
> 首先会按照 传递数组中的 key值(entrySet)，来计算 插槽 得到 map<Integer，List<K>> partitioned，然后遍历 map，将插槽一样的数据放入一个 map 集合中并开启异步任务，依次遍历完 partitioned 中的所有插槽。

### 服务端优化

#### 持久化配置

Rdis的持久化虽然可以保证数据安全，但也会带来很多额外的开销，因此持久化请遵循下列建议：

#### 慢查询

慢查询：在Redis执行时耗时超过某个阈值的命令，称为慢查询。

#### 命令及安全配置

Redis会绑定在0.0.0.0:6379，这样将会将Redis服务暴露到公网上，而Redis如果没有做身份认证，会出现严重的安全漏洞.漏洞重现方式：https://cloud.tencent.com/developer/article/1039000

漏洞出现的核心的原因有以下几点：

- Redis服务暴露到公网
- Redis未设置密码
- 利用了Redis的config seti命令动态修改Redis配置
- 使用了Root账号权限启动Redis

为了避免这样的漏洞，这里给出一些建议：

1. Redis一定要设置密码

2. 禁止线上使用下面命令：keys、flushall、flushdb、config set等命令。可以**利用rename-command禁用**。

3. bind:限制网卡，禁止外网网卡访问

4. 开启防火墙

5. 不要使用Root账户启动Redis

6. 尽量不是有默认的端口

#### 内存配置

当Redis内存不足时，可能导致Key频繁被删除、响应时间变长、QPS不稳定等问题。当内存使用率达到90%以上时就需要我们警惕，并快速定位到内存占用的原因。

**数据内存的问题**

Redis:提供了一些命令，可以查看到Redis目前的内存分配状态：

- info memory
- memory xxx

**内存缓存区配置**

内存缓存区常见的有三种：

- **复制缓冲区：	**主从复制的 repl_backlog_buf，如果太小可能导致频繁的全量复制，影响性能。通过 repl_backlog_size 来设置，默认 1mb
- **AOF 缓冲区：**AOF 刷盘之前的缓存区域，AOF 执行 rewrite 的缓冲区。无法设置容量上限
- **客户端缓冲区：**分为输入缓冲区和输出缓冲区，输入缓冲区最大1G且不能设置。输出缓冲区可以设置

### **集群最佳实践

集群虽然具备高可用特性，能实现自动故障恢复，但是如果使用不当，也会存在一些问题：

1. 集群完整性问题

在Rdis的**默认配置**中，如果发现**任意一个插槽不可用，则整个集群都会停止对外服务**：

为了保证高可用特性，这里建议将cluster-require-full-coverage配置为 no

2. 集群带宽问题

**集群节点之间**会不断的**互相Ping**来确定集群中其它节点的状态。每次Ping携带的信息至少包括：

- 插槽信息
   - 集群状态信息

集群中节点越多，集群状态信息数据量也越大，10个节点的相关信息可能达到1kb,此时每次集群互通需要的带宽会非常高。

解决途径：

1. 避免大集群，集群节点数不要太多，最好少于1000，如果业务庞大，则建立多个集群。
   2. 避免在单个物理机中运行太多Redis实例
   3. 配置合适的cluster--node-timeout值

3. 数据倾斜问题

4. 客户端性能问题

5. 命令的集群兼容性问题

mset 的 key 问题，插槽是否在一个结点上

6. lua和事务问题

mset 的 key 问题，插槽是否在一个结点上

## **原理篇

### 数据结构

#### **动态字符串SDS

我们都知道Redis中保存的Key是字符串，value往往是字符串或者字符串的集合。可见字符串是Redis中最常用的一种数据结构。

Redis构建了一种新的字符串结构，称为**简单动态字符串**（Simple Dynamic String）,简称 **SDS**。

#### IntSet：底层 整数数组

IntSet是Redist中set集合的一种实现方式，基于**整数数组**来实现，并且具备 **长度可变、有序** 等特征。

---

**IntSet 升级**

**总结**

Intset可以看做是特殊的整数数组，具备一些特点：

1. Redis会确保Intset中的元素**唯一、有序**
2. 具备类型**升级机制**，可以节省内存空间
3. 底层采用**二分查找**方式来查询

#### **Dict：底层 hash 表

我们知道Redis是一个键值型(Key-Value Pair)的数据库，我们可以根据键实现快速的增删改查。而键与值的映射关系正是通过Dict来实现的。

Dict由三部分组成，分别是：哈希表(DictHashTable)、哈希节点（DictEntry)、字典(Dict)

> size 必须为 **2 的 n 次幂**，用于得到掩码 做 & 运算(和 % 的效果一致)。

当我们向Dict添加键值对时，Redis首先根据 key 计算出hash值(h)，然后利用 **h & sizemask** 来计算元素应该存储到数组中的哪个索引位置。

---

**Dict 的扩容**

![image-20230217231608202]( 笔记/Redis/图片/image-20230217231608202.png)

---

**Dict 的收缩**

![image-20230217231648887]( 笔记/Redis/图片/image-20230217231648887.png)

---

**Dict 的 rehash**

![image-20230217234901463]( 笔记/Redis/图片/image-20230217234901463.png)

> 由于为需要主线程操作，若是一次性复制完，十分影响性能，故采用**渐进式rehsh**

![image-20230217235532470]( 笔记/Redis/图片/image-20230217235532470.png)

> **每次增伤改查，只迁移旧数组中的一列**

![image-20230217235841015]( 笔记/Redis/图片/image-20230217235841015.png)

---

**总结**

Dict的结构:

- 类似 java 的 HashTable，底层是**数组加链表**来解决哈希冲突
- Dict包含两个哈希表，ht[0]平常用，**ht[1]用来rehash**

Dict的伸缩;

- 当LoadFactor 大于5或者 LoadFactor 大于1并且没有子进程任务时，Dict 扩容
- 当LoadFactor小于0.1时，Dict收缩
- 扩容大小为第一个大于等于used + 1的 2^n
- 收缩大小为第一个大于等于used 的 2^n
- Dict采用**渐进式 rehash**，每次访问 Dict 时执行一次rehash
- rehash 时 ht[0] 只减不增，**新增操作只在 ht[1] 执行**，**其它操作在两个哈希表**

#### ZipList：连续内存块，双端，省内存

zipList 是一种特殊的 “双端链表”，由一系列特殊编码的 **连续内存块组成 **。**可以在任意一端进行压入/弹出操作**,并且该操作的时间复杂度为O(1)。

---

![image-20230218141412021]( 笔记/Redis/图片/image-20230218141412021.png)

---

**ZipList 的连锁更新问题**

ZipList的每个Entry都包含previous_entry_length来记录上一个节点的大小，长度是1个或5个字节:

- 如果前一节点的长度小于254字节，则采用1个字节来保存这个长度值
- 如果前一节点的长度大于等于254字节，则采用5个字节来保存这个长度值，第一个字节为0xfe，后四个字节才是真实长度数据

> **注意：**并未解决，因为发生的概率十分低，必须连续的字节长度为 250~253 的 entry

**总结**

ZipList特性:

1. 压缩列表的可以看做一种连续内存空间的"双向链表"
2. 列表的节点之间不是通过指针连接，而是记录上一节点和本节点长度来寻址，内存占用较低
3. 如果列表数据过多，导致链表过长，可能影响查询性能
4. 增或删较大数据时**有可能发生连续更新问题**

#### QuickList：是结点为 ZipList 的双端链表

![image-20230218210853769]( 笔记/Redis/图片/image-20230218210853769.png)

**总结**

QuickList的特点:

- 是**一个节点为 ZipList 的双端链表**
- 节点采用 ZipList，解决了传统链表的内存占用问题
- 控制了 ZipList 大小，解决连续内存空间申请效率问题
- **中间节点可以压缩**，进一步节省了内存

#### SkipList：双向链表，查询效率高

SkipList**（跳表）**首先是链表，但与传统链表相比有几点差异:

- 元素按照**升序**排列存储
- 节点可能 包含多个指针 ，指针跨度不同。

![image-20230219112127711]( 笔记/Redis/图片/image-20230219112127711.png)

**总结**

skipList的特点:

- 跳跃表是一个**双向链表**，每个节点都包含 score 和 ele 值
- 节点按照 **score 值排序**，score 值一样则按照 ele 字典排序
- 每个节点都可以包含**多层指针**，层数是1到32之间的随机数
- 不同层指针到下一个节点的跨度不同，**层级越高，跨度越大**
- 增删改查效率与红黑树基本一致，实现却更简单

#### RedisObject

Redis中的任意数据类型的**键和值**都会**被封装**为一个RedisObject，也叫做Redis对象，源码如下:

**encoding**

#### 五种数据结构

##### String：RedisObj / SDS

String是Redis中最常见的数据存储类型:

- 其基本编码方式是 **RAW**，基于 *简单动态字符串(SDS）*实现，存储上限为512mb。

- 如果存储的 *SDS 长度小于44字节*，则会采用 **EMBSTR** 编码，此时object head与SDS是一段连续空间。申请内存时只需要调用一次内存分配函数，效率更高。

- <img src=" 笔记/Redis/图片/image-20230219133630328.png" alt="image-20230219133630328" style="zoom:33%;" />

- > 原因：Redis 内部内存分配，每次分配的空间为 2 的 n 次幂

- 如果存储的字符串是**整数值**，并且大小在 LONG_MAX 范围内，则会采用 **INT** 编码：直接将数据保存在 RedisObject 的 *ptr 指针位置*（刚好8字节)，不再需要SDS了。

- <img src=" 笔记/Redis/图片/image-20230219220613165.png" alt="image-20230219220613165" style="zoom: 45%;" />

##### List：QuickList

Redis的List类型可以从首、尾操作列表中的元素

- LinkedList:普通链表，可以从双端访问，内存占用较高，内存碎片较多.

- ZipList :压缩列表，可以从双端访问，内存占用低，存储上限低
-  QuickList : LinkedList + ZipList，可以从双端访问，内存占用较低，包含多个ZipList，存储上限高
- 在3.2版本之前，Redis采用ZipList和LinkedList来实现List，当元素数量小于512并且元素大小小于64字节时采用zipList编码，超过则采用LinkedList编码。
- 在3.2版本之后，Redis统一采用QuickList来实现List

##### Set：Dict

Set是Redis中的单列集合，满足下列特点:

- 不保证有序性
- 保证元素唯一（可以判断元素是否存在)
-  求交集、并集、差集

可以看出，Set对查询元素的效率要求非常高，思考一下，什么样的数据结构可以满足?

- HashTable，也就是Redis中的Dict，不过Dict是双列集合（可以存键、值对）

Set是Redis中的集合，不一定确保元素有序，*可以满足元素唯一、查询效率要求极高*

- 为了查询效率和唯一性，set采用HT编码（Dict)。Dict中的 key 用来存储元素 value 统一为 null。
- 当存储的所有数据都是 *整数*，并且元素数量不超过set-max-intset-entries时，Set会采用IntSet编码，以节省内存。

##### ZSet：ZipList  /  SkipList+ Dict

ZSet也就是SortedSet，其中每一个元素都需要指定一个score值和member值:

- 可以根据score值排序后
- member必须唯—
- 可以根据member查询分数

因此，zset底层数据结构必须满足 **键值存储、键必须唯一、可排序** 这几个需求。之前学习的哪种编码结构可以满足?

- **SkipList**: 可以排序，并且可以同时存储 score 和 ele 值(member)
- **HT (Dict)** : 可以键值存储，并且可以根据 key 找 value

---

当**元素数量不多**时，HT 和 SkipList 的优势不明显，而且更耗内存。因此zset还会采用 **ZipList 结构**来节省内存，不过需要同时满足两个条件:

1. 元素数量小于 zset_max_ziplist_entries，默认值128
2. 每个元素都小于 zset_max_ziplist_value 字节，默认值64

**ziplist本身没有排序功能**，而且没有键值对的概念，因此需要有zset通过编码实现:

- ZipList 是连续内存，因此 score 和 element 是紧挨在一起的两个 entry，element 在前 score 在后

- score越小越接近队首，score越大越接近队尾，**按照score值升序排列**

##### Hash：ZipList / Dict

Hash结构与Redis中的Zset非常类似:

- 都是键值存储
- 都需求根据键获取值
- 键必须唯一

**区别**如下:

- zset的键是member，值是score; hash的键和值都是任意值

- zset要根据score排序; hash则无需排序

因此，Hash底层采用的编码与Zset也基本一致，只需要把排序有关的SkipList去掉即可:

- Hash结构**默认采用ZipList编码**，用以节省内存。ZipList中相邻的两个entry分别保存field和value

- 当数据量较大时，Hash结构会转为HT编码，也就是Dict，触发条件有两个:

1. ZipList中的元素数量超过了hash-max-ziplist-entries（默认512)

在 插入元素后 进行判断，若超过，则执行转换成 HT

2. ZipList中的任意entry大小超过了hash-max-ziplist-value （默认64字节)

在 插入元素前 就先进行一次集体插入元素的判断，若中间出现超过的情况，则执行转换；全部判断完，在执行插入指令

### **网络模型

#### 用户空间和内核空间

**用户态 和 内核态 切换**

#### 阻塞lO

#### 非阻塞IO

#### IO多路复用

无论是阻塞IO还是非阻塞lO，用户应用在一阶段都需要调用 recvfrom 来获取数据，差别在于无数据时的处理方案:

- 如果调用recvfrom时，恰好**没有**数据，阻塞IO会使进程阻塞，非阻塞IO使CPU空转，都不能充分发挥CPU的作用。
- 如果调 recvfrom时，恰好**有**数据，则用户进程可以直接进入第二阶段，读取并处理数据

比如服务端处理客户端Socket请求时，在单线程情况下，只能依次处理每一个socket，如果正在处理的socket恰好未就绪（数据不可读或不可写），线程就会被阻塞，所有其它客户端socket都必须等待，性能自然会很差。

那么问题来了:用户进程 **如何知道内核中数据是否就绪** 呢?

**文件描述符**（File Descriptor):简称FD，是一个从0开始递增的无符号整数，用来关联Linux中的一个文件。在Linux中，一切皆文件，例如常规文件、视频、硬件设备等，当然也包括网络套接字(Socket)。

**IO多路复用**︰是利用单个线程来同时监听多个FD，并在某个FD可读、可写时得到通知，从而避免无效的等待，充分利用CPU资源

不过监听FD的方式、通知的方式又有多种实现，常见的有:

- select
- poll
- epoll

**差异**:

- *select 和 poll* 只会通知用户进程有FD就绪，但不确定具体是哪个FD，需要用户进程**逐个遍历FD**来确认
- *epoll* 则会在通知用户进程FD就绪的同时，把已就绪的 FD 写入用户空间

##### Select

![image-20230220233542732]( 笔记/Redis/图片/image-20230220233542732.png)

> 用户态 和 核心态 内存空间不共享

##### poll

##### epoll

**区别**

select：每次调用，都得把 监听的fd数组，拷贝到内核；结果也需要全部拷贝到用户空间。

epoll：

- 将select功能拆分，直接操作需要监听的fd(epoll_ctl)，减少了fd数据的拷贝(如：直接删除已经监听到的fd，而select需要删除后对应的数据后，将新的整个fd数组拷贝到内核)
- 只拷贝就绪的 fd
- 使用了红黑树，查询性能变化幅度不大

##### 总结

**select**模式存在的三个问题:

- 能监听的FD最大不超过1024
- 每次 select 都需要  **把所有要监听的FD都拷贝到内核空间**
- 每次都要  **遍历所有FD**  来判断就绪状态

**poll**模式的问题:

- poll利用  **链表**  解决了 select 中监听 FD 上限的问题，但  **依然要遍历**  所有FD，如果监听较多，性能会下降(**理论上无限**，需要遍历)

**epoll**模式中如何解决这些问题的?

- 基于 epoll 实例中的  **红黑树**  保存要监听的FD，理论上无上限，而且增删改查效率都非常高，*性能**不会**随监听的FD数量增多而下降*
- 每个FD**只需要执行一次epoll_ctl添加到红黑树**，以后每次epoll_wait无需传递任何参数，**无需重复拷贝FD到内核空间**
- 内核会将**就绪的FD直接拷贝到用户空间的指定位置**，用户进程**无需遍历**所有FD就能知道就绪的FD是谁

##### 事件通知机制

当FD有数据可读时，我们调用epoll_wait就可以得到通知。但是事件通知的模式有两种:

- LevelTriggered:简称LT。当FD有数据可读时，会重复通知多次，直至数据处理完成。是Epoll的默认模式。

- EdgeTriggered:简称ET。当FD有数据可读时，只会被通知一次，不管数据是否处理完成

**结论**:

- ET模式避免了LT模式可能出现的**惊群现象**(多个进程同时监听一个 fd，其实只需要一个进程进行读取就行，LT每次重复通知都会唤醒所有的进程)
- **ET模式**最好**结合非阻塞IO读取FD数据**，相比LT会复杂一些 (性能较好)

##### Web 服务流程

#### 信号驱动lO

#### 异步lO

#### **Redis网络模型

**Redis到底是单线程还是多线程?**

- 如果仅仅聊Redis的  **核心业务部分**（命令处理)，答案是**单线程**
- 如果是聊**整个Redis**，那么答案就是**多线程**

在Redis版本迭代过程中，在两个重要的时间节点上引入了多线程的支持:

- Redis v4.0:引入多线程异步处理一些耗时较长的任务，例如异步删除命令**unlink**
- Redis v6.0∶在**核心网络模型中引入 多线程**，进一步提高对于多核CPU的利用率

1. **为什么Redis要选择单线程?**

- 抛开持久化不谈，Redis是 **纯内存操作，执行速度非常快，**它的**性能瓶颈是网络延迟**而不是执行速度，因此多线程并不会带来巨大的性能提升。
- 多线程会导致过多的**上下文切换**，带来不必要的**开销**
- 引入多线程会面临 **线程安全问题**，必然要引入 **线程锁** 这样的安全手段，实现复杂度增高，而且性能也会大打折扣

##### 单线程网络模型

来看下Redis  **单线程网络模型**  的整个流程:

服务器初始化：

---

---

##### 多线程网络模型

![image-20230222012230114]( 笔记/Redis/图片/image-20230222012230114.png)

### 通信协议

#### RESP协议

Redis是一个**CS架构**的软件，通信一般分两步（不包括pipeline和PubSub) :

1. 客户端（client）向服务端( server)发送一条命令
2. 服务端解析并执行命令，返回响应结果给客户端

因此客户端发送命令的格式、服务端响应结果的格式必须有一个规范，这个规范就是通信协议。

而在Redis中采用的是RESP ( Redis Serialization Protocol）协议:

- Redis 1.2版本引入了RESP协议
- Redis 2.0版本中成为与Redis服务端通信的标准，称为RESP2
- Redis 6.0版本中，从RESP2升级到了RESP3协议，增加了更多数据类型并且支持6.0的新特性--客户端缓存

但目前，**默认使用的依然是RESP2协议**，也是我们要学习的协议版本（以下简称RESP）。

##### 数据类型

#### 模拟Redis客户端

```java
public class Main {
    static Socket s;
    static PrintWriter writer;
    static BufferedReader reader;
    public static void main ( String[] args) {
        try {
            //1.建立连接
            String host = "192.168.150.101";
            int port = 6379;
            s = new Socket(host,port);   
            //2.获取输出流、输入流
            writer = new Printwriter(new OutputStreamwriter( s.getOutputStream(),StandardCharsets.UTF_8));
            reader = new BufferedReader(new InputStreamReader(s.getInputStream()，StandardCharsets.UTF_8));

//3.发出请求
 //3.1.获取授权
 sendRequest("auth", "123456");
 Object obj = handleResponse();
 System.out.println( "obj = " + obj);
 
 //3.2.set name 虎哥
 sendRequest("set", "name", "yoshino");
 
 //4.解析响应
 obj = handleResponse();
 System.out.println( "obj = " + obj);
 
 sendRequest("get", "name");
 Object obj = handleResponse();
 System.out.println( "obj = " + obj);
 } catch (IOException e) {
 e.printStackTrace();
 } finally {
 //5.释放连接
 try {
 if ( reader != null) reader.close();
 if (writer != null) writer.close();
 if (s != null) s.close();
 }catch (IOException e) {
 e.printStackTrace();
 }
 }
 }
 
 private static Object handleResponse() throws IOException {
 //读取首字节
		int prefix = reader.read();
 //判断数据类型标示
		switch (prefix) {
 case '+'://单行字符串，直接读一行
 	return reader.readLine();
 case '-'://异常，也读一行
 	throw new RuntimeException(reader.readLine());
 case ':'://数字
 	return Long.parseLong( reader.readLine());
 case '$'://多行字符串
 	//先读长度
 int len = Integer.parseInt(reader.readLine());
 if ( len == -1){
 		return null;
 }
 if (len == 0){
 	return "";
 }
 //再读数据，读len个字节。我们假设没有特殊字符，所以读一行（简化）
 return reader.readLine();
 case '*':
 	return readBulkString();
 default:
 	throw new RuntimeException("错误的数据格式! ");
 }
 return null;
 }
 
 private static Object readBulkString() throws IOException {
 //获取数组大小
 int len = Integer.parseInt(reader.readLine());
 if (len <= 0) {
 		return null;
 }
 //遍历，依次读取每个元素
 List<0bject> list = new ArrayList<> (len);//遍历，依次读取每个元素
 for (int i = 0; i < len; i++){
 	list.add (handleResponse());
 }
 return list;
 }

private static void sendRequest(String ... args) {
        writer.println("*" + args.length);
        for(String arg : args) {
            writer.println("$" + arg.getBytes (StandardCharsets.UTF_8).length);
        	writer.println(arg);
        }

writer.flush();
    }   
}
```

### **内存策略

**Redis内存回收**

Redis之所以性能强，最主要的原因就是基于内存存储。然而单节点的Redis其内存大小不宜过大，会影响持久化或主从同步性能。

我们可以通过修改配置文件来设置Redis的最大内存:

```
# 格式:
# maxmemory <bytes>
#例如∶
maxmemory 1gb
```

当内存使用达到上限时，就无法存储更多数据了。

#### 过期策略

**这里有两个问题需要我们思考:**

1. Redis是如何知道一个key是否过期呢?

利用两个Dict分别记录key-value对及key-ttl对

2. 是不是TTL到期就立即删除了呢?
 - 惰性删除
 - 顾名思义并不是在TTL到期后就立刻删除，而是在 **访问一个 key ** 的时候，**检查该 key 的存活时间** ，如果已经过期才执行删除
 - <img src=" 笔记/Redis/图片/image-20230222172457776.png" alt="image-20230222172457776" style="zoom:33%;" /> 
 - 问题：已过期但一直未被访问的数据堆积
 - 周期删除
 - 通过一个 **定时任务** ，周期性的 **抽样部分过期的 key** ，然后执行删除。执行周期有两种
 - Redis会设置一个定时任务serverCron()，按照server.hz的频率来执行过期key清理，模式为SLOW
 - Redis的每个事件循环前会调用 beforeSleep() 函数，执行过期key清理，模式为FAST
 - <img src=" 笔记/Redis/图片/image-20230222175047153.png" alt="image-20230222175047153" style="zoom: 50%;" /> 
 - <img src=" 笔记/Redis/图片/image-20230222215459080.png" alt="image-20230222215459080" style="zoom:50%;" />

**DB结构**

**总结**

RedisKey的TTL记录方式:

- 在RedisDB中通过一个 Dict 记录每个 Key 的 TTL 时间

过期 key 的删除策略:

- **惰性**清理: 每次**查找 key 时**判断是否过期，如果过期则删除
- **定期**清理: **定期抽样**部分key，判断是否过期，如果过期则删除。

定期清理的两种模式:

- SLOW 模式执行频率默认为10，每次不超过25ms
- FAST 模式执行频率不固定，但两次间隔不低于2ms，每次耗时不超过1ms

#### 淘汰策略

**内存淘汰**︰就是当Redis内存使用**达到设置的阈值**时，Redis **主动挑选部分key** 删除以释放更多内存的流程。

---

**performEvictions() 方法 流程图**

![image-20230222224214047]( 笔记/Redis/图片/image-20230222224214047.png)

## TODO

1. 逻辑过期实现的缓存击穿 和 缓存穿透 没有兼容
   - 访问店铺时，调用了一个 url，可以将 访问逻辑过期实现的 url 加上 hot （原因：击穿需要预先手动添加，为如热点店铺）

## 知识点

### spring 框架事务失效问题(aop)

> 事务实现需要依赖其代理对象（由spring管理创建）中的方法
>
> 该图调用是使用 this. 获取的，事务失效

还需要额外添加：

```xml
<dependency>
 <groupId>org.aspectj</groupId>
 <artifactId>aspectjweaver</artifactId>
</dependency>
```

![image-20230124155954031]( 笔记/Redis/图片/image-20230124155954031.png)

### 获取上传文件代码

```、java
@PostMapping("blog")
public Result uploadImage(@RequestParam("file") MultipartFile image) {
    try {
        // 获取原始文件名称
        String originalFilename = image.getOriginalFilename();
        // 生成新文件名
        String fileName = createNewFileName(originalFilename);
        // 保存文件
        image.transferTo(new File(SystemConstants.IMAGE_UPLOAD_DIR, fileName));
        // 返回结果
        log.debug("文件上传成功，{}", fileName);
        return Result.ok(fileName);
    } catch (IOException e) {
        throw new RuntimeException("文件上传失败", e);
    }
}

@GetMapping("/blog/delete")
public Result deleteBlogImg(@RequestParam("name") String filename) {
    File file = new File(SystemConstants.IMAGE_UPLOAD_DIR, filename);
    if (file.isDirectory()) {
        return Result.fail("错误的文件名称");
    }
    FileUtil.del(file);
    return Result.ok();
}

private String createNewFileName(String originalFilename) {
    // 获取后缀
    String suffix = StrUtil.subAfter(originalFilename, ".", true);
    // 生成目录
    String name = UUID.randomUUID().toString();
    int hash = name.hashCode();
    int d1 = hash & 0xF;
    int d2 = (hash >> 4) & 0xF;
    // 判断目录是否存在
    File dir = new File(SystemConstants.IMAGE_UPLOAD_DIR, StrUtil.format("/blogs/{}/{}", d1, d2));
    if (!dir.exists()) {
        dir.mkdirs();
    }
    // 生成文件名
    return StrUtil.format("/blogs/{}/{}/{}.{}", d1, d2, name, suffix);
}

public class SystemConstants {
    public static final String IMAGE_UPLOAD_DIR = " \\nginx-1.18.0\\html\\hmdp\\imgs\\"; // nginx 前端服务器存图片的位置
    public static final String USER_NICK_NAME_PREFIX = "user_";
    public static final int DEFAULT_PAGE_SIZE = 5;
    public static final int MAX_PAGE_SIZE = 10;
}
```

### fork

- fork 是linux指令，创建子进程，只复制页表数据给子进程
- 主进程 - 页表虚拟内存 - 真实内存

- fork采用的是copy-on-write技术：

- 当主进程执行读操作时，访问共享内存；

- 当主进程执行写操作时，则会拷贝一份数据，执行写操作。

wengyoshino/Redis

简介

发行版

贡献者

近期动态

wengyoshino/Redis .gitee-modal { width: 500px !important; }

简介

发行版

贡献者

近期动态

搜索帮助

wengyoshino/Redis