9 RDB

9.1 RDB的介绍

在指定的时间间隔内将内存中的数据集快照写入磁盘， 也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里

Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。 整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能 如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。

Fork的作用是复制一个与当前进程一样的进程。新进程的所有数据（变量、环境变量、程序计数器等） 数值都和原进程一致，但是是一个全新的进程，并作为原进程的子进程

• 在Linux程序中，fork()会产生一个和父进程完全相同的子进程，但子进程在此后多会exec系统调用，出于效率考虑，Linux中引入了“写时复制技术”

• 一般情况父进程和子进程会共用同一段物理内存，只有进程空间的各段的内容要发生变化时，才会将父进程的内容复制一份给子进程。

在配置文件中我们可以看到

# The filename where to dump the DB
dbfilename dump.rdb
​
# Note that you must specify a directory here, not a file name.
dir ./

这两个是配置rdb快照文件保存位置的，默认是当前服务启动目录下的dumo.rdb文件

9.2 触发rdb快照

9.2.1 配置文件中默认的快照配置

# Unless specified otherwise, by default Redis will save the DB:
#   * After 3600 seconds (an hour) if at least 1 key changed
#   * After 300 seconds (5 minutes) if at least 100 keys changed
#   * After 60 seconds if at least 10000 keys changed
#
# You can set these explicitly by uncommenting the three following lines.
#
# save 3600 1
# save 300 100
# save 60 10000

当变化的时间越多，重复的时间越短

具体配置文件可以查看快照配置文件

9.2.2 命令操作

• save save时只管保存，其它不管，全部阻塞。手动保存。不建议

• bgsaveRedis会在后台异步进行快照操作， 快照同时还可以响应客户端请求。

可以通过lastsave 命令获取最后一次成功执行快照的时间

执行flushall命令，也会产生dump.rdb文件，但里面是空的，无意义

9.2.3 停止rdb

动态停止RDB：redis-cli config set save ""save后给空值，表示禁用保存策略

9.3 rdb的备份和回复

9.3.1 rdb的备份

先通过config get dir 查询rdb文件的目录

将*.rdb的文件拷贝到别的地方

9.3.2 rdb的恢复

• 关闭Redis

• 先把备份的文件拷贝到工作目录下 cp dump2.rdb dump.rdb

• 启动Redis, 备份数据会直接加载

9.4 rdb的优势和劣势

9.4.1 优势

• 适合大规模的数据恢复

• 对数据完整性和一致性要求不高更适合使用

• 节省磁盘空间

• 恢复速度快

9.4.2 劣势

• Fork的时候，内存中的数据被克隆了一份，大致2倍的膨胀性需要考虑

• 虽然Redis在fork时使用了写时拷贝技术,但是如果数据庞大时还是比较消耗性能。

• 在备份周期在一定间隔时间做一次备份，所以如果Redis意外down掉的话，就会丢失最后一次快照后的所有修改。

10 AOF

10.1 AOF介绍

10.1.1 简介

AOF（Append Only File）以日志的形式来记录每个写操作（增量保存），将Redis执行过的所有写指令记录下来(读操作不记录)， 只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作

10.1.2 AOF持久化流程

1. 客户端的请求写命令会被append追加到AOF缓冲区内；

2. AOF缓冲区根据AOF持久化策略[always,everysec,no]将操作sync同步到磁盘的AOF文件中；

3. AOF文件大小超过重写策略或手动重写时，会对AOF文件rewrite重写，压缩AOF文件容量；

4. Redis服务重启时，会重新load加载AOF文件中的写操作达到数据恢复的目的

10.2 AOF的使用

AOF默认不开启

# AOF and RDB persistence can be enabled at the same time without problems.
# If the AOF is enabled on startup Redis will load the AOF, that is the file
# with the better durability guarantees.
#
# Please check http://redis.io/topics/persistence for more information.
​
appendonly no
​

可以在redis.conf中配置文件名称，默认为 appendonly.aof

# The name of the append only file (default: "appendonly.aof")
​
appendfilename "appendonly.aof"

若AOF和RDB同时开启，系统默认取AOF的数据

AOF的备份机制和性能虽然和RDB不同, 但是备份和恢复的操作同RDB一样，都是拷贝备份文件，需要恢复时再拷贝到Redis工作目录下，启动系统即加载。

恢复数据：

• 修改默认的appendonly no，改为yes

• 将有数据的aof文件复制一份保存到对应目录(查看目录：config get dir)

• n 恢复：重启redis然后重新加载

如遇到AOF文件损坏，通过进行恢复

redis-check-aof--fix appendonly.aof

AOF同步频率设置:

