案例参考：黑马程序员

一.分布式缓存

– 基于 Redis 集群解决单机 Redis 存在的问题

单机的 Redis 存在四大问题：

1.Redis 持久化

Redis 持久化分为两种

RDB 持久化
AOF 持久化

1.1 RDB 持久化

RDB 全称 Redis Database Backup file（Redis 数据备份文件），也就是快照，把内存中的所有数据存到磁盘上。故障修复后，读取磁盘文件恢复数据。快照文件为 RDB 文件，默认存在当前目录。

1.1.1 什么时候执行 RDB

有四种情况会执行 RDB

执行 save 命令
执行 bgsave 命令
redis 手动关机
触发执行 RDB 条件

save 命令

执行下面的命令，可以立即执行一次 RDB：

save 命令会导致主进程执行 RDB，这个过程中其它所有命令都会被阻塞。只有在数据迁移时可能用到。

bgsave 命令

下面的命令可以异步执行 RDB：

这个命令执行后会开启独立进程完成 RDB，主进程可以持续处理用户请求，不受影响。

手动停机 redis

触发执行 RDB 条件

Redis 内部有触发 RDB 的机制，可以在 redis.conf 文件中找到，格式如下：

# 900秒内，如果至少有1个key被修改，则执行bgsave ， 如果是save "" 则表示禁用RDB
save 900 1
save 300 10
save 60 10000

RDB 的其它配置也可以在 redis.conf 文件中设置：

# 是否压缩 ,建议不开启，压缩也会消耗cpu，磁盘的话不值钱
rdbcompression yes

# RDB文件名称
dbfilename dump.rdb

# 文件保存的路径目录
dir ./

1.1.2 RDB 原理

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

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

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

1.1.3 总结

RDB 方式 bgsave 的基本流程？

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

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

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

RDB 的缺点？

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

1.2 AOP 持久化

1.2.1 AOP 原理

AOF：Append Only File（追加文件），就是把 redis 的每一条命令都记录在 AOF 文件中，相当于命令日志文件，故障恢复后，执行每一条命令达到恢复数据的目的。

1.2.2 AOF 配置

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

# 是否开启AOF功能，默认是no
appendonly yes
# AOF文件的名称
appendfilename "appendonly.aof"

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

# 表示每执行一次写命令，立即记录到AOF文件
appendfsync always
# 写命令执行完先放入AOF缓冲区，然后表示每隔1秒将缓冲区数据写到AOF文件，是默认方案
appendfsync everysec
# 写命令执行完先放入AOF缓冲区，由操作系统决定何时将缓冲区内容写回磁盘
appendfsync no

三种策略对比：

1.2.3.AOF 文件重写

因为是记录命令，AOF 文件会比 RDB 文件大的多。而且 AOF 会记录对同一个 key 的多次写操作，但只有最后一次写操作才有意义。通过执行 bgrewriteaof 命令，可以让 AOF 文件执行重写功能，用最少的命令达到相同效果。

如图，AOF 原本有三个命令，但是set num 123 和 set num 666都是对 num 的操作，第二次会覆盖第一次的值，因此第一个命令记录下来没有意义。

所以重写命令后，AOF 文件内容就是：mset name jack num 666

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

# AOF文件比上次文件 增长超过多少百分比则触发重写
auto-aof-rewrite-percentage 100
# AOF文件体积最小多大以上才触发重写
auto-aof-rewrite-min-size 64mb

1.3.RDB 与 AOF 对比

RDB 和 AOF 各有自己的优缺点，如果对数据安全性要求较高，在实际开发中往往会结合两者来使用。

2.Redis 主从同步

2.1 搭建主从架构

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

2.1.1 准备实例和配置

要在一台虚拟机上开启 3 个实例，必须准备三个不同的配置文件和目录。

假设我们这里准备的端口为 7001，7002，7003，其中 7001 作为主节点。

这里注意修改每个实例的配置文件。

将持久化模式改为 RDB 模式
修改端口，修改数据保存位置（将dir ./改到对应的实例位置）

修改每个实例的声明 ip，因为虚拟机本身有多个 ip，避免混乱，将每个实例指定 ip

启动每个实例

2.1.2 开启主从关系

配置主从关系有临时和永久两种方法：

修改配置文件（永久生效）
- 在 redis.conf 中添加一行配置：slaveof <masterip> <masterport>
使用 redis-cli 客户端连接到 redis 服务，执行 slaveof(5.0 以前)/replicaof 命令（重启失效）:
- 该命令在从机上输入

1	slaveof slaveof <masterip> <masterport>

2.2 数据同步原理

2.2.1 全量同步

主从如果是第一次连接，那么就执行全量同步。

那么怎么判断是否为第一次连接呢？

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

这里是通过replid来判断是否为第一次连接，当连接过后，主节点就会将自己的 replid 发送给从节点，以后从节点的 replid 就和主节点一样了。具体过程：

完整流程描述：

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

2.2.2 增量同步

全量同步需要生成 RDB 文件，然后将 RDB 文件网络传输给 slave，成本过高，因此在第一次之后，其他时候大多数都是做增量同步，即只更新 master 和 slave 存在差异的部分数据。

那么 master 怎么知道 slave 与自己的数据差异在哪里呢?

这就得说说repl_baklog 文件了

repl_baklog 文件是一个环形数组，大小固定。

repl_baklog 中会记录 Redis 处理过的命令日志及 offset，包括 master 当前的 offset，和 slave 已经拷贝到的 offset：

slave 与 master 的 offset 之间的差异，就是 salve 需要增量拷贝的数据了。

随着不断有数据写入，master 的 offset 逐渐变大，slave 也不断的拷贝，追赶 master 的 offset，直到数组被填满：：

此时，如果有新的数据写入，就会覆盖数组中的旧数据。不过，旧的数据只要是绿色的，说明是已经被同步到 slave 的数据，即便被覆盖了也没什么影响。因为未同步的仅仅是红色部分。

但是，如果 slave 出现网络阻塞，导致 master 的 offset 远远超过了 slave 的 offset：

如果 master 继续写入新数据，其 offset 就会覆盖旧的数据，直到将 slave 现在的 offset 也覆盖：

棕色框中的红色部分，就是尚未同步，但是却已经被覆盖的数据。此时如果 slave 恢复，需要同步，却发现自己的 offset 都没有了，无法完成增量同步了。只能做全量同步。

2.3 主从同步优化

主从同步可以保证主从数据的一致性，非常重要。

可以从以下几个方面来优化 Redis 主从就集群：

在 master 中配置 repl-diskless-sync yes 启用无磁盘复制，避免全量同步时的磁盘 IO。
Redis 单节点上的内存占用不要太大，减少 RDB 导致的过多磁盘 IO
适当提高 repl_baklog 的大小，发现 slave 宕机时尽快实现故障恢复，尽可能避免全量同步
限制一个 master 上的 slave 节点数量，如果实在是太多 slave，则可以采用主-从-从链式结构，减少 master 压力

主从从架构图：

2.4 小结

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

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

什么时候执行全量同步？

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

什么时候执行增量同步？

slave 节点断开又恢复，并且在 repl_baklog 中能找到 offset 时

3.Redis 哨兵模式

3.1 哨兵模式原理

3.1.1 哨兵集群结构及作用

哨兵作用：

监控：Sentinel 会不断检查您的 master 和 slave 是否按预期工作
自动故障恢复：如果 master 故障，Sentinel 会将一个 slave 提升为 master。当故障实例恢复后也以新的 master 为主
通知：Sentinel 充当 Redis 客户端的服务发现来源，当集群发生故障转移时，会将最新信息推送给 Redis 的客户端

3.1.2 监控原理

Sentinel 基于心跳机制监测服务状态，每隔 1 秒向集群的每个实例发送 ping 命令：