# 始终同步，每次Redis的写入都会立刻记入日志；性能较差但数据完整性比较好
# appendfsync always
​
# 每秒同步，每秒记入日志一次，如果宕机，本秒的数据可能丢失。默认
appendfsync everysec
​
# redis不主动进行同步，把同步时机交给操作系统。
# appendfsync no

10.3 Rewite压缩

10.3.1 介绍

AOF采用文件追加方式，文件会越来越大为避免出现此种情况，新增了重写机制, 当AOF文件的大小超过所设定的阈值时，Redis就会启动AOF文件的内容压缩， 只保留可以恢复数据的最小指令集.可以使用命令bgrewriteaof

AOF文件持续增长而过大时，会fork出一条新进程来将文件重写(也是先写临时文件最后再rename)，redis4.0版本后的重写，是指上就是把rdb 的快照，以二级制的形式附在新的aof头部，作为已有的历史数据，替换掉原来的流水账操作。

触发机制：Redis会记录上次重写时的AOF大小，默认配置是当AOF文件大小是上次rewrite后大小的一倍且文件大于64M时触发

auto-aof-rewrite-percentage：设置重写的基准值，文件达到100%时开始重写（文件是原来重写后文件的2倍时触发）

auto-aof-rewrite-min-size：设置重写的基准值，最小文件64MB。达到这个值开始重写。

重写流程

1. bgrewriteaof触发重写，判断是否当前有bgsave或bgrewriteaof在运行，如果有，则等待该命令结束后再继续执行。

2. 主进程fork出子进程执行重写操作，保证主进程不会阻塞。

3. 子进程遍历redis内存中数据到临时文件，客户端的写请求同时写入aof_buf缓冲区和aof_rewrite_buf重写缓冲区保证原AOF文件完整以及新AOF文件生成期间的新的数据修改动作不会丢失。

4. 子进程写完新的AOF文件后，向主进程发信号，父进程更新统计信息。主进程把aof_rewrite_buf中的数据写入到新的AOF文件。

5. 使用新的AOF文件覆盖旧的AOF文件，完成AOF重写。

10.4 AOF优势和劣势

10.4.1 AOF的优势

• 备份机制更稳健，丢失数据概率更低。

• 可读的日志文本，通过操作AOF稳健，可以处理误操作

10.4.2 AOF的劣势

• 比起RDB占用更多的磁盘空间。

• 恢复备份速度要慢。

• 每次读写都同步的话，有一定的性能压力。

• 存在个别Bug，造成恢复不能。

10.5 AOF和RDB的对比

• RDB持久化方式能够在指定的时间间隔能对你的数据进行快照存储

• AOF持久化方式记录每次对服务器写的操作,当服务器重启的时候会重新执行这些命令来恢复原始的数据,AOF命令以redis协议追加保存每次写的操作到文件末尾.

• Redis还能对AOF文件进行后台重写,使得AOF文件的体积不至于过大

• 只做缓存：如果你只希望你的数据在服务器运行的时候存在,你也可以不使用任何持久化方式.

• 同时开启两种持久化方式：

  • 在这种情况下,当redis重启的时候会优先载入AOF文件来恢复原始的数据, 因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整.

RDB的数据不实时，同时使用两者时服务器重启也只会找AOF文件。那要不要只使用AOF呢？

建议不要，因为RDB更适合用于备份数据库(AOF在不断变化不好备份)， 快速重启，而且不会有AOF可能潜在的bug，留着作为一个万一的手段。

11 主从复制

11.1 概述

主从复制是指主机数据更新后根据配置和策略， 自动同步到备机的master/slaver机制，Master以写为主，Slave以读为主

好处：

• 读写分离，性能扩展

• 容灾快速恢复

11.2 模拟主从复制环境

[root@VM-4-3-centos myredis]# ll
total 104
-rw-r--r-- 1 root root    94 Oct  8 21:56 redis7379.conf
-rw-r--r-- 1 root root    94 Oct  8 21:54 redis7380.conf
-rw-r--r-- 1 root root    94 Oct  8 21:55 redis7381.conf
-rw-r--r-- 1 root root 92251 Oct  8 21:48 redis.conf

其中redis.conf为原先的配置文件，其余文件形式如下：

include /myredis/redis.conf
pidfile /var/run/redis_7379.pid
port 7379
dbfilename dump7379.rdb

启动三个服务