主观下线：如果某 sentinel 节点发现某实例未在规定时间响应，则认为该实例主观下线。
客观下线：若超过指定数量（quorum）的 sentinel 都认为该实例主观下线，则该实例客观下线。quorum 值最好超过 Sentinel 实例数量的一半。

3.1.3 故障恢复原理

一旦发现 master 故障，sentinel 需要在 salve 中选择一个作为新的 master，选择依据是这样的：

首先会判断 slave 节点与 master 节点断开时间长短，如果超过指定值（down-after-milliseconds * 10）则会排除该 slave 节点
然后判断 slave 节点的 slave-priority 值，越小优先级越高，如果是 0 则永不参与选举
如果 slave-prority 一样，则判断 slave 节点的 offset 值，越大说明数据越新，优先级越高
最后是判断 slave 节点的运行 id 大小，越小优先级越高。

当选出一个新的 master 后，该如何实现切换呢？

流程如下：

sentinel 给备选的 slave1 节点发送slaveof no one命令，让该节点成为 master
sentinel 给所有其它 slave 发送 slaveof 192.168.150.101 7002 命令，让这些 slave 成为新 master 的从节点，开始从新的 master 上同步数据。
最后，sentinel 将故障节点标记为 slave，当故障节点恢复后会自动成为新的 master 的 slave 节点

先让新皇登基，然后让旧皇俯首。

3.1.4.小结

Sentinel 的三个作用是什么？

监控
故障转移
通知

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

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

故障转移步骤有哪些？

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

3.2 搭建哨兵集群

我们搭建的三个 sentinel 实例信息如下：

节点	IP	PORT
s1	192.168.150.101	27001
s2	192.168.150.101	27002
s3	192.168.150.101	27003

3.2.1 准备实例和配置

创建三个文件夹，名字分别叫 s1、s2、s3

# 进入/tmp目录
cd /tmp
# 创建目录
mkdir s1 s2 s3

在 s1 目录创建一个 sentinel.conf 文件，添加下面的内容：

port 27001
sentinel announce-ip 192.168.150.101
sentinel monitor mymaster 192.168.150.101 7001 2
sentinel down-after-milliseconds mymaster 5000
sentinel failover-timeout mymaster 60000
dir "/tmp/s1"

解读：

port 27001：是当前 sentinel 实例的端口
sentinel monitor mymaster 192.168.150.101 7001 2：指定主节点信息
- mymaster：主节点名称，自定义，任意写
- 192.168.150.101 7001：主节点的 ip 和端口
- 2：选举 master 时的 quorum 值

这里只需要指定主节点的 ip 和端口即可，他会自动找到其他的子节点。

将 sentinel.conf 拷贝到其他两个目录中，并且修改端口和文件目录。

（在/tmp 目录执行下列命令）：

# 方式一：逐个拷贝
cp s1/sentinel.conf s2
cp s1/sentinel.conf s3
# 方式二：管道组合命令，一键拷贝
echo s2 s3 | xargs -t -n 1 cp s1/sentinel.conf

修改 s2、s3 两个文件夹内的配置文件，将端口分别修改为 27002、27003：

1 2	sed -i -e 's/27001/27002/g' -e 's/s1/s2/g' s2/sentinel.conf sed -i -e 's/27001/27003/g' -e 's/s1/s3/g' s3/sentinel.conf

启动

# 第1个
redis-sentinel s1/sentinel.conf
# 第2个
redis-sentinel s2/sentinel.conf
# 第3个
redis-sentinel s3/sentinel.conf

3.2.2 测试

将 7001 宕机，发现其中一个从节点继承上来。

3.3 RedisTemplate

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

3.3.1 配置文件

我们配置了哨兵，则 redis 的 ip 就是动态的，所以我们只需要配置哨兵的 ip 即可。

spring:
  redis:
    sentinel:
      master: mymaster
      nodes:
        - 192.168.150.101:27001
        - 192.168.150.101:27002
        - 192.168.150.101:27003

3.3.2 配置读写分离

在 redis 的配置类中，添加一个新的 bean：

@Bean
public LettuceClientConfigurationBuilderCustomizer clientConfigurationBuilderCustomizer(){
    return clientConfigurationBuilder -> clientConfigurationBuilder.readFrom(ReadFrom.REPLICA_PREFERRED);
}

我们要主节点写操作，从节点读操作。

这个 bean 中配置的就是读写策略，包括四种：

MASTER：从主节点读取
MASTER_PREFERRED：优先从 master 节点读取，master 不可用才读取 replica
REPLICA：从 slave（replica）节点读取
REPLICA _PREFERRED：优先从 slave（replica）节点读取，所有的 slave 都不可用才读取 master

4.Redis 分片集群

4.1 为什么要分片集群？

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

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

那么我们就需要使用分片集群来解决以上问题：

分片集群特征：

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

4.2 搭建分片集群

需要准备的实例：

IP	PORT	角色
192.168.150.101	7001	master
192.168.150.101	7002	master
192.168.150.101	7003	master
192.168.150.101	8001	slave
192.168.150.101	8002	slave
192.168.150.101	8003	slave

4.2.1 准备实例与配置

删除之前的 7001、7002、7003 这几个目录，重新创建出 7001、7002、7003、8001、8002、8003 目录：

# 进入/tmp目录
cd /tmp
# 删除旧的，避免配置干扰
rm -rf 7001 7002 7003
# 创建目录
mkdir 7001 7002 7003 8001 8002 8003

在/tmp 下准备一个新的 redis.conf 文件，内容如下：

port 6379
# 开启集群功能
cluster-enabled yes
# 集群的配置文件名称，不需要我们创建，由redis自己维护
cluster-config-file /tmp/6379/nodes.conf
# 节点心跳失败的超时时间
cluster-node-timeout 5000
# 持久化文件存放目录
dir /tmp/6379
# 绑定地址
bind 0.0.0.0
# 让redis后台运行
daemonize yes
# 注册的实例ip
replica-announce-ip 192.168.150.101
# 保护模式
protected-mode no
# 数据库数量
databases 1
# 日志
logfile /tmp/6379/run.log

# 进入/tmp目录
cd /tmp
# 执行拷贝
echo 7001 7002 7003 8001 8002 8003 | xargs -t -n 1 cp redis.conf

修改每个目录下的 redis.conf，将其中的 6379 修改为与所在目录一致：

# 进入/tmp目录
cd /tmp
# 修改配置文件
printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t sed -i 's/6379/{}/g' {}/redis.conf

启动

# 进入/tmp目录
cd /tmp
# 一键启动所有服务
printf '%s\n' 7001 7002 7003 8001 8002 8003 | xargs -I{} -t redis-server {}/redis.conf

4.2.2 创建集群

在 Redis5.0 之前创建集群比较麻烦，5.0 之后集群管理命令都集成到了 redis-cli 中。

Redis5.0 之前

Redis5.0 之前集群命令都是用 redis 安装包下的 src/redis-trib.rb 来实现的。因为 redis-trib.rb 是有 ruby 语言编写的所以需要安装 ruby 环境。

1
2
3

# 安装依赖
yum -y install zlib ruby rubygems
gem install redis

然后通过命令来管理集群：

# 进入redis的src目录
cd /tmp/redis-6.2.4/src
# 创建集群
./redis-trib.rb create --replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003

Redis5.0 以后

我们使用的是 Redis6.2.4 版本，集群管理以及集成到了 redis-cli 中，格式如下：

1	redis-cli --cluster create --cluster-replicas 1 192.168.150.101:7001 192.168.150.101:7002 192.168.150.101:7003 192.168.150.101:8001 192.168.150.101:8002 192.168.150.101:8003

命令说明：

redis-cli --cluster或者./redis-trib.rb：代表集群操作命令
create：代表是创建集群
--replicas 1或者--cluster-replicas 1 ：指定集群中每个 master 的副本个数为 1，此时节点总数 ÷ (replicas + 1) 得到的就是 master 的数量。因此节点列表中的前 n 个就是 master，其它节点都是 slave 节点，随机分配到不同 master

通过命令可以查看集群状态：

1	redis-cli -p 7001 cluster nodes

集群操作时，需要给redis-cli加上-c参数才可以。

4.3 散列插槽

4.3.1.插槽原理

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

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

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

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

如图，在 7001 这个节点执行 set a 1 时，对 a 做 hash 运算，对 16384 取余，得到的结果是 15495，因此要存储到 103 节点。

到了 7003 后，执行get num时，对 num 做 hash 运算，对 16384 取余，得到的结果是 2765，因此需要切换到 7001 节点

4.3.1.小结

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

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

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

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

4.4 集群伸缩

添加节点的命令：

4.4.1 需求分析

需求：向集群中添加一个新的 master 节点，并向其中存储 num = 10

启动一个新的 redis 实例，端口为 7004
添加 7004 到之前的集群，并作为一个 master 节点
给 7004 节点分配插槽，使得 num 这个 key 可以存储到 7004 实例

这里需要两个新的功能：

添加一个节点到集群中
将部分插槽分配到新插槽

4.4.2 创建新的 redis 实例

创建一个文件夹：

1	mkdir 7004

拷贝配置文件：

1	cp redis.conf /7004

修改配置文件：

1	sed /s/6379/7004/g 7004/redis.conf

启动

1	redis-server 7004/redis.conf

4.4.3 添加新节点到 redis

添加节点的语法如下：

执行命令：

1	redis-cli --cluster add-node 192.168.150.101:7004 192.168.150.101:7001

通过命令查看集群状态：

1	redis-cli -p 7001 cluster nodes

如图，7004 加入了集群，并且默认是一个 master 节点：

但是，可以看到 7004 节点的插槽数量为 0，因此没有任何数据可以存储到 7004 上

4.4.4 转移插槽

我们要将 num 存储到 7004 节点，因此需要先看看 num 的插槽是多少：

如上图所示，num 的插槽为 2765.

我们可以将 0~3000 的插槽从 7001 转移到 7004，命令格式如下：

具体命令如下：

建立连接：

得到下面的反馈：

询问要移动多少个插槽，我们计划是 3000 个：

新的问题来了：

那个 node 来接收这些插槽？？

显然是 7004，那么 7004 节点的 id 是多少呢？

复制这个 id，然后拷贝到刚才的控制台后：

这里询问，你的插槽是从哪里移动过来的？

all：代表全部，也就是三个节点各转移一部分
具体的 id：目标节点的 id
done：没有了

这里我们要从 7001 获取，因此填写 7001 的 id：

填完后，点击 done，这样插槽转移就准备好了：

确认要转移吗？输入 yes：

然后，通过命令查看结果：

可以看到：

目的达成。

4.5 故障转移

集群初识状态是这样的：

其中 7001、7002、7003 都是 master，我们计划让 7002 宕机。

4.5.1 自动故障转移

当集群中有一个 master 宕机会发生什么呢？

直接停止一个 redis 实例，例如 7002：

1	redis-cli -p 7002 shutdown

首先是该实例与其它实例失去连接
然后是疑似宕机：

最后是确定下线，自动提升一个 slave 为新的 master：

当 7002 再次启动，就会变为一个 slave 节点了：

4.5.2 手动故障转移

利用 cluster failover 命令可以手动让集群中的某个 master 宕机，切换到执行 cluster failover 命令的这个 slave 节点，实现无感知的数据迁移。其流程如下：

这种 failover 命令可以指定三种模式：

缺省：默认的流程，如图 1~6 步
force：省略了对 offset 的一致性校验
takeover：直接执行第 5 歩，忽略数据一致性、忽略 master 状态和其它 master 的意见

案例需求：在 7002 这个 slave 节点执行手动故障转移，重新夺回 master 地位

步骤如下：

1）利用 redis-cli 连接 7002 这个节点

2）执行 cluster failover 命令

如图：

效果：

4.6 RedisTemplate 访问分片集群

RedisTemplate 底层同样基于 lettuce 实现了分片集群的支持，而使用的步骤与哨兵模式基本一致：

引入 redis 的 starter 依赖
配置分片集群地址
配置读写分离

与哨兵模式相比，其中只有分片集群的配置方式略有差异，如下：

spring:
  redis:
    cluster:
      nodes:
        - 192.168.150.101:7001
        - 192.168.150.101:7002
        - 192.168.150.101:7003
        - 192.168.150.101:8001
        - 192.168.150.101:8002
        - 192.168.150.101:8003

二.多级缓存

该部分省略了很多细节，建议看视频，这里只是方便复习。

视频：高级篇-多级缓存-01-什么是多级缓存_哔哩哔哩_bilibili

1.什么是多级缓存？

1.1 传统缓存及存在的问题

传统缓存：

请求到达 Tomcat
查 redis
1. 命中，返回数据
2. 未命中，查询数据库，返回数据

存在的问题：

所有请求都要先到达 Tomcat，Tomcat 的性能决定了整个系统的性能。
Redis 缓存失效时，会对数据库造成冲击

1.2 多级缓存是什么

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

浏览器访问静态资源时，优先读取浏览器本地缓存
访问非静态资源（ajax 查询数据）时，访问服务端
请求到达 Nginx 后，优先读取 Nginx 本地缓存
如果 Nginx 本地缓存未命中，则去直接查询 Redis（不经过 Tomcat）
如果 Redis 查询未命中，则查询 Tomcat
请求进入 Tomcat 后，优先查询 JVM 进程缓存
如果 JVM 进程缓存未命中，则查询数据库

在多级缓存架构中，Nginx 内部需要编写本地缓存查询、Redis 查询、Tomcat 查询的业务逻辑，因此这样的 nginx 服务不再是一个反向代理服务器，而是一个编写业务的 Web 服务器了。

因此Nginx 服务也需要搭建集群来提高并发，再用一个专门的 nginx 来做反向代理，另外，我们的Tomcat 服务将来也会部署为集群模式：

可见，多级缓存的关键有两个：

一个是在 nginx 中编写业务，实现 nginx 本地缓存、Redis、Tomcat 的查询（OpenResty 框架结合 Lua 语言）
另一个就是在Tomcat 中实现 JVM 进程缓存

2.JVM 进程缓存

2.1 初识 Caffeine

GitHub 地址：https://github.com/ben-manes/caffeine

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

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

Caffeine是一个基于 Java8 开发的，提供了近乎最佳命中率的高性能的本地缓存库。目前 Spring 内部的缓存使用的就是 Caffeine。

2.2 基本 API 及缓存驱逐策略

缓存使用的基本 API：

缓存不过就是存取数据
存数据cache.put(String key,String val)
取数据
- cache.getIfPresent(String key),有数据返回数据，无数据返回 null
- cache.get(String key,Lambda表达式),有数据则返回数据，没有数据通过后面的表达式查数据库并写入缓存，返回数据

@Test
void testBasicOps() {
    // 构建cache对象
    Cache<String, String> cache = Caffeine.newBuilder().build();

    // 存数据
    cache.put("gf", "迪丽热巴");

    // 取数据
    String gf = cache.getIfPresent("gf");
    System.out.println("gf = " + gf);

    // 取数据，包含两个参数：
    // 参数一：缓存的key
    // 参数二：Lambda表达式，表达式参数就是缓存的key，方法体是查询数据库的逻辑
    // 优先根据key查询JVM缓存，如果未命中，则执行参数二的Lambda表达式
    String defaultGF = cache.get("defaultGF", key -> {
        // 根据key去数据库查询数据
        return "柳岩";
    });
    System.out.println("defaultGF = " + defaultGF);
}