[root@VM-4-3-centos myredis]# redis-server redis7379.conf
[root@VM-4-3-centos myredis]# redis-server redis7380.conf
[root@VM-4-3-centos myredis]# redis-server redis7381.conf
[root@VM-4-3-centos myredis]# ps -ef | grep redis
root     11882     1  0 22:44 ?        00:00:00 redis-server *:7379
root     11904     1  0 22:44 ?        00:00:00 redis-server *:7380
root     11910     1  0 22:44 ?        00:00:00 redis-server *:7381
root     11938 11666  0 22:44 pts/2    00:00:00 grep --color=auto redis

查看三台主机的运行情况

info replication打印主从复制的相关信息

127.0.0.1:7381> info replication
# Replication
role:master
connected_slaves:0
master_failover_state:no-failover
master_replid:c7bc6e8b3de76c02e840f1e8dbf44a45ad8de31a
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

使用slaveof <ip> <port>命令成为某个实例的从服务器

127.0.0.1:7380> slaveof 127.0.0.1 7379
OK
127.0.0.1:7380> info replication
# Replication
role:slave
master_host:127.0.0.1
master_port:7379
master_link_status:down
master_last_io_seconds_ago:-1
master_sync_in_progress:0
slave_repl_offset:1
master_link_down_since_seconds:1665240856
slave_priority:100
slave_read_only:1
connected_slaves:0
master_failover_state:no-failover
master_replid:fcfd9b42be2f292f9ce5d186eb534626b674e4cb
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

• 在主机上写，在从机上可以读取数据

• 在从机上写数据报错

• 主机挂掉，重启就行，一切如初

• 从机重启需重设：slaveof <ip> <port>

11.3 主从复制的三个使用场景

11.3.1一主二仆

Q:当从机shutdown后再次加入主服务器后，是从头开始复制还是从切入点开始复制？

A:从头开始复制

Q:主机shutdown后情况如何？从机是上位还是原地待命?

A:原地待命

Q:主机又回来了后，主机新增记录，从机还能否顺利复制？

A:可以的

主从复制的原理：

• Slave启动成功连接到master后会发送一个sync命令

• Master接到命令启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令， 在后台进程执行完毕之后，master将传送整个数据文件到slave,以完成一次完全同步

• 全量复制：而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中。

• 增量复制：Master继续将新的所有收集到的修改命令依次传给slave,完成同步（每次主服务器进行写操作后，会和从服务器进行数据同步）

• 但是只要是重新连接master,一次完全同步（全量复制)将被自动执行

11.3.2 薪火相传

上一个Slave可以是下一个slave的Master，Slave同样可以接收其他 slaves的连接和同步请求，那么该slave作为了链条中下一个的master, 可以有效减轻master的写压力,去中心化降低风险。

风险是一旦某个slave宕机，后面的slave都没法备份

11.3.3 反客为主

当一个master宕机后，后面的slave可以立刻升为master，其后面的slave不用做任何修改。

用slaveof no one 将从机变为主机。

11.4 哨兵模式

11.4.1 概念

反客为主的自动版，能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库

11.4.2 使用方法

目录下新建sentinel.conf文件

sentinel monitor mymaster 127.0.0.1 7379 1

其中mymaster为监控对象起的服务器名称， 1 为至少有多少个哨兵同意迁移的数量。

注意sentinel.conf为固定写法，不能随意更改

启动哨兵：

redis-sentinel  /myredis/sentinel.conf 

当主机挂掉，从机选举中产生新的主机

哪个从机会被选举为主机呢？根据优先级别：replica-priority

优先级在redis.conf中默认：replica-priority 100，值越小优先级越高

• 优先级在redis.conf中默认：slave-priority 100，值越小优先级越高

• 偏移量是指获得原主机数据最全的

• 每个redis实例启动后都会随机生成一个40位的runid

# However a special priority of 0 marks the replica as not able to perform the
# role of master, so a replica with priority of 0 will never be selected by
# Redis Sentinel for promotion.
#
# By default the priority is 100.
replica-priority 100

11.4.3 复制延迟

由于所有的写操作都是先在Master上操作，然后同步更新到Slave上，所以从Master同步到Slave机器有一定的延迟，当系统很繁忙的时候，延迟问题会更加严重，Slave机器数量的增加也会使这个问题更加严重。

12 Redis集群

12.1 概念

12.1.1 引入

容量不够，redis如何进行扩容？

并发写操作， redis如何分摊？

另外，主从模式，薪火相传模式，主机宕机，导致ip地址发生变化，应用程序中配置需要修改对应的主机地址、端口等信息。

之前通过代理主机来解决，但是redis3.0中提供了解决方案。就是无中心化集群配置。

代理主机：客户端请求代理服务器，代理服务器分发请求

无中心化集群：任何一台服务器都可以作为集群的入口

12.1.2 Redis集群的概念