Caffeine 提供了三种缓存驱逐策略：

基于容量：设置缓存的数量上限

// 创建缓存对象
Cache<String, String> cache = Caffeine.newBuilder()
    .maximumSize(1) // 设置缓存大小上限为 1
    .build();

基于时间：设置缓存的有效时间

// 创建缓存对象
Cache<String, String> cache = Caffeine.newBuilder()
    // 设置缓存有效期为 10 秒，从最后一次写入开始计时
    .expireAfterWrite(Duration.ofSeconds(10))
    .build();

基于引用：设置缓存为软引用或弱引用，利用 GC 来回收缓存数据。性能较差，不建议使用。

注意：在默认情况下，当一个缓存元素过期的时候，Caffeine 不会自动立即将其清理和驱逐。而是在一次读或写操作后，或者在空闲时间完成对失效数据的驱逐。

2.3 使用 Caffeine

一般在配置类中将某张表的需要缓存作为 bean 注入 Spring 容器中，在配置类中生命好缓存驱逐策略。

定义两个 Caffeine 的缓存对象，分别保存商品、库存的缓存数据。

在 item-service 的com.heima.item.config包下定义CaffeineConfig类：

package com.heima.item.config;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.heima.item.pojo.Item;
import com.heima.item.pojo.ItemStock;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class CaffeineConfig {

    @Bean
    public Cache<Long, Item> itemCache(){
        return Caffeine.newBuilder()
                .initialCapacity(100)
                .maximumSize(10_000)
                .build();
    }

    @Bean
    public Cache<Long, ItemStock> stockCache(){
        return Caffeine.newBuilder()
                .initialCapacity(100)
                .maximumSize(10_000)
                .build();
    }
}

改造业务代码：利用 APIcache.get()，来获取数据，即可在查询时顺便写入缓存

3.Lua 语法入门

视频：高级篇-多级缓存-06-Lua 语法-初识 Lua_哔哩哔哩_bilibili

Nginx 编程需要用到 Lua 语言，因此我们必须先入门 Lua 的基本语法。

4.实现多级缓存

多级缓存的实现离不开 Nginx 编程，而 Nginx 编程又离不开 OpenResty。

4.1 OpenResty 安装及快速入门

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

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

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

视频链接：高级篇-多级缓存-09-多级缓存-安装 OpenResty_哔哩哔哩_bilibili

4.2 请求参数处理

如何获取前端传递的商品参数呢？

4.2.1.获取参数的 API

OpenResty 中提供了一些 API 用来获取不同类型的前端请求参数：

4.3 查询 Tomcat

由 Nginx 转发到 Tomcat 集群的时候默认的负载均衡是轮询，那么就会导致一个请求会在多个 Tomcat 服务器缓存，并且需要在轮询完所有服务器才能真正的利用到缓存，缓存命中率太低，这是非常不好的。

所以我们需要改变 nginx 的负载均衡策略，让同一个请求每次都能转发到同一台 Tomcat 服务器。

4.3.1 原理

nginx 提供了基于请求路径做负载均衡的算法：

nginx 根据请求路径做 hash 运算，把得到的数值对 tomcat 服务的数量取余，余数是几，就访问第几个服务，实现负载均衡。

例如：

我们的请求路径是 /item/10001
tomcat 总数为 2 台（8081、8082）
对请求路径/item/1001 做 hash 运算求余的结果为 1
则访问第一个 tomcat 服务，也就是 8081

只要 id 不变，每次 hash 运算结果也不会变，那就可以保证同一个商品，一直访问同一个 tomcat 服务，确保 JVM 缓存生效。

4.3.2 实现

修改/usr/local/openresty/nginx/conf/nginx.conf文件，实现基于 ID 做负载均衡。

首先，定义 tomcat 集群，并设置基于路径做负载均衡：

upstream tomcat-cluster {
    hash $request_uri;
    server 192.168.150.1:8081;
    server 192.168.150.1:8082;
}

然后，修改对 tomcat 服务的反向代理，目标指向 tomcat 集群：

1
2
3

location /item {
    proxy_pass http://tomcat-cluster;
}

重新加载 OpenResty

1	nginx -s reload

4.4 Redis 缓存预热

Redis 缓存会面临冷启动问题：

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

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

4.4.1 编码实现

缓存预热需要在项目启动时完成，并且必须是拿到 RedisTemplate 之后。

这里我们利用 InitializingBean 接口来实现，因为InitializingBean 可以在对象被 Spring 创建并且成员变量全部注入后执行。

@Component
public class RedisHandler implements InitializingBean {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private IItemService itemService;
    @Autowired
    private IItemStockService stockService;

    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);
        }

        // 3.查询商品库存信息
        List<ItemStock> stockList = stockService.list();
        // 4.放入缓存
        for (ItemStock stock : stockList) {
            // 2.1.item序列化为JSON
            String json = MAPPER.writeValueAsString(stock);
            // 2.2.存入redis
            redisTemplate.opsForValue().set("item:stock:id:" + stock.getId(), json);
        }
    }
}

4.6.查询 Redis 缓存

具体操作：高级篇-多级缓存-16-多级缓存-查询 Redis_哔哩哔哩_bilibili

现在，Redis 缓存已经准备就绪，我们可以再 OpenResty 中实现查询 Redis 的逻辑了。如下图红框所示：

当请求进入 OpenResty 之后：

优先查询 Redis 缓存
如果 Redis 缓存未命中，再查询 Tomcat

4.7.Nginx 本地缓存

具体操作：高级篇-多级缓存-17-多级缓存-nginx 本地缓存_哔哩哔哩_bilibili

现在，整个多级缓存中只差最后一环，也就是 nginx 的本地缓存了。如图：

5.缓存同步

大多数情况下，浏览器查询到的都是缓存数据，如果缓存数据与数据库数据存在较大差异，可能会产生比较严重的后果。

所以我们必须保证数据库数据、缓存数据的一致性，这就是缓存与数据库的同步。

5.1.数据同步策略

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

设置有效期：给缓存设置有效期，到期后自动删除。再次查询时更新

优势：简单、方便
缺点：时效性差，缓存过期之前可能不一致
场景：更新频率较低，时效性要求低的业务

同步双写：在修改数据库的同时，直接修改缓存

优势：时效性强，缓存与数据库强一致
缺点：有代码侵入，耦合度高；
场景：对一致性、时效性要求较高的缓存数据

异步通知：修改数据库时发送事件通知，相关服务监听到通知后修改缓存数据

优势：低耦合，可以同时通知多个缓存服务
缺点：时效性一般，可能存在中间不一致状态
场景：时效性要求一般，有多个服务需要同步

而异步实现又可以基于 MQ 或者 Canal 来实现：

1）基于 MQ 的异步通知：

解读：

商品服务完成对数据的修改后，只需要发送一条消息到 MQ 中。
缓存服务监听 MQ 消息，然后完成对缓存的更新

依然有少量的代码侵入。

2）基于 Canal 的通知

解读：

商品服务完成商品修改后，业务直接结束，没有任何代码侵入
Canal 监听 MySQL 变化，当发现变化后，立即通知缓存服务
缓存服务接收到 canal 通知，更新缓存

代码零侵入

5.2.安装 Canal

5.2.1.认识 Canal

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

Canal 是基于 mysql 的主从同步来实现的，MySQL 主从同步的原理如下：

1）MySQL master 将数据变更写入二进制日志( binary log），其中记录的数据叫做 binary log events
2）MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
3）MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

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

5.2.2.安装 Canal

1.开启 MySQL 主从

Canal 是基于 MySQL 的主从同步功能，因此必须先开启 MySQL 的主从功能才可以。

这里以之前用 Docker 运行的 mysql 为例：

1.1.开启 binlog

打开 mysql 容器挂载的日志文件，我的在/tmp/mysql/conf目录:

修改文件：

1	vi /tmp/mysql/conf/my.cnf

添加内容：

1 2	log-bin=/var/lib/mysql/mysql-bin binlog-do-db=heima

配置解读：

log-bin=/var/lib/mysql/mysql-bin：设置 binary log 文件的存放地址和文件名，叫做 mysql-bin
binlog-do-db=heima：指定对哪个 database 记录 binary log events，这里记录 heima 这个库

最终效果：

[mysqld]
skip-name-resolve
character_set_server=utf8
datadir=/var/lib/mysql
server-id=1000
log-bin=/var/lib/mysql/mysql-bin
binlog-do-db=heima

1.2.设置用户权限

接下来添加一个仅用于数据同步的账户，出于安全考虑，这里仅提供对 heima 这个库的操作权限。

1
2
3

create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%' identified by 'canal';
FLUSH PRIVILEGES;

重启 mysql 容器即可

1	docker restart mysql

测试设置是否成功：在 mysql 控制台，或者 Navicat 中，输入命令：

1	show master status;

2.安装 Canal

2.1.创建网络

我们需要创建一个网络，将 MySQL、Canal、MQ 放到同一个 Docker 网络中：

1	docker network create heima

让 mysql 加入这个网络：

1	docker network connect heima mysql

2.3.安装 Canal

将 canal 的镜像压缩包上传到虚拟机，然后通过命令导入：

1	docker load -i canal.tar

然后运行命令创建 Canal 容器：

docker run -p 11111:11111 --name canal \
-e canal.destinations=heima \
-e canal.instance.master.address=mysql:3306  \
-e canal.instance.dbUsername=canal  \
-e canal.instance.dbPassword=canal  \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false  \
-e canal.instance.filter.regex=heima\\..* \
--network heima \
-d canal/canal-server:v1.1.5

说明:

-p 11111:11111：这是 canal 的默认监听端口
-e canal.instance.master.address=mysql:3306：数据库地址和端口，如果不知道 mysql 容器地址，可以通过docker inspect 容器id来查看
-e canal.instance.dbUsername=canal：数据库用户名
-e canal.instance.dbPassword=canal ：数据库密码
-e canal.instance.filter.regex=：要监听的表名称

表名称监听支持的语法：

mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\)
常见例子：
1.  所有表：.*   or  .*\\..*
2.  canal schema下所有表： canal\\..*
3.  canal下的以canal打头的表：canal\\.canal.*
4.  canal schema下的一张表：canal.test1
5.  多个规则组合使用然后以逗号隔开：canal\\..*,mysql.test1,mysql.test2

5.3.监听 Canal

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

我们可以利用 Canal 提供的 Java 客户端，监听 Canal 通知消息。当收到变化的消息时，完成对缓存的更新。

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

与 SpringBoot 完美整合，自动装配，比官方客户端要简单好用很多。

5.3.1.引入依赖：

<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

5.3.2.编写配置：

1
2
3

canal:
  destination: heima # canal的集群名字，要与安装canal时设置的名称一致
  server: 192.168.150.101:11111 # canal服务地址

5.3.3.修改 Item 实体类

通过@Id、@Column、等注解完成 Item 与数据库表字段的映射：

package com.heima.item.pojo;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.annotation.Transient;

import javax.persistence.Column;
import java.util.Date;

@Data
@TableName("tb_item")
public class Item {
    @TableId(type = IdType.AUTO)
    @Id
    private Long id;//商品id
    @Column(name = "name")
    private String name;//商品名称
    private String title;//商品标题
    private Long price;//价格（分）
    private String image;//商品图片
    private String category;//分类名称
    private String brand;//品牌名称
    private String spec;//规格
    private Integer status;//商品状态 1-正常，2-下架
    private Date createTime;//创建时间
    private Date updateTime;//更新时间
    @TableField(exist = false)
    @Transient
    private Integer stock;
    @TableField(exist = false)
    @Transient
    private Integer sold;
}

5.3.4.编写监听器

通过实现EntryHandler<T>接口编写监听器，监听 Canal 消息。注意两点：

实现类通过@CanalTable("tb_item")指定监听的表信息
EntryHandler 的泛型是与表对应的实体类

package com.heima.item.canal;

import com.github.benmanes.caffeine.cache.Cache;
import com.heima.item.config.RedisHandler;
import com.heima.item.pojo.Item;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

@CanalTable("tb_item")
@Component
public class ItemHandler implements EntryHandler<Item> {

    @Autowired
    private RedisHandler redisHandler;
    @Autowired
    private Cache<Long, Item> itemCache;

    @Override
    public void insert(Item item) {
        // 写数据到JVM进程缓存
        itemCache.put(item.getId(), item);
        // 写数据到redis
        redisHandler.saveItem(item);
    }

    @Override
    public void update(Item before, Item after) {
        // 写数据到JVM进程缓存
        itemCache.put(after.getId(), after);
        // 写数据到redis
        redisHandler.saveItem(after);
    }

    @Override
    public void delete(Item item) {
        // 删除数据到JVM进程缓存
        itemCache.invalidate(item.getId());
        // 删除数据到redis
        redisHandler.deleteItemById(item.getId());
    }
}

在这里对 Redis 的操作都封装到了 RedisHandler 这个对象中，是我们之前做缓存预热时编写的一个类，内容如下：