Redis 集群实现了对Redis的水平扩容，即启动N个redis节点，将整个数据库分布存储在这N个节点中，每个节点存储总数据的1/N。

Redis 集群通过分区（partition）来提供一定程度的可用性（availability）： 即使集群中有一部分节点失效或者无法进行通讯， 集群也可以继续处理命令请求。

12.2 模拟集群环境

与11.2类似

redie.conf增加

include /myredis/redis.conf
pidfile /var/run/redis_7379.pid
port 7379
dbfilename dump7379.rdb
​
# 打开集群模式
cluster-enabled yes
# 设定节点配置文件名
cluster-config-file nodes-7379.conf
# 设定节点失联时间，超过该时间（毫秒），集群自动进行主从切换。
cluster-node-timeout 15000 

[root@VM-4-3-centos myredis]# ll
total 104
-rw-r--r-- 1 root root    94 Oct  8 21:56 redis7379.conf
-rw-r--r-- 1 root root    94 Oct  8 21:54 redis7380.conf
-rw-r--r-- 1 root root    94 Oct  8 21:55 redis7381.conf
-rw-r--r-- 1 root root    94 Oct  8 23:48 redis7389.conf
-rw-r--r-- 1 root root    94 Oct  8 23:49 redis7390.conf
-rw-r--r-- 1 root root    94 Oct  8 23:50 redis7391.conf
-rw-r--r-- 1 root root 92251 Oct  8 21:48 redis.conf

将6台服务器启动，当实例启动后，会生成各自的节点配置文件nodes-<port>.conf

将六个节点合成一个集群

进入redis安装地址注意是安装环境不是客户端所在地址

redis-cli --cluster create --cluster-replicas 1 ip地址:7379 ip地址:7380 ip地址:7381 ip地址:7389 ip地址:7390 ip地址:7391

• 此处不要用127.0.0.1， 请用真实IP地址

• 由于redis6.2.1及以上自带rb环境redis-trib.rb，以下的需要自行安装

--replicas 1 采用最简单的方式配置集群，一台主机，一台从机，正好三组。

他会直接帮我们分配集群，输入yes可以接受分配

一个集群至少要有三个主节点。分配原则尽量保证每个主数据库运行在不同的IP地址，每个从库和主库不在一个IP地址上

看到以下提示消息即完成操作

[OK] All nodes agree about slots configuration 
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered

使用-c 采用集群策略连接，设置数据会自动切换到相应的写主机

redis-cli -c -p 7379

通过cluster nodes 命令查看集群信息

12.3 slots

记得我们启动集群的时候吗

[OK] All nodes agree about slots configuration 
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered

这个slots是插槽

一个 Redis 集群包含 16384 个插槽（hash slot）， 数据库中的每个键都属于这 16384 个插槽的其中一个，

集群使用公式 CRC16(key) % 16384 来计算键 key 属于哪个槽， 其中 CRC16(key) 语句用于计算键 key 的 CRC16 校验和 。

集群中的每个节点负责处理一部分插槽。 举个例子， 如果一个集群可以有主节点， 其中：

• 节点 A 负责处理 0 号至 5460 号插槽。

• 节点 B 负责处理 5461 号至 10922 号插槽。

• 节点 C 负责处理 10923 号至 16383 号插槽。

ip: 7379 > set k1 v1
-> Redirected to slot [12706] at ip:7381

注意，在集群中mset有不同的写法，需要定义组的概念，从而使key中{}内相同内容的键值对放到一个slot中去。

ip:7379 > mset k1 v1 k2 v2 k3 v3
(error) CROSSLOT Keys in request don't hash to the same slot
ip:7379 > mset k1{cust} v1 k2{cust} v2 k3{cust} v3
-> Redirected to slot [4847] at ip:7379

查询集群中的值

CLUSTER GETKEYSINSLOT <slot> <count> 返回 count 个 slot 槽中的键

注意，只能看到自己机器负责的插槽的值

12.4 故障恢复

Q:如果主节点下线？从节点能否自动升为主节点？

A:可以

Q:主节点恢复后，主从关系会如何？

A:主节点回来变成从机

Q:如果所有某一段插槽的主从节点都宕掉，redis服务是否还能继续?

A:

• 如果某一段插槽的主从都挂掉，而cluster-require-full-coverage 为yes ，那么 ，整个集群都挂掉

• 如果某一段插槽的主从都挂掉，而cluster-require-full-coverage 为no ，那么，该插槽数据全都不能使用，也无法存储。

cluster-require-full-coverage为redis.conf中的参数

12.5 集群的Jedis开发

即使连接的不是主机，集群会自动切换主机存储。主机写，从机读。

无中心化主从集群。无论从哪台主机写的数据，其他主机上都能读到数据。