package com.heima.item.config;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.heima.item.pojo.Item;
import com.heima.item.pojo.ItemStock;
import com.heima.item.service.IItemService;
import com.heima.item.service.IItemStockService;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class RedisHandler implements InitializingBean {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private IItemService itemService;
    @Autowired
    private IItemStockService stockService;

    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);
        }

        // 3.查询商品库存信息
        List<ItemStock> stockList = stockService.list();
        // 4.放入缓存
        for (ItemStock stock : stockList) {
            // 2.1.item序列化为JSON
            String json = MAPPER.writeValueAsString(stock);
            // 2.2.存入redis
            redisTemplate.opsForValue().set("item:stock:id:" + stock.getId(), json);
        }
    }

    public void saveItem(Item item) {
        try {
            String json = MAPPER.writeValueAsString(item);
            redisTemplate.opsForValue().set("item:id:" + item.getId(), json);
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    public void deleteItemById(Long id) {
        redisTemplate.delete("item:id:" + id);
    }
}

三.最佳实践

参考视频：黑马程序员

1. Redis 的键值设计

1.1 优雅的 key 结构

Redis 的 key 最好遵循以下最佳实践约定：

遵循基本格式：[业务名称]:[数据名]:[id]，保证可读性和可管理性
长度不超过 44 字节，保证简洁性
不包含特殊字符，包含空格、换行、单双引号以及其他转义字符

例如：登陆业务，保存用户信息

1.2 拒绝 BigKey

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

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

通过以下命令可以判断元素大小：

推荐值：

单个 key 的 value 小于 10KB
对于集合类型的 key，建议元素数量小于 1000

1.2.1 BigKey 的危害

网络阻塞
- 对 BigKey 执行读请求时，少量的 QPS 就可能导致带宽使用率被占满，导致 Redis 实例，乃至所在物理机变慢
数据倾斜
- BigKey 所在的 Redis 实例内存使用率远超其他实例，无法使数据分片的内存资源达到均衡
Redis 阻塞
- 对元素较多的 hash、list、zset 等做运算会耗时较旧，使主线程被阻塞
CPU 压力
- 对 BigKey 的数据序列化和反序列化会导致 CPU 的使用率飙升，影响 Redis 实例和本机其它应用

1.2.2 如何发现 BigKey

①redis-cli –bigkeys

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

命令：redis-cli -a 密码 --bigkeys

②scan 扫描

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

scan 命令调用完后每次会返回 2 个元素，第一个是下一次迭代的光标，第一次光标会设置为 0，当最后一次 scan 返回的光标等于 0 时，表示整个 scan 遍历结束了，第二个返回的是 List，一个匹配的 key 的数组

import com.heima.jedis.util.JedisConnectionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.ScanResult;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class JedisTest {
    private Jedis jedis;

    @BeforeEach
    void setUp() {
        // 1.建立连接
        // jedis = new Jedis("192.168.150.101", 6379);
        jedis = JedisConnectionFactory.getJedis();
        // 2.设置密码
        jedis.auth("123321");
        // 3.选择库
        jedis.select(0);
    }

    final static int STR_MAX_LEN = 10 * 1024;
    final static int HASH_MAX_LEN = 500;

    @Test
    void testScan() {
        int maxLen = 0;
        long len = 0;

        String cursor = "0";
        do {
            // 扫描并获取一部分key
            ScanResult<String> result = jedis.scan(cursor);
            // 记录cursor
            cursor = result.getCursor();
            List<String> list = result.getResult();
            if (list == null || list.isEmpty()) {
                break;
            }
            // 遍历
            for (String key : list) {
                // 判断key的类型
                String type = jedis.type(key);
                switch (type) {
                    case "string":
                        len = jedis.strlen(key);
                        maxLen = STR_MAX_LEN;
                        break;
                    case "hash":
                        len = jedis.hlen(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    case "list":
                        len = jedis.llen(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    case "set":
                        len = jedis.scard(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    case "zset":
                        len = jedis.zcard(key);
                        maxLen = HASH_MAX_LEN;
                        break;
                    default:
                        break;
                }
                if (len >= maxLen) {
                    System.out.printf("Found big key : %s, type: %s, length or size: %d %n", key, type, len);
                }
            }
        } while (!cursor.equals("0"));
    }

    @AfterEach
    void tearDown() {
        if (jedis != null) {
            jedis.close();
        }
    }

}

③ 第三方工具

利用第三方工具，如 Redis-Rdb-Tools 分析 RDB 快照文件，全面分析内存使用情况
https://github.com/sripathikrishnan/redis-rdb-tools

④ 网络监控

自定义工具，监控进出 Redis 的网络数据，超出预警值时主动告警
一般阿里云搭建的云服务器就有相关监控页面

1.2.3 如何删除 BigKey

BigKey 内存占用较多，即便时删除这样的 key 也需要耗费很长时间，导致 Redis 主线程阻塞，引发一系列问题。

redis 3.0 及以下版本
- 如果是集合类型，则遍历 BigKey 的元素，先逐个删除子元素，最后删除 BigKey

Redis 4.0 以后
- Redis 在 4.0 后提供了异步删除的命令：unlink

非字符串的 bigkey，不要使用 del 删除，使用 hscan、sscan、zscan 方式渐进式删除，同时要注意防止 bigkey 过期时间自动删除问题(例如一个 200 万的 zset 设置 1 小时过期，会触发 del 操作，造成阻塞，而且该操作不会不出现在慢查询中(latency 可查))，查找方法和删除方法。

1.3 恰当的数据类型

例 1：比如存储一个 User 对象，我们有三种存储方式：

① 方式一：json 字符串

user:1	{“name”: “Jack”, “age”: 21}

优点：实现简单粗暴

缺点：数据耦合，不够灵活

② 方式二：字段打散

user:1:name	Jack
user:1:age	21

优点：可以灵活访问对象任意字段

缺点：占用空间大、没办法做统一控制

③ 方式三：hash（推荐）

user:1	name	jack
user:1	age	21

优点：底层使用 ziplist，空间占用小，可以灵活访问对象的任意字段

缺点：代码相对复杂

例 2：假如有 hash 类型的 key，其中有 100 万对 field 和 value，field 是自增 id，这个 key 存在什么问题？如何优化？

key	field	value
someKey	id:0	value0
	.....	.....
	id:999999	value999999

存在的问题：

hash 的 entry 数量超过 500 时，会使用哈希表而不是 ZipList，内存占用较多
可以通过 hash-max-ziplist-entries 配置 entry 上限。但是如果 entry 过多就会导致 BigKey 问题

方案一

拆分为 string 类型

key	value
id:0	value0
.....	.....
id:999999	value999999

存在的问题：

string 结构底层没有太多内存优化，内存占用较多

想要批量获取这些数据比较麻烦

方案二

拆分为小的 hash，将 id / 100 作为 key，将 id % 100 作为 field，这样每 100 个元素为一个 Hash

key	field	value
key:0	id:00	value0
	.....	.....
	id:99	value99
key:1	id:00	value100
	.....	.....
	id:99	value199
....
key:9999	id:00	value999900
	.....	.....
	id:99	value999999

package com.heima.test;

import com.heima.jedis.util.JedisConnectionFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.ScanResult;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class JedisTest {
    private Jedis jedis;

    @BeforeEach
    void setUp() {
        // 1.建立连接
        // jedis = new Jedis("192.168.150.101", 6379);
        jedis = JedisConnectionFactory.getJedis();
        // 2.设置密码
        jedis.auth("123321");
        // 3.选择库
        jedis.select(0);
    }

    @Test
    void testSetBigKey() {
        Map<String, String> map = new HashMap<>();
        for (int i = 1; i <= 650; i++) {
            map.put("hello_" + i, "world!");
        }
        jedis.hmset("m2", map);
    }

    @Test
    void testBigHash() {
        Map<String, String> map = new HashMap<>();
        for (int i = 1; i <= 100000; i++) {
            map.put("key_" + i, "value_" + i);
        }
        jedis.hmset("test:big:hash", map);
    }

    @Test
    void testBigString() {
        for (int i = 1; i <= 100000; i++) {
            jedis.set("test:str:key_" + i, "value_" + i);
        }
    }

    @Test
    void testSmallHash() {
        int hashSize = 100;
        Map<String, String> map = new HashMap<>(hashSize);
        for (int i = 1; i <= 100000; i++) {
            int k = (i - 1) / hashSize;
            int v = i % hashSize;
            map.put("key_" + v, "value_" + v);
            if (v == 0) {
                jedis.hmset("test:small:hash_" + k, map);
            }
        }
    }

    @AfterEach
    void tearDown() {
        if (jedis != null) {
            jedis.close();
        }
    }
}

1.4 总结

Key 的最佳实践
- 固定格式：[业务名]:[数据名]:[id]
- 足够简短：不超过 44 字节
- 不包含特殊字符
Value 的最佳实践：
- 合理的拆分数据，拒绝 BigKey
- 选择合适数据结构
- Hash 结构的 entry 数量不要超过 1000
- 设置合理的超时时间

2. 批处理优化

2.1 Pipeline

2.1.1 我们的客户端与 redis 服务器是这样交互的

单个命令的执行流程

N 条命令的执行流程

redis 处理指令是很快的，主要花费的时候在于网络传输。于是乎很容易想到将多条指令批量的传输给 redis

2.1.2 MSet

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

mset
hmset

利用 mset 批量插入 10 万条数据

@Test
void testMxx() {
    String[] arr = new String[2000];
    int j;
    long b = System.currentTimeMillis();
    for (int i = 1; i <= 100000; i++) {
        j = (i % 1000) << 1;
        arr[j] = "test:key_" + i;
        arr[j + 1] = "value_" + i;
        if (j == 0) {
            jedis.mset(arr);
        }
    }
    long e = System.currentTimeMillis();
    System.out.println("time: " + (e - b));
}

2.1.3 Pipeline

MSET 虽然可以批处理，但是却只能操作部分数据类型，因此如果有对复杂数据类型的批处理需要，建议使用 Pipeline

@Test
void testPipeline() {
    // 创建管道
    Pipeline pipeline = jedis.pipelined();
    long b = System.currentTimeMillis();
    for (int i = 1; i <= 100000; i++) {
        // 放入命令到管道
        pipeline.set("test:key_" + i, "value_" + i);
        if (i % 1000 == 0) {
            // 每放入1000条命令，批量执行
            pipeline.sync();
        }
    }
    long e = System.currentTimeMillis();
    System.out.println("time: " + (e - b));
}

2.2 集群下的批处理

如 MSET 或 Pipeline 这样的批处理需要在一次请求中携带多条命令，而此时如果 Redis 是一个集群，那批处理命令的多个 key 必须落在一个插槽中，否则就会导致执行失败。大家可以想一想这样的要求其实很难实现，因为我们在批处理时，可能一次要插入很多条数据，这些数据很有可能不会都落在相同的节点上，这就会导致报错了

这个时候，我们可以找到 4 种解决方案

1653126446641

第一种方案：串行执行，所以这种方式没有什么意义，当然，执行起来就很简单了，缺点就是耗时过久。

第二种方案：串行 slot，简单来说，就是执行前，客户端先计算一下对应的 key 的 slot，一样 slot 的 key 就放到一个组里边，不同的，就放到不同的组里边，然后对每个组执行 pipeline 的批处理，他就能串行执行各个组的命令，这种做法比第一种方法耗时要少，但是缺点呢，相对来说复杂一点，所以这种方案还需要优化一下

第三种方案：并行 slot，相较于第二种方案，在分组完成后串行执行，第三种方案，就变成了并行执行各个命令，所以他的耗时就非常短，但是实现呢，也更加复杂。

第四种：hash_tag，redis 计算 key 的 slot 的时候，其实是根据 key 的有效部分来计算的，通过这种方式就能一次处理所有的 key，这种方式耗时最短，实现也简单，但是如果通过操作 key 的有效部分，那么就会导致所有的 key 都落在一个节点上，产生数据倾斜的问题，所以我们推荐使用第三种方式。

2.2.1 串行化执行代码实践

public class JedisClusterTest {

    private JedisCluster jedisCluster;

    @BeforeEach
    void setUp() {
        // 配置连接池
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(8);
        poolConfig.setMaxIdle(8);
        poolConfig.setMinIdle(0);
        poolConfig.setMaxWaitMillis(1000);
        HashSet<HostAndPort> nodes = new HashSet<>();
        nodes.add(new HostAndPort("192.168.150.101", 7001));
        nodes.add(new HostAndPort("192.168.150.101", 7002));
        nodes.add(new HostAndPort("192.168.150.101", 7003));
        nodes.add(new HostAndPort("192.168.150.101", 8001));
        nodes.add(new HostAndPort("192.168.150.101", 8002));
        nodes.add(new HostAndPort("192.168.150.101", 8003));
        jedisCluster = new JedisCluster(nodes, poolConfig);
    }

    @Test
    void testMSet() {
        jedisCluster.mset("name", "Jack", "age", "21", "sex", "male");

    }

    @Test
    void testMSet2() {
        Map<String, String> map = new HashMap<>(3);
        map.put("name", "Jack");
        map.put("age", "21");
        map.put("sex", "Male");
        //对Map数据进行分组。根据相同的slot放在一个分组
        //key就是slot，value就是一个组
        Map<Integer, List<Map.Entry<String, String>>> result = map.entrySet()
                .stream()
                .collect(Collectors.groupingBy(
                        entry -> ClusterSlotHashUtil.calculateSlot(entry.getKey()))
                );
        //串行的去执行mset的逻辑
        for (List<Map.Entry<String, String>> list : result.values()) {
            String[] arr = new String[list.size() * 2];
            int j = 0;
            for (int i = 0; i < list.size(); i++) {
                j = i<<2;
                Map.Entry<String, String> e = list.get(0);
                arr[j] = e.getKey();
                arr[j + 1] = e.getValue();
            }
            jedisCluster.mset(arr);
        }
    }

    @AfterEach
    void tearDown() {
        if (jedisCluster != null) {
            jedisCluster.close();
        }
    }
}

2.2.2 Spring 集群环境下批处理代码

@Test
 void testMSetInCluster() {
     Map<String, String> map = new HashMap<>(3);
     map.put("name", "Rose");
     map.put("age", "21");
     map.put("sex", "Female");
     stringRedisTemplate.opsForValue().multiSet(map);


     List<String> strings = stringRedisTemplate.opsForValue().multiGet(Arrays.asList("name", "age", "sex"));
     strings.forEach(System.out::println);

 }

原理分析

在 RedisAdvancedClusterAsyncCommandsImpl 类中

首先根据 slotHash 算出来一个 partitioned 的 map，map 中的 key 就是 slot，而他的 value 就是对应的对应相同 slot 的 key 对应的数据

通过 RedisFuture mset = super.mset(op);进行异步的消息发送

@Override
public RedisFuture<String> mset(Map<K, V> map) {

    Map<Integer, List<K>> partitioned = SlotHash.partition(codec, map.keySet());

    if (partitioned.size() < 2) {
        return super.mset(map);
    }

    Map<Integer, RedisFuture<String>> executions = new HashMap<>();

    for (Map.Entry<Integer, List<K>> entry : partitioned.entrySet()) {

        Map<K, V> op = new HashMap<>();
        entry.getValue().forEach(k -> op.put(k, map.get(k)));

        RedisFuture<String> mset = super.mset(op);
        executions.put(entry.getKey(), mset);
    }

    return MultiNodeExecution.firstOfAsync(executions);
}

3. 服务器端优化-持久化配置

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

用来做缓存的 Redis 实例尽量不要开启持久化功能
建议关闭 RDB 持久化功能，使用 AOF 持久化
利用脚本定期在 slave 节点做 RDB，实现数据备份
设置合理的 rewrite 阈值，避免频繁的 bgrewrite
配置 no-appendfsync-on-rewrite = yes，禁止在 rewrite 期间做 aof，避免因 AOF 引起的阻塞
部署有关建议：
- Redis 实例的物理机要预留足够内存，应对 fork 和 rewrite
- 单个 Redis 实例内存上限不要太大，例如 4G 或 8G。可以加快 fork 的速度、减少主从同步、数据迁移压力
- 不要与 CPU 密集型应用部署在一起
- 不要与高硬盘负载应用一起部署。例如：数据库、消息队列

4. 服务器端优化-慢查询优化

4.1 什么是慢查询

并不是很慢的查询才是慢查询，而是：在 Redis 执行时耗时超过某个阈值的命令，称为慢查询。

慢查询的危害：由于 Redis 是单线程的，所以当客户端发出指令后，他们都会进入到 redis 底层的 queue 来执行，如果此时有一些慢查询的数据，就会导致大量请求阻塞，从而引起报错，所以我们需要解决慢查询问题。

1653129590210

慢查询的阈值可以通过配置指定：

slowlog-log-slower-than：慢查询阈值，单位是微秒。默认是 10000，建议 1000

慢查询会被放入慢查询日志中，日志的长度有上限，可以通过配置指定：

slowlog-max-len：慢查询日志（本质是一个队列）的长度。默认是 128，建议 1000

1653130457771

修改这两个配置可以使用：config set 命令：

1653130475979

4.2 如何查看慢查询

知道了以上内容之后，那么咱们如何去查看慢查询日志列表呢：

slowlog len：查询慢查询日志长度
slowlog get [n]：读取 n 条慢查询日志
slowlog reset：清空慢查询列表

1653130858066

5. 服务器端优化-命令及安全配置

安全可以说是服务器端一个非常重要的话题，如果安全出现了问题，那么一旦这个漏洞被一些坏人知道了之后，并且进行攻击，那么这就会给咱们的系统带来很多的损失，所以我们这节课就来解决这个问题。

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

为什么会出现不需要密码也能够登录呢，主要是 Redis 考虑到每次登录都比较麻烦，所以 Redis 就有一种 ssh 免秘钥登录的方式，生成一对公钥和私钥，私钥放在本地，公钥放在 redis 端，当我们登录时服务器，再登录时候，他会去解析公钥和私钥，如果没有问题，则不需要利用 redis 的登录也能访问，这种做法本身也很常见，但是这里有一个前提，前提就是公钥必须保存在服务器上，才行，但是 Redis 的漏洞在于在不登录的情况下，也能把秘钥送到 Linux 服务器，从而产生漏洞

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

Redis 未设置密码
利用了 Redis 的 config set 命令动态修改 Redis 配置
使用了 Root 账号权限启动 Redis

所以：如何解决呢？我们可以采用如下几种方案

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

Redis 一定要设置密码
禁止线上使用下面命令：keys、flushall、flushdb、config set 等命令。可以利用 rename-command 禁用。
bind：限制网卡，禁止外网网卡访问
开启防火墙
不要使用 Root 账户启动 Redis
尽量不是有默认的端口

6. 服务器端优化-Redis 内存划分和内存配置

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

有关碎片问题分析

Redis 底层分配并不是这个 key 有多大，他就会分配多大，而是有他自己的分配策略，比如 8,16,20 等等，假定当前 key 只需要 10 个字节，此时分配 8 肯定不够，那么他就会分配 16 个字节，多出来的 6 个字节就不能被使用，这就是我们常说的碎片问题

进程内存问题分析：

这片内存，通常我们都可以忽略不计

缓冲区内存问题分析：

一般包括客户端缓冲区、AOF 缓冲区、复制缓冲区等。客户端缓冲区又包括输入缓冲区和输出缓冲区两种。这部分内存占用波动较大，所以这片内存也是我们需要重点分析的内存问题。

内存占用	说明
数据内存	是 Redis 最主要的部分，存储 Redis 的键值信息。主要问题是 BigKey 问题、内存碎片问题
进程内存	Redis 主进程本身运⾏肯定需要占⽤内存，如代码、常量池等等；这部分内存⼤约⼏兆，在⼤多数⽣产环境中与 Redis 数据占⽤的内存相⽐可以忽略。
缓冲区内存	一般包括客户端缓冲区、AOF 缓冲区、复制缓冲区等。客户端缓冲区又包括输入缓冲区和输出缓冲区两种。这部分内存占用波动较大，不当使用 BigKey，可能导致内存溢出。

于是我们就需要通过一些命令，可以查看到 Redis 目前的内存分配状态：

info memory：查看内存分配的情况

1653132073570

memory xxx：查看 key 的主要占用情况

1653132098823

接下来我们看到了这些配置，最关键的缓存区内存如何定位和解决呢？

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

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

以上复制缓冲区和 AOF 缓冲区不会有问题，最关键就是客户端缓冲区的问题

客户端缓冲区：指的就是我们发送命令时，客户端用来缓存命令的一个缓冲区，也就是我们向 redis 输入数据的输入端缓冲区和 redis 向客户端返回数据的响应缓存区，输入缓冲区最大 1G 且不能设置，所以这一块我们根本不用担心，如果超过了这个空间，redis 会直接断开，因为本来此时此刻就代表着 redis 处理不过来了，我们需要担心的就是输出端缓冲区

1653132410073

我们在使用 redis 过程中，处理大量的 big value，那么会导致我们的输出结果过多，如果输出缓存区过大，会导致 redis 直接断开，而默认配置的情况下，其实他是没有大小的，这就比较坑了，内存可能一下子被占满，会直接导致咱们的 redis 断开，所以解决方案有两个

1、设置一个大小

2、增加我们带宽的大小，避免我们出现大量数据从而直接超过了 redis 的承受能力

7. 服务器端集群优化-集群还是主从

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

集群完整性问题
集群带宽问题
数据倾斜问题
客户端性能问题
命令的集群兼容性问题
lua 和事务问题

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

大家可以设想一下，如果有几个 slot 不能使用，那么此时整个集群都不能用了，我们在开发中，其实最重要的是可用性，所以需要把如下配置修改成 no，即有 slot 不能使用时，我们的 redis 集群还是可以对外提供服务

1653132740637

问题 2、集群带宽问题

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

插槽信息
集群状态信息

集群中节点越多，集群状态信息数据量也越大，10 个节点的相关信息可能达到 1kb，此时每次集群互通需要的带宽会非常高，这样会导致集群中大量的带宽都会被 ping 信息所占用，这是一个非常可怕的问题，所以我们需要去解决这样的问题

解决途径：

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

问题 3、命令的集群兼容性问题

有关这个问题咱们已经探讨过了，当我们使用批处理的命令时，redis 要求我们的 key 必须落在相同的 slot 上，然后大量的 key 同时操作时，是无法完成的，所以客户端必须要对这样的数据进行处理，这些方案我们之前已经探讨过了，所以不再这个地方赘述了。

问题 4、lua 和事务的问题

lua 和事务都是要保证原子性问题，如果你的 key 不在一个节点，那么是无法保证 lua 的执行和事务的特性的，所以在集群模式是没有办法执行 lua 和事务的

那我们到底是集群还是主从

单体 Redis（主从 Redis）已经能达到万级别的 QPS，并且也具备很强的高可用特性。如果主从能满足业务需求的情况下，所以如果不是在万不得已的情况下，尽量不搭建 Redis 集群

Redis高级篇及最佳实践

一.分布式缓存

1.Redis 持久化

1.1 RDB 持久化

1.1.1 什么时候执行 RDB

1.1.2 RDB 原理

1.1.3 总结

1.2 AOP 持久化

1.2.1 AOP 原理

1.2.2 AOF 配置

1.2.3.AOF 文件重写

1.3.RDB 与 AOF 对比

2.Redis 主从同步

2.1 搭建主从架构

2.1.1 准备实例和配置

2.1.2 开启主从关系

2.2 数据同步原理

2.2.1 全量同步

2.2.2 增量同步

2.3 主从同步优化

2.4 小结

3.Redis 哨兵模式

3.1 哨兵模式原理

3.1.1 哨兵集群结构及作用

3.1.2 监控原理

3.1.3 故障恢复原理

3.1.4.小结

3.2 搭建哨兵集群

3.2.1 准备实例和配置

3.2.2 测试

3.3 RedisTemplate

3.3.1 配置文件

3.3.2 配置读写分离

4.Redis 分片集群

4.1 为什么要分片集群？

4.2 搭建分片集群

4.2.1 准备实例与配置

4.2.2 创建集群

4.3 散列插槽

4.3.1.插槽原理

4.3.1.小结

4.4 集群伸缩

4.4.1 需求分析

4.4.2 创建新的 redis 实例

4.4.3 添加新节点到 redis

4.4.4 转移插槽

4.5 故障转移

4.5.1 自动故障转移

4.5.2 手动故障转移

4.6 RedisTemplate 访问分片集群

二.多级缓存

1.什么是多级缓存？

1.1 传统缓存及存在的问题

1.2 多级缓存是什么

2.JVM 进程缓存

2.1 初识 Caffeine

2.2 基本 API 及缓存驱逐策略

2.3 使用 Caffeine

3.Lua 语法入门

4.实现多级缓存

4.1 OpenResty 安装及快速入门

4.2 请求参数处理

4.2.1.获取参数的 API

4.3 查询 Tomcat

4.3.1 原理

4.3.2 实现

4.4 Redis 缓存预热

4.4.1 编码实现

4.6.查询 Redis 缓存

4.7.Nginx 本地缓存

5.缓存同步

5.1.数据同步策略

5.2.安装 Canal

5.2.1.认识 Canal

5.2.2.安装 Canal

1.开启 MySQL 主从

1.1.开启 binlog

1.2.设置用户权限

2.安装 Canal

2.1.创建网络