public class JedisClusterTest {
    public static void main(String[] args) { 
        hostAndPort = new HostAndPort("ip",port);
        JedisCluster jedisCluster=new JedisCluster(hostAndPort);
        jedisCluster.set("k1", "v1");
        System.out.println(jedisCluster.get("k1"));
    }
}

12.6 Redis的好处和坏处

12.6.1 好处

• 实现扩容

• 分摊压力

• 无中心配置相对简单

12.6.2 坏处

• 多键操作是不被支持的

• 多键的Redis事务是不被支持的。lua脚本不被支持

• 由于集群方案出现较晚，很多公司已经采用了其他的集群方案，而代理或者客户端分片的方案想要迁移至redis cluster，需要整体迁移而不是逐步过渡，复杂度较大。

13 Redis常见的问题

13.1 缓存穿透

13.1.1 问题描述

key对应的数据在数据源并不存在，每次针对此key的请求从缓存获取不到，请求都会压到数据源，从而可能压垮数据源。比如用一个不存在的用户id获取用户信息，不论缓存还是数据库都没有，若黑客利用此漏洞进行攻击可能压垮数据库。

现象：

• 应用服务器压力变大

• redis命中率降低，导致一直查询数据库

原因：

• redis查不到数据库

• 出现很多非正常url访问

13.1.2 解决方案

一个一定不存在缓存及查询不到的数据，由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。

• 对空值缓存：如果一个查询返回的数据为空（不管是数据是否不存在），我们仍然把这个空结果（null）进行缓存，设置空结果的过期时间会很短，最长不超过五分钟

• 设置可访问的名单（白名单）：使用bitmaps类型定义一个可以访问的名单，名单id作为bitmaps的偏移量，每次访问和bitmap里面的id进行比较，如果访问id不在bitmaps里面，进行拦截，不允许访问。

• 采用布隆过滤器：(布隆过滤器（Bloom Filter）是1970年由布隆提出的。它实际上是一个很长的二进制向量(位图)和一系列随机映射函数（哈希函数）。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误识别率和删除困难。将所有可能存在的数据哈希到一个足够大的bitmaps中，一个一定不存在的数据会被 这个bitmaps拦截掉，从而避免了对底层存储系统的查询压力。

• 进行实时监控：当发现Redis的命中率开始急速降低，需要排查访问对象和访问的数据，和运维人员配合，可以设置黑名单限制服务

13.2 缓存击穿

13.2.1 问题描述

key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。

现象：

• 数据库访问压力瞬时增大

• redis里面没有出现大量的key过期

• redis正常运行

原因：

• redis某个key过期了

• 大量访问使用这个key

13.2.2 解决方案

key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：缓存被“击穿”的问题。

• 预先设置热门数据：在redis高峰访问之前，把一些热门数据提前存入到redis里面，加大这些热门数据key的时长

• 实时调整：现场监控哪些数据热门，实时调整key的过期时长

• 使用锁：

  • 就是在缓存失效的时候（判断拿出来的值为空），不是立即去load db。

  • 先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX）去set一个mutex key

  • 当操作返回成功时，再进行load db的操作，并回设缓存,最后删除mutex key；

  • 当操作返回失败，证明有线程在load db，当前线程睡眠一段时间再重试整个get缓存的方法。

13.3 缓存雪崩

13.3.1 问题描述

key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。

缓存雪崩与缓存击穿的区别在于这里针对很多key缓存，前者则是某一个key

现象：

• 数据库压力变大造成服务器崩溃

原因：

• 在极少的时间段内，出现大量的key集中过期的情况

13.3.2 解决方案

• 构建多级缓存架构：nginx缓存 + redis缓存 +其他缓存（ehcache等）

• 使用锁或队列：用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。不适用高并发情况

• 设置过期标志更新缓存：记录缓存数据是否过期（设置提前量），如果过期会触发通知另外的线程在后台去更新实际key的缓存。

• 将缓存失效时间分散开：比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。

13.4 分布式锁

13.4.1 问题描述

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题

分布式锁主流的实现方案：

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

• 基于缓存（Redis等）

• 基于Zookeeper

每一种分布式锁解决方案都有各自的优缺点：

性能：redis最高

可靠性：zookeeper最高

13.4.2 使用SETNX

现在大部分推荐使用SETNX + EXPIRE来实现分布式锁，可以见基本操作

核心代码如下

public void testLockLua() {
    String uuid = UUID.randomUUID().toString();
    String skuId = "25"; // 25号商品的数据
    String locKey = "lock:" + skuId;
​
    Boolean lock = redisTemplate.opsForValue().setIfAbsent(locKey, uuid, 3, TimeUnit.SECONDS);
​
    if (lock) {
        Object value = redisTemplate.opsForValue().get("num");
        if (StringUtils.isEmpty(value)) {
            return;
        }
        int num = Integer.parseInt(value + "");
        redisTemplate.opsForValue().set("num", String.valueOf(++num));
        
        // 定义lua 脚本
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        // 使用redis执行lua执行
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(script);
        redisScript.setResultType(Long.class);
        // 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。
        redisTemplate.execute(redisScript, Arrays.asList(locKey), uuid);
    } else {
        try {
            Thread.sleep(1000);
            testLockLua();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

13.4.3 使用HSET

上面的代码虽然做到了设置过期时间，自动释放锁，防误删，原子化等，但它有一个致命问题，就是不满足锁的一个可重入性。

所谓可重入性，通俗来说，就是你在一个方法中加了锁之后，若它调了另外一个方法，这个方法也需要获取同一把锁的时候，它无需再进行等待操作。

比如你用钥匙进了你家门，就不用再用钥匙进你家厨房了

解决锁的可重入性，可以参考ReentrantLock，ReentrantLock的底层通过一个计数器c，当c=0时，说明该锁没有被任意一个线程持有，而c!=0时就说明该锁被某个线程持有，而且，你会发现，你在线程中锁了多少次，这个计数器就是多少。(这也是为啥你lock了几次，就要调用unlock几次的原因)。

因此，这个时候我们就想要一种数据结构，类似于Map<String, Map<String, Integer>>的形式，也就是Redis中的Hash结构。

这里我们直接上源码，这里实现了Lock接口，满足了锁的一个规范。

@Slf4j
@SuppressWarnings({"all"})
public class RedisDistributedLock implements Lock {
​
    private final StringRedisTemplate redisTemplate;
​
    /**
     * 锁的名称，Map<String, <Map<String, Integer>> 中的第一个String
     */
    private final String lockName;
​
    /**
     * Map<String, <Map<String, Integer>> 中的第二个String
     */
    private final String value;
​
    private static final long EXPIRE_TIME = 5L;
​
    public RedisDistributedLock(StringRedisTemplate redisTemplate, String lockName, String uuid) {
        this.redisTemplate = redisTemplate;
        this.lockName = lockName;
        // uuid从工厂传进来, 保证唯一性
        this.value = uuid + "-" + Thread.currentThread().getId();
    }
​
    @Override
    public void lock() {
        tryLock();
    }
​
    @Override
    public void unlock() {
        // 解锁的Lua脚本
        String script = LuaScript.LUA_UNLOCK_SCRIPT;
        Long result = redisTemplate.execute(new DefaultRedisScript<>(script, Long.class), List.of(lockName), value);
        if (null == result) {
            throw new RuntimeException("The Lock doesn't exists.");
        }
        log.info("unlock {} success. ", lockName);
    }
​
    @Override
    public boolean tryLock() {
        return tryLock(-1L, TimeUnit.SECONDS);
    }
​
    @Override
    public boolean tryLock(long time, TimeUnit unit) {
        if (-1L == time) {
            // 加锁的Lua脚本
            String script = LuaScript.LUA_LOCK_SCRIPT;
            while(!redisTemplate.execute(new DefaultRedisScript<>(script,Boolean.class), Arrays.asList(lockName), value,String.valueOf(EXPIRE_TIME))) {
                ThreadUtils.sleep(EXPIRE_TIME);
            }
            log.info("lock {} success. ", lockName);
            return true;
        }
        return false;
    }
​
    @Override
    public void lockInterruptibly() throws InterruptedException {
        // unused
    }
​
    @Override
    public Condition newCondition() {
        // unused
        return null;
    }
}

LUA脚本如下：

public class LuaScript {
​
    private LuaScript() {}
​
    public static final String LUA_LOCK_SCRIPT =
            "if redis.call('exists',KEYS[1]) == 0 or redis.call('hexists',KEYS[1],ARGV[1]) == 1 then    " +
                "redis.call('hincrby',KEYS[1],ARGV[1],1)    " +
                "redis.call('expire',KEYS[1],ARGV[2])    " +
                "return 1  " +
            "else   " +
                "return 0 " +
            "end";
​
    public static final String LUA_UNLOCK_SCRIPT =
            "if redis.call('HEXISTS',KEYS[1],ARGV[1]) == 0 then    " +
                "return nil  " +
            "elseif redis.call('HINCRBY',KEYS[1],ARGV[1],-1) == 0 then    " +
                "return redis.call('del',KEYS[1])  " +
            "else    " +
                "return 0 " +
            "end";
}

这里我们使用工厂模式来获得Redis锁

@Component
@SuppressWarnings("unused")
public class DistributedLockFactory {
​
    private final String uuid;
​
    private final StringRedisTemplate redisTemplate;
​
    public DistributedLockFactory(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.uuid = IdUtil.simpleUUID();
    }
​
    public Lock getDistributedLock(LockEnum type) {
        return switch (type) {
            // Java17新特性
            case REDIS -> new RedisDistributedLock(redisTemplate, LockNameConstants.REDIS_LOCK_NAME, uuid);
            // TODO 补充
            case ZOOKEEPER, MYSQL -> null;
        };
    }
​
}

测试：

public String sale() {
    String retMessage;
​
    Lock redisLock = distributedLockFactory.getDistributedLock(LockEnum.REDIS);
    redisLock.lock();
    try {
        //1 查询库存信息
        String result = redisTemplate.opsForValue().get("inventory001");
        //2 判断库存是否足够
        int inventoryNumber = result == null ? 0 : Integer.parseInt(result);
        //3 扣减库存，每次减少一个
        if (inventoryNumber > 0) {
            redisTemplate.opsForValue().set("inventory001", String.valueOf(--inventoryNumber));
            retMessage = "成功卖出一个商品,库存剩余: " + inventoryNumber;
            ThreadUtils.sleep(20);
            testReEntry();
        } else {
            retMessage = "商品卖完了";
        }
        log.info("{} \t 服务端口号: {}", retMessage, port);
    } finally {
        redisLock.unlock();
    }
    return retMessage + "\t" + "服务端口号" + port;
}
​
private void testReEntry() {
    Lock redisLock = distributedLockFactory.getDistributedLock(LockEnum.REDIS);
    redisLock.lock();
    try
    {
        log.info("测试可重入锁");
    }finally {
        redisLock.unlock();
    }
}

13.4.5 总结

为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

• 互斥性。在任意时刻，只有一个客户端能持有锁。

• 不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。

• 加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

• 加锁和解锁必须具有原子性。

13.5 Redis大Key的删除

13.5.1 定义

Redis大Key其实不是指的是Key，而是指Key对应的Value。

Redis的String类型，最大可以存512MB的值，List、Hash、Set、ZSet等，可以存储2³²个元素（大约42亿多点），那么多大的Key算大Key呢？

一般来说，我们String需要控制在10KB以下，而集合类型的控制元素个数在5000个以下。

13.5.2 如何导致的大Key

一种，比如社交类的场景，如粉丝数增多，另外一种，可能由于报表等常年累月的积累，有可能导致大Key。

我们知道，由于Redis工作线程是单线程的，大Key删除会导致网络浏览阻塞、超时等一系列问题，那么我们就应该去防止大KEY的产生

13.5.3 如何检测大Key

一般来说，有两种方法：

1. 使用--bigkeys：扫描所有的键，给出每种数据结构Top 1 bigkey

[root@localhost ~]# redis-cli -a Redis密码 --bigkeys
Warning: Using a password with '-a' or '-u' option on the command line interface may not be safe.
​
# Scanning the entire keyspace to find biggest keys as well as
# average sizes per key type.  You can use -i 0.1 to sleep 0.1 sec
# per 100 SCAN commands (not usually needed).
​
[00.00%] Biggest string found so far '"k5889"' with 5 bytes
[00.00%] Biggest string found so far '"k607267"' with 7 bytes
[00.00%] Biggest string found so far '"k1802564"' with 8 bytes
[50.00%] Sampled 1000000 keys so far
[100.00%] Sampled 2000000 keys so far
​
-------- summary -------
​
Sampled 2000000 keys in the keyspace!
Total key length in bytes is 14888896 (avg len 7.44)
​
Biggest string found '"k1802564"' has 8 bytes
​
0 lists with 0 items (00.00% of keys, avg size 0.00)
0 hashs with 0 fields (00.00% of keys, avg size 0.00)
2000000 strings with 14888896 bytes (100.00% of keys, avg size 7.44)
0 streams with 0 entries (00.00% of keys, avg size 0.00)
0 sets with 0 members (00.00% of keys, avg size 0.00)
0 zsets with 0 members (00.00% of keys, avg size 0.00)

2. 使用Memery Usage：获取一个键的所占内存

127.0.0.1:6379> MEMORY USAGE k10000
(integer) 64

13.5.4 大Key的删除

13.5.4.1 String类型

String类型一般用Del即可，如果过于庞大可以使用Unlink

13.5.4.2 Hash类型

使用hscan每次获取少量field-value，再使用hdel删除每个field

public void delBigHash(String bigHashKey) {
    ScanParams scanParams = new ScanParams().count(100);
    String cursor = "0";
    do {
        ScanResult<Entry<String, String>> scanResult = jedis.hscan(bigHashKey, cursor, scanParams);
        List<Entry<String, String>> entryList = scanResult.getResult();
        if (CollectionUtils.isNotEmpty(entryList)) {
            for (Entry<String, String> entry : entryList) {
                jedis.hdel(bigHashKey, entry.getKey());
            }
        }
        cursor = scanResult.getCursor();
    } while (!"0".equals(cursor));
    jedis.del(bigHashKey);
}

13.5.4.3 List类型

使用ltrim渐进式逐步删除，直到全部删除完成

public void delBigList(String bigListKey){
    long llen = jedis.llen(bigListKey);
    int counter = 0;
    int left = 100;
    while (counter < llen) {
        jedis.ltrim(bigListKey, left, llen);
        counter += left;
    }
    jedis.del(bigListKey);
}

13.5.4.4 Set类型

使用sscan每次获取部分元素，再使用srem命令删除每个元素

public void delBigSet(String bigSetKey){
    ScanParams scanParams = new ScanParams().count(100);
    String cursor = "0";
    do {
        ScanResult<String> scanResult = jedis.sscan(bigSetKey, cursor, scanParams);
        List<String> memberList = scanResult.getResult();
        if (CollectionUtils.isNotEmpty(memberList)) {
            for (String member : memberList) {
                jedis.srem(bigSetKey, member);
            }
        }
        cursor = scanResult.getCursor();
    } while (!"0".equals(cursor));
    jedis.del(bigSetKey);
}

13.5.4.5 ZSet类型

使用zscan每次获取部分元素，再使用ZREMRANGEBYRANK命令删除每个元素

public void delBigZSet(String bigZSetKey){
    ScanParams scanParams = new ScanParams().count(100);
    String cursor = "0";
    do {
        ScanResult<Tuple> scanResult = jedis.zscan(bigZSetKey, cursor, scanParams);
        List<Tuple> tupleList = scanResult.getResult();
        if (CollectionUtils.isNotEmpty(tupleList)) {
            for (Tuple tuple : tupleList) {
                jedis.zrem(bigZSetKey, tuple.getElement());
            }
        }
        cursor = scanResult.getCursor();
    } while (!"0".equals(cursor));
    jedis.del(bigZSetKey);
}

14 ACL

14.1 简介

Redis ACL是Access Control List（访问控制列表）的缩写，该功能允许根据可以执行的命令和可以访问的键来限制某些连接。

在Redis 5版本之前，Redis 安全规则只有密码控制 还有通过rename 来调整高危命令比如 flushdb ， KEYS* ， shutdown 等。Redis 6 则提供ACL的功能对用户进行更细粒度的权限控制 ：

• 接入权限:用户名和密码

• 可以执行的命令

• 可以操作的 KEY

14.2 命令

使用acl list命令展现用户权限列表

使用acl cat命令查看添加权限指令类别,加参数类型名可以查看类型下具体命令

使用acl set user命令创建和编辑用户ACL

14.2 IO多线程

IO多线程其实指客户端交互部分的网络IO交互处理模块多线程，而非执行命令多线程。Redis6执行命令依然是单线程。

另外，多线程IO默认也是不开启的，需要再配置文件中配置

io-threads-do-reads yes

io-threads 4

14.3 工具支持 Cluster

之前老版Redis想要搭集群需要单独安装ruby环境，Redis 5 将 redis-trib.rb 的功能集成到 redis-cli 。另外官方 redis-benchmark 工具开始支持 cluster 模式了，通过多线程的方式对多个分片进行压测。

14.4 其他新功能

Redis6新功能还有：

1、RESP3新的 Redis 通信协议：优化服务端与客户端之间通信

2、Client side caching客户端缓存：基于 RESP3 协议实现的客户端缓存功能。为了进一步提升缓存的性能，将客户端经常访问的数据cache到客户端。减少TCP网络交互。

3、Proxy集群代理模式：Proxy 功能，让 Cluster 拥有像单实例一样的接入方式，降低大家使用cluster的门槛。不过需要注意的是代理不改变 Cluster 的功能限制，不支持的命令还是不会支持，比如跨 slot 的多Key操作。

4、Modules API

Redis 6中模块API开发进展非常大，因为Redis Labs为了开发复杂的功能，从一开始就用上Redis模块。Redis可以变成一个框架，利用Modules来构建不同系统，而不需要从头开始写然后还要BSD许可。Redis一开始就是一个向编写各种系统开放的平台。