目录

四、Redis04

4.1 Redis集群应用场景

4.2 集群

4.2.1 基本原理

4.2.2 主从复制的作用

4.3 配置集群（一台虚拟机）

4.3.1 规划网络

4.3.2 创建节点

4.3.3 创建目录

4.3.4 配置redis7001.conf

4.3.5 配置其余文件

4.3.6 后台启动redis

4.3.7 创建Redis的集群

4.3.8 使用cli连接redis集群

4.3.9 检查集群的状态

4.3.10 添加空白主节点

4.3.11 为空白主节点配置从节点

4.3.12 为空白主节点分配插槽

4.3.13 删除节点

4.3.14 集群自动组网

4.3.15 故障检测

4.4 slots介绍

4.4.1 在集群中录入值

4.4.2 集群的优点和不足

4.5 Redis的应用问题

4.5.1 缓存穿透

4.5.2 缓存击穿

4.5.3 缓存雪崩

4.6 分布式锁

4.6.1 问题描述

4.6.2 使用redis实现分布式锁

4.6.3 redis实现分布式锁步骤

4.6.4 分布式锁问题及优化

4.6.5 lua脚本

---

四、Redis04

4.1 Redis集群应用场景

为什么需要redis集群？

当主备复制场景，无法满足主机的单点故障时，需要引入集群配置。

一般数据库要处理的读请求远大于写请求 ，针对这种情况，我们优化数据库可以采用读写分离的策略。我们可以部 署一台主服务器主要用来处理写请求，部署多台从服务器 ，处理读请求。

 

4.2 集群

4.2.1 基本原理

哨兵选举机制，如果有半数节点发现某个异常节点，共同决定改异常节点的状态，如果该节点是主节点，对应的备节点自动顶替为主节点。Sentinel（哨兵）是Redis 的高可用性解决方案：由一个或多个Sentinel 实例 组成的Sentinel 系统可以监视任意多个主服务器，以及这些主服务器属下的所有从服务器，并在被监视的主服务器进入下线状态时，自动将下线主服务器属下的某个从服务器升级为新的主服务器。

4.2.2 主从复制的作用

1、数据冗余：主从复制实现了数据的热备份，是持久化之外的一种数据冗余方式。
2、故障恢复：当主节点出现问题时，可以由从节点提供服务，实现快速的故障恢复；实际上是一种服务的冗余。
3、负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，由从节点提供读服务（即写Redis数据时应用连接主节点，读Redis数据时应用连接从节点），分担服务器负载；尤其是在写少读多的场景下，通过多个从节点分担读负载，可以大大提高Redis服务器的并发量。
4、读写分离：可以用于实现读写分离，主库写、从库读，读写分离不仅可以提高服务器的负载能力，同时可根据需求的变化，改变从库的数量。
5、高可用基石：除了上述作用以外，主从复制还是哨兵和集群能够实施的基础，因此说主从复制是Redis高可用的基础

4.3 配置集群（一台虚拟机）

配置集群所需要的环境

Redis集群至少需要3个节点，因为投票容错机制要求超过半数节点认为某个节点挂了该节点才是挂了，所以2个节点无法构成集群。
​
要保证集群的高可用，需要每个节点都有从节点，也就是备份节点，所以Redis集群至少需要6台服务器。因为我没有那么多服务器，也启动不了那么多虚拟机，所在这里搭建的是伪分布式集群，即一台服务器虚拟运行6个redis实例，修改端口号为（7001-7006），当然实际生产环境的Redis集群搭建和这里是一样的。

4.3.1 规划网络

用一台虚拟机模拟6个节点，一台机器6个节点，创建出3 master、3 salve 环境。虚拟机是 CentOS7 ，ip地址192.168.111.127

4.3.2 创建节点

首先在 192.168.111.127 机器上 /usr/lwl/soft/redis/目录下创建 redis_cluster 目录；
[root@localhost redis]# mkdir redis_cluster
[root@localhost redis]# ls
bin  redis_cluster  

4.3.3 创建目录

在 redis_cluster 目录下，创建名为7001、7002，7003、7004、7005，7006的目录
[root@localhost redis_cluster]# mkdir 7001 7002 7003 7004 7005 7006
[root@localhost redis_cluster]# ls
7001  7002  7003  7004  7005  7006

4.3.4 配置redis7001.conf

配置集群时，建议不要配置密码

复制配置文件到redis_cluster目录下:一定要关闭密码
[root@localhost redis_cluster]# cp /usr/lwl/soft/redis/bin/redis6379.conf redis.conf
[root@localhost redis_cluster]# ls
7001  7002  7003  7004  7005  7006  redis.conf
​
这里复制的文件是在安装目录下拷贝过来的，如果导致开启失败，可以复制解压目录下的原始文件重新进行配置
​
7001目录内新建一个文件redis.conf，内容如下
​
include /usr/lwl/soft/redis/redis_cluster/redis.conf    #包含上一层目录的那个配置文件中的内容
port 7001                           #端口号
pidfile "/var/run/redis_7001.pid"
dbfilename "dump_7001.rdb"
dir "/usr/lwl/soft/redis/redis_cluster/7001"
logfile "/usr/lwl/soft/redis/redis_cluster/7001/redis_err_7001.log"
cluster-enabled yes                 #开启集群
cluster-config-file nodes-7001.conf  #集群节点文件
cluster-node-timeout 15000          #集群超时时间

4.3.5 配置其余文件

快捷命令：将7001中的redis.conf复制到7002 7003 7004 7005 7006中

echo ./7002 ./7003 ./7004 ./7005 ./7006 | xargs -n 1 cp -v /usr/lwl/soft/redis/redis_cluster/7001/redis.conf
​
复制过程:
[root@localhost redis_cluster]# echo ./7002 ./7003 ./7004 ./7005 ./7006 | xargs -n 1 cp -v /usr/lwl/soft/redis/redis_cluster/7001/redis.conf
‘/usr/lwl/soft/redis/redis_cluster/7001/redis.conf’ -> ‘./7002/redis.conf’
‘/usr/lwl/soft/redis/redis_cluster/7001/redis.conf’ -> ‘./7003/redis.conf’
‘/usr/lwl/soft/redis/redis_cluster/7001/redis.conf’ -> ‘./7004/redis.conf’
‘/usr/lwl/soft/redis/redis_cluster/7001/redis.conf’ -> ‘./7005/redis.conf’
‘/usr/lwl/soft/redis/redis_cluster/7001/redis.conf’ -> ‘./7006/redis.conf’
​
然后进入每个目录下，去修改其中的值
例如：要修改7002目录下的redis.conf文件，可以使用命令
    :%s/7001/7002/g
    一次性将全部的7001替换成7002，其他目录的配置也是相同

4.3.6 后台启动redis

要一次性后台启动六台redis，可以编写一个脚本

脚本功能，如果查询出来的进程数为0，则启动这几个redis服务，如果不为0，就关闭redis服务

[root@localhost redis_cluster]# ps -ef|grep -w redis|grep -v grep|wc -l
6
该指令功能：查询除了当前指令产生的redis进程之外的进程数量
grep -w redis：完全匹配redis的进程
grep -v grep：忽略包含grep的进程
wc -l：查询数量
​
脚本功能，如果查询出来的进程数为0，则启动这几个redis服务，如果不为0，就关闭redis服务
​
#!/bin/bash
l=`ps -ef|grep -w redis|grep -v grep|wc -l`
if [ $l -eq 0 ]
        then
/usr/lwl/soft/redis/bin/redis-server /usr/lwl/soft/redis/redis_cluster/7001/redis.conf
/usr/lwl/soft/redis/bin/redis-server /usr/lwl/soft/redis/redis_cluster/7002/redis.conf
/usr/lwl/soft/redis/bin/redis-server /usr/lwl/soft/redis/redis_cluster/7003/redis.conf
/usr/lwl/soft/redis/bin/redis-server /usr/lwl/soft/redis/redis_cluster/7004/redis.conf
/usr/lwl/soft/redis/bin/redis-server /usr/lwl/soft/redis/redis_cluster/7005/redis.conf
/usr/lwl/soft/redis/bin/redis-server /usr/lwl/soft/redis/redis_cluster/7006/redis.conf
echo "已开启redis服务"
        else
/usr/lwl/soft/redis/bin/redis-cli -p 7001 shutdown
/usr/lwl/soft/redis/bin/redis-cli -p 7002 shutdown
/usr/lwl/soft/redis/bin/redis-cli -p 7003 shutdown
/usr/lwl/soft/redis/bin/redis-cli -p 7004 shutdown
/usr/lwl/soft/redis/bin/redis-cli -p 7005 shutdown
/usr/lwl/soft/redis/bin/redis-cli -p 7006 shutdown
echo "已关闭redis服务"
fi
​
如果这里启动失败，可以回到4.4.4中重新复制一份文件进行配置

执行脚本文件：

[root@localhost redis_cluster]# sh redisstart.sh 
[root@localhost redis_cluster]# ps -ef|grep redis
root       7186      1  0 16:59 ?        00:00:02 ./redis-server 0.0.0.0:6379
root       7240      1  0 17:33 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7001 [cluster]
root       7245      1  0 17:33 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7002 [cluster]
root       7247      1  0 17:33 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7003 [cluster]
root       7252      1  0 17:33 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7004 [cluster]
root       7257      1  0 17:33 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7005 [cluster]
root       7265      1  0 17:33 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7006 [cluster]
root       7270   7045  0 17:34 pts/0    00:00:00 grep --color=auto redis

在创建集群时，可能会出现错误

Increased of open files to 10032 (it was originally set to 1024)

 

针对如上错误，作如下处理：

1、查看打开文件的上限和redis服务进程，修改上限：
    输入如下命令，查看其上限：
    ulimit -a

 

2、设置上限
ulimit -n 10032

 

然后重启redis即可

4.3.7 创建Redis的集群

创建集群的命令：安装目录下的redis-cli --cluster create 各个节点及端口  --cluster-replicas 比例 
/usr/lwl/soft/redis/bin/redis-cli --cluster create 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 --cluster-replicas 1
​
最后的这个 -replicas 1 代表着 主节点数/从节点数的比例
如果六个节点且分配比例为1，那么一般前三个是主节点，后三个是从节点

​
​

[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli --cluster create 127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 --cluster-replicas 1
>>> Performing hash slots allocation on 6 nodes...
Master[0] -> Slots 0 - 5460
Master[1] -> Slots 5461 - 10922
Master[2] -> Slots 10923 - 16383
Adding replica 127.0.0.1:7005 to 127.0.0.1:7001
Adding replica 127.0.0.1:7006 to 127.0.0.1:7002
Adding replica 127.0.0.1:7004 to 127.0.0.1:7003
>>> Trying to optimize slaves allocation for anti-affinity
[WARNING] Some slaves are in the same host as their master
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   replicates 16577299b09086cf992b25ea047a53449064a62a
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
Can I set the above configuration? (type 'yes' to accept): #这里输入yes代表同意分配
确认yes之后，会进行分配，分配结果如下：
分配原则尽量保证每个主数据库运行在不同的IP地址，每个从库和主库不在一个IP地址上。

 

4.3.8 使用cli连接redis集群

首先，使用cli连接集群必须要加 -c

连接集群语法：安装目录下/redis-cli  -c -h ip地址 -p 端口号
/usr/lwl/soft/redis/bin/redis-cli -c -h 127.0.0.1 -p 7006
​
查看集群节点的信息： cluster nodes
​
​
#这里连接集群时，使用端口号 7001~7006 都可以
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli -c -h 127.0.0.1 -p 7006
127.0.0.1:7006> cluster nodes      # 查看集群节点的信息
9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006@17006 myself,slave e97325b5f5d020cfd4940421cdece4c3dc7e53d1 0 1676631290000 6 connected
a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004@17004 slave 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 0 1676631290187 4 connected
68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005@17005 slave 16577299b09086cf992b25ea047a53449064a62a 0 1676631291000 5 connected
16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002@17002 master - 0 1676631291199 2 connected 5461-10922
5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001@17001 master - 0 1676631288161 1 connected 0-5460
e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003@17003 master - 0 1676631289177 3 connected 10923-16383

4.3.9 检查集群的状态

执行检查集群状态的命令：出现的界面和配置集群出现的状态一样
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli --cluster check 127.0.0.1:7002
127.0.0.1:7002 (16577299...) -> 0 keys | 5462 slots | 1 slaves.
127.0.0.1:7003 (e97325b5...) -> 0 keys | 5461 slots | 1 slaves.
127.0.0.1:7001 (5fc3b5dc...) -> 0 keys | 5461 slots | 1 slaves.
[OK] 0 keys in 3 masters.
0.00 keys per slot on average.
>>> Performing Cluster Check (using node 127.0.0.1:7002)
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   slots: (0 slots) slave
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.

4.3.10 添加空白主节点

配置文件 7007 /redis.conf

[root@localhost redis_cluster]# mkdir 7007
[root@localhost redis_cluster]# cp 7001/redis.conf 7007/redis.conf
[root@localhost redis_cluster]# cd 7007
[root@localhost 7007]# ls
redis.conf
[root@localhost 7007]# vim redis.conf
这里使用  :%s/7001/7007/g   对文件进行修改
​
启动7007服务
[root@localhost 7007]# /usr/lwl/soft/redis/bin/redis-server redis.conf 
[root@localhost 7007]# ps -ef|grep redis
root       7362      1  0 20:04 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7007 [cluster]

添加7007节点到集群

/usr/lwl/soft/redis/bin/redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7001
前面的IP加端口号是要添加的redis节点，后面的IP和端口号是集群中的任意一个节点。

 

检查节点的状态

[root@localhost 7007]# /usr/lwl/soft/redis/bin/redis-cli --cluster check 127.0.0.1:7001
127.0.0.1:7001 (5fc3b5dc...) -> 0 keys | 5461 slots | 1 slaves.
127.0.0.1:7007 (c00baee8...) -> 0 keys | 0 slots | 0 slaves.
127.0.0.1:7002 (16577299...) -> 0 keys | 5462 slots | 1 slaves.
127.0.0.1:7003 (e97325b5...) -> 0 keys | 5461 slots | 1 slaves.
[OK] 0 keys in 4 masters.
0.00 keys per slot on average.
>>> Performing Cluster Check (using node 127.0.0.1:7001)
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
M: c00baee8694cfdbd06d66c751adb42b34cb8716e 127.0.0.1:7007  #新添加的节点
   slots: (0 slots) master                  #新添加的节点为主节点，但是没有分配插槽
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   slots: (0 slots) slave
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.

4.3.11 为空白主节点配置从节点

新建一个目录7008，并配置服务

[root@localhost redis_cluster]# mkdir 7008
[root@localhost redis_cluster]# cp 7001/redis.conf 7008/redis.conf
[root@localhost redis_cluster]# cd 7008
[root@localhost 7008]# ls
redis.conf
[root@localhost 7008]# vim redis.conf
这里使用  :%s/7001/7008/g   对文件进行修改
​
启动7008服务
[root@localhost 7008]# /usr/lwl/soft/redis/bin/redis-server redis.conf 
[root@localhost 7008]# ps -ef|grep redis
root       7383      1  0 20:28 ?        00:00:00 /usr/lwl/soft/redis/bin/redis-server 0.0.0.0:7008 [cluster]

配置从节点

配置节点的命令：
/usr/lwl/soft/redis/bin/redis-cli --cluster add-node 127.0.0.1:7008 127.0.0.1:7002 --cluster-slave --cluster-master-id c00baee8694cfdbd06d66c751adb42b34cb8716e
/usr/lwl/soft/redis/bin/redis-cli：安装目录下的连接服务
127.0.0.1:7008：要添加的节点的IP地址及其端口号
127.0.0.1:7002：当前集群中的随意一个节点的IP地址及端口号
c00baee8694cfdbd06d66c751adb42b34cb8716e：要配置为主节点的ID（这里就是7007的一长串字符）
​
​
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli --cluster add-node 127.0.0.1:7008 127.0.0.1:7002 --cluster-slave --cluster-master-id c00baee8694cfdbd06d66c751adb42b34cb8716e
>>> Adding node 127.0.0.1:7008 to cluster 127.0.0.1:7002
>>> Performing Cluster Check (using node 127.0.0.1:7002)
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   slots: (0 slots) slave
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
M: c00baee8694cfdbd06d66c751adb42b34cb8716e 127.0.0.1:7007
   slots: (0 slots) master
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.
>>> Send CLUSTER MEET to node 127.0.0.1:7008 to make it join the cluster.
Waiting for the cluster to join
​
>>> Configure node as replica of 127.0.0.1:7007.
[OK] New node added correctly.

​

查看配置后的节点信息：7007和7008

[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli --cluster check 127.0.0.1:7002
127.0.0.1:7002 (16577299...) -> 0 keys | 5462 slots | 1 slaves.
127.0.0.1:7003 (e97325b5...) -> 0 keys | 5461 slots | 1 slaves.
127.0.0.1:7007 (c00baee8...) -> 0 keys | 0 slots | 1 slaves.
127.0.0.1:7001 (5fc3b5dc...) -> 0 keys | 5461 slots | 1 slaves.
[OK] 0 keys in 4 masters.
0.00 keys per slot on average.
>>> Performing Cluster Check (using node 127.0.0.1:7002)
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   slots: (0 slots) slave
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
S: f187059affef4732ac4ab5a776d3c7f7056834e0 127.0.0.1:7008  #从节点
   slots: (0 slots) slave
   replicates c00baee8694cfdbd06d66c751adb42b34cb8716e
M: c00baee8694cfdbd06d66c751adb42b34cb8716e 127.0.0.1:7007  #主节点
   slots: (0 slots) master
   1 additional replica(s)
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.

4.3.12 为空白主节点分配插槽

添加的主节点还不能使用，因为没有分配slots

slot的概念。slot对于Redis集群而言，就是一个存放数据的地方，就是一个槽。对于每一个Master而言，会存在一个slot的范围，而Slave则没有。在Redis集群中，依然是Master可以读、写，而Slave只读。

输入指令，开始进行分配插槽，最后的IP地址和端口号，只是进入分配的一个接口
[root@localhost 7008]# /usr/lwl/soft/redis/bin/redis-cli --cluster reshard 127.0.0.1:7002
>>> Performing Cluster Check (using node 127.0.0.1:7002)
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   slots: (0 slots) slave
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
S: f187059affef4732ac4ab5a776d3c7f7056834e0 127.0.0.1:7008
   slots: (0 slots) slave
   replicates c00baee8694cfdbd06d66c751adb42b34cb8716e
M: c00baee8694cfdbd06d66c751adb42b34cb8716e 127.0.0.1:7007
   slots: (0 slots) master
   1 additional replica(s)
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.
How many slots do you want to move (from 1 to 16384)?

 

最后会询问是否同意分配计划：

    Moving slot 8160 from 16577299b09086cf992b25ea047a53449064a62a
Do you want to proceed with the proposed reshard plan (yes/no)? yes
Moving slot 5461 from 127.0.0.1:7002 to 127.0.0.1:7007: 
Moving slot 5462 from 127.0.0.1:7002 to 127.0.0.1:7007: 

分配后查看节点状态：

[root@localhost 7008]# /usr/lwl/soft/redis/bin/redis-cli --cluster check 127.0.0.1:7001
127.0.0.1:7001 (5fc3b5dc...) -> 0 keys | 5461 slots | 1 slaves.
127.0.0.1:7007 (c00baee8...) -> 0 keys | 2700 slots | 1 slaves.
127.0.0.1:7002 (16577299...) -> 0 keys | 2762 slots | 1 slaves.
127.0.0.1:7003 (e97325b5...) -> 0 keys | 5461 slots | 1 slaves.
[OK] 0 keys in 4 masters.
0.00 keys per slot on average.
>>> Performing Cluster Check (using node 127.0.0.1:7001)
M: 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
M: c00baee8694cfdbd06d66c751adb42b34cb8716e 127.0.0.1:7007  #7007节点已经拥有了2700插槽
   slots:[5461-8160] (2700 slots) master
   1 additional replica(s)
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
S: f187059affef4732ac4ab5a776d3c7f7056834e0 127.0.0.1:7008
   slots: (0 slots) slave
   replicates c00baee8694cfdbd06d66c751adb42b34cb8716e
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[8161-10922] (2762 slots) master
   1 additional replica(s)
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
S: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004
   slots: (0 slots) slave
   replicates 5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.

4.3.13 删除节点

删除从节点

命令：redis客户端 --cluster del-node [集群中的任意一个节点：端口号] 被删除的节点的id
​
删除7008从节点
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli --cluster del-node 127.0.0.1:7002 f187059affef4732ac4ab5a776d3c7f7056834e0
>>> Removing node f187059affef4732ac4ab5a776d3c7f7056834e0 from cluster 127.0.0.1:7002
>>> Sending CLUSTER FORGET messages to the cluster...
>>> SHUTDOWN the node.

删除主节点

删除主节点需要先使用 reshard 把主节点的slots移到其他节点才可以
归还过程和分配插槽的过程一致
​
命令：redis客户端 --cluster del-node [集群中的任意一个节点：端口号] 被删除的节点的id
​
归还插槽后，删除主节点7007
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli --cluster del-node 127.0.0.1:7002 c00baee8694cfdbd06d66c751adb42b34cb8716e
>>> Removing node c00baee8694cfdbd06d66c751adb42b34cb8716e from cluster 127.0.0.1:7002
>>> Sending CLUSTER FORGET messages to the cluster...
>>> SHUTDOWN the node.

4.3.14 集群自动组网

集群关机之后，集群重启，只需要直接启动各个节点，不需要重新组网，redis会根据node.conf自动组网

前提是集群关闭的时候，是正常的关闭

正常的关闭，是先连接上之后，再shutdown进行关闭

4.3.15 故障检测

验证集群是否生效

关闭一个主节点查看对应的备用节点是不是能够顶替主节点成为主节点
​
注：
1、关闭的时候一定要使用shutdown命令不要使用kill命令
2、关闭主节点以后需要耐心等待一会儿 让他重新分配一下空间
​
关闭主机7001
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli  -p 7001 shutdown
查看节点状态
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli  --cluster check 127.0.0.1:7003
Could not connect to Redis at 127.0.0.1:7001: Connection refused
127.0.0.1:7003 (e97325b5...) -> 0 keys | 5461 slots | 1 slaves.
127.0.0.1:7002 (16577299...) -> 1 keys | 5462 slots | 1 slaves.
127.0.0.1:7004 (a14f9a6f...) -> 4 keys | 5461 slots | 0 slaves.
[OK] 5 keys in 3 masters.
0.00 keys per slot on average.
>>> Performing Cluster Check (using node 127.0.0.1:7003)
M: e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
M: 16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
S: 9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006
   slots: (0 slots) slave
   replicates e97325b5f5d020cfd4940421cdece4c3dc7e53d1
M: a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004   #7004代替7001成为了主机
   slots:[0-5460] (5461 slots) master
S: 68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005
   slots: (0 slots) slave
   replicates 16577299b09086cf992b25ea047a53449064a62a
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.
​
​
#重启7001服务
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-server 7001/redis.conf
#连接集群
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli -c -p 7002
127.0.0.1:7002> cluster nodes
a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 127.0.0.1:7004@17004 master - 0 1676783272031 9 connected 0-5460
#7001重启后成为7004的从机
5fc3b5dc391e2ef67d13bbbfd63d694319b66fd4 127.0.0.1:7001@17001 slave a14f9a6fed50a6f920e3ceb94f3bd06cd0febded 0 1676783271009 9 connected
16577299b09086cf992b25ea047a53449064a62a 127.0.0.1:7002@17002 myself,master - 0 1676783271000 8 connected 5461-10922
9e3f7eca62137abb171c19dc15af84ed4b0afb11 127.0.0.1:7006@17006 slave e97325b5f5d020cfd4940421cdece4c3dc7e53d1 0 1676783273045 6 connected
e97325b5f5d020cfd4940421cdece4c3dc7e53d1 127.0.0.1:7003@17003 master - 0 1676783270000 3 connected 10923-16383
68650f71b5631d9b73f719cf75cfc9da5ceb1997 127.0.0.1:7005@17005 slave 16577299b09086cf992b25ea047a53449064a62a 0 1676783271000 8 connected

存/取数据的时候查看对应的端口号
​
因为在测试存值时，添加了一个age属性的值存到了7001中，现在7004接替了7001，所以数据应该存在7004中
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli -c -p 7002
127.0.0.1:7002> get age
-> Redirected to slot [741] located at 127.0.0.1:7004  #重定向到7004中
"18"

4.4 slots介绍

[OK] All 16384 slots covered.
一个 Redis 集群包含 16384 个插槽（hash slot）， 数据库中的每个键都属于这 16384 个插槽的其中一个， 

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

集群中的每个节点负责处理一部分插槽。 举个例子， 如果一个集群可以有主节点， 其中：
节点 A 负责处理 0 号至 5460 号插槽。 
节点 B 负责处理 5461 号至 10922 号插槽。
节点 C 负责处理 10923 号至 16383 号插槽。

4.4.1 在集群中录入值

在redis-cli每次录入、查询键值，redis都会计算出该key应该送往的插槽，如果不是该客户端对应服务器的插槽，redis会报错，并告知应前往的redis实例地址和端口。

redis-cli客户端提供了 –c 参数实现自动重定向。

如 redis-cli  -c –p 7001登入后，再录入、查询键值对可以自动重定向。
​
[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli -c -p 7001
127.0.0.1:7001> set name lwl
-> Redirected to slot [5798] located at 127.0.0.1:7007  #设置值，计算出为7007，重定向到7007
OK
127.0.0.1:7007> set age 18
-> Redirected to slot [741] located at 127.0.0.1:7001   #设置值
OK
127.0.0.1:7001> get name
-> Redirected to slot [5798] located at 127.0.0.1:7007   #取出值
"lwl"

解决批量存储引发的问题

不在一个slot下的键值，是不能使用mget,mset等多键操作。
127.0.0.1:7007> mset k1 v1 k2 v2 k3 v3
(error) CROSSSLOT Keys in request don't hash to the same slot
​
​
解决批量存储  组 {组的名字}
可以通过{}来定义组的概念，从而使key中{}内相同内容的键值对放到一个slot中去。（按组分配插槽）
127.0.0.1:7007> mset k1{w} v1 k2{w} v2 k3{w} v3
-> Redirected to slot [3696] located at 127.0.0.1:7001
OK

4.4.2 集群的优点和不足

优点：

实现扩容
分摊压力
无中心配置相对简单

不足：

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

4.5 Redis的应用问题

4.5.1 缓存穿透

缓存穿透是指缓存和数据库中都没有的数据，导致所有的请求都落到数据库上，造成数据库短时间内承受大量请求而崩掉。

问题描述：

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

 

解决方案：

​

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


（1）    对空值缓存：如果一个查询返回的数据为空（不管是数据是否不存在），我们仍然把这个空结果（null）进行缓存，设置空结果的过期时间会很短，最长不超过五分钟
（2）    设置可访问的名单（白名单）：
使用bitmaps(位图)类型定义一个可以访问的名单，名单id作为bitmaps的偏移量，每次访问和bitmap里面的id进行比较，如果访问id不在bitmaps里面，进行拦截，不允许访问。
（3）    采用布隆过滤器：(布隆过滤器（Bloom Filter）是1970年由布隆提出的。它实际上是一个很长的二进制向量(位图)和一系列随机映射函数（哈希函数）。
布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误识别率和删除困难。)
将所有可能存在的数据哈希到一个足够大的bitmaps(位图)中，一个一定不存在的数据会被 这个bitmaps拦截掉，从而避免了对底层存储系统的查询压力。
（4）   进行实时监控：当发现Redis的命中率开始急速降低，需要排查访问对象和访问的数据，和运维人员配合，可以设置黑名单限制服务。
命中率 = 缓存次数/总的查询次数

4.5.2 缓存击穿

缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力。和缓存雪崩不同的是，缓存击穿指并发查同一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。

问题描述：

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

 

解决方案：

    key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：缓存被“击穿”的问题。
    
（1）预先设置热门数据：在redis高峰访问之前，把一些热门数据提前存入到redis里面，加大这些热门数据key的时长
（2）实时调整：现场监控哪些数据热门，实时调整key的过期时长
（3）使用锁：
    ①   就是在缓存失效的时候（判断拿出来的值为空），不是立即去加载数据库。
    ②   先使用缓存工具的某些带成功操作返回值的操作（比如Redis的SETNX）去set一个mutex key
    ③   当操作返回成功时，再进行load db的操作，并回设缓存,最后删除mutex key；
    ④   当操作返回失败，证明有线程在load db，当前线程睡眠一段时间再重试整个get缓存的方法。

 

4.5.3 缓存雪崩

缓存雪崩是指缓存同一时间大面积的失效，所以，后面的请求都会落到数据库上，造成数据库短时间内承受大量请求而崩掉。

问题描述：

    key对应的数据存在，但在redis中过期，此时若有大量并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
    缓存雪崩与缓存击穿的区别在于缓存雪崩针对很多key缓存，缓存击穿则是某一个key正常访问

 

当缓存失效的时候，大量访问进入到数据库存储层

 

解决方案:

缓存失效时的雪崩效应对底层系统的冲击非常可怕！
​

（1）   构建多级缓存架构：nginx缓存 + redis缓存 +其他缓存（ehcache等）
（2）   使用锁或队列：
用加锁或者队列的方式保证来保证不会有大量的线程对数据库一次性进行读写，从而避免失效时大量的并发请求落到底层存储系统上。不适用高并发情况
（3）   设置过期标志更新缓存：
记录缓存数据是否过期（设置提前量），如果过期会触发通知另外的线程在后台去更新实际key的缓存。
（4）   将缓存失效时间分散开：
比如我们可以在原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。

4.6 分布式锁

4.6.1 问题描述

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

分布式锁主流的实现方案：
1. 基于数据库实现分布式锁
2. 基于缓存（Redis等）
3. 基于Zookeeper
4. 基于redission实现分布式锁
​
​
每一种分布式锁解决方案都有各自的优缺点：
1. 性能：redis最高
2. 可靠性：zookeeper最高
3. 使用redission实现分布式锁，可以实现自动续期
这里，我们就基于redis实现分布式锁。

4.6.2 使用redis实现分布式锁

redis:命令
set命令的选项
EX second ：设置键的过期时间为 second 秒。 
    SET key value EX second 效果等同于 SETEX key second value 。
    
PX millisecond ：设置键的过期时间为 millisecond 毫秒。 
    SET key value PX millisecond 效果等同于 PSETEX key millisecond value 。
    
NX ：只在键不存在时，才对键进行设置操作。 
    SET key value NX 效果等同于 SETNX key value 。
    
XX ：只在键已经存在时，才对键进行设置操作。

实现图如下：使用nx命令即可

 

1. 多个客户端同时获取锁（setnx）
2. 获取成功，执行业务逻辑{从db获取数据，放入缓存}，执行完成释放锁（del）
3. 其他客户端等待重试

4.6.3 redis实现分布式锁步骤

1、创建SpringBoot项目

2、添加依赖

        <!--Java操作redis-->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
        </dependency>
        <!--封装了一些redis启动时要配置的信息-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
​
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
            <version>2.6.0</version>
        </dependency>

3、配置连接信息

#设置reis的索引
spring.redis.database=6
#设置连接redis的密码
spring.redis.password=密码
#设置的redis的服务器
spring.redis.host=192.168.111.127
#端口号
spring.redis.port=6379
#连接超时时间（毫秒）
spring.redis.timeout=1800000
#连接池最大连接数（使用负值表示没有限制）
spring.redis.lettuce.pool.max-active=20
#最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-wait=-1
#连接池中的最大空闲连接
spring.redis.lettuce.pool.max-idle=5
#连接池中的最小空闲连接
spring.redis.lettuce.pool.min-idle=0

4、配置类

​

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.JedisPoolConfig;
​
import java.time.Duration;
​
/**
 *这里虽然会报错，但是不影响运行
 */
​
@EnableCaching
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
    /**
     * 连接池的设置
     */
    @Bean
    public JedisPoolConfig getJedisPoolConfig() {
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        return jedisPoolConfig;
    }
​
​
    /**
     * RedisTemplate
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        // 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会跑出异常
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        template.setConnectionFactory(factory);
        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        return template;
    }
​
​
    /**
     * 缓存处理
     * @param factory
     * @return
     */
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置序列化（解决乱码的问题）,过期时间600秒
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofSeconds(600))
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    }
}

5、编写代码

先在redis中设置一个值：set num 10

[root@localhost redis_cluster]# /usr/lwl/soft/redis/bin/redis-cli -a lwl
Warning: Using a password with '-a' or '-u' option on the command line interface may not be safe.
127.0.0.1:6379> select 6
OK
127.0.0.1:6379> set num 10
OK

controller代码：

/**
 * 使用redis实现分布式锁
 */
@RestController
public class RedisLock {
    /**
     * 所有操作都封装在 RedisTemplate 中，所以将 RedisTemplate 作为bean注入进来
     */
    @Resource
    private RedisTemplate redisTemplate;
​
    @GetMapping("testLock")
    public String testLock(){
        //测试是否能连接redis数据库
        Object num = redisTemplate.opsForValue().get("num");
        System.out.println("num = " + num);
​
        //1、获取锁，这里的setIfAbsent方法就是setnx方法，只有当数据库中没有这个lock关键字时，才能够让返回值为true
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent("lock", "test");
        //2、根绝排它锁返回的结果进行操作
        if (aBoolean){
            //2.1.1 抢到排它锁，对数据进行处理
            redisTemplate.opsForValue().decrement("num");
            //2.1.2 将锁删除
            redisTemplate.delete("lock");
        }else {
            //2.2.1 没有抢到锁,线程进行休眠
            try {
                Thread.sleep(100);
                //2.2.2 休眠后还要继续去抢占锁,调用这个方法
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return null;
    }
}

4.6.4 分布式锁问题及优化

1、无法释放锁

进行压力测试时
问题：setnx刚好获取到锁，业务逻辑出现异常（比如算术异常），导致锁无法释放
​
解决：设置过期时间，自动释放锁。

设置过期时间有两种方式：

1. 首先想到通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
2. 在set时指定过期时间（推荐）

 

设置过期时间：

代码中设置过期时间

//第三个参数代表时间，最后一个参数代表单位 
Boolean lock = redisTemplate.opsForValue().setIfAbsent("lock","111",3,TimeUnit.SECONDS);
​
其他的都不变

2、释放时，释放错误

场景：如果业务逻辑的执行时间是7s。执行流程如下
1. index1业务逻辑没执行完，3秒后锁被自动释放。 
2. index2获取到锁，执行index1的业务逻辑，3秒后锁被自动释放。
3. index3获取到锁，执行index1的业务逻辑，1秒后执行完成。 
4. index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。
 
最终等于没锁的情况。
​
解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁

 

controller层代码：

/**
 * 使用redis实现分布式锁
 */
@RestController
public class RedisLock {
    /**
     * 所有操作都封装在 RedisTemplate 中，所以将 RedisTemplate 作为bean注入进来
     */
    @Resource
    private RedisTemplate redisTemplate;
​
    @GetMapping("testLock")
    public String testLock(){
        //获取uuid作为锁的唯一值，防止误删
        String uuid = UUID.randomUUID().toString();
        //1、获取锁，这里的setIfAbsent方法就是setnx方法，
        //设置过期时间为3秒
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);
        //2、根绝排它锁返回的结果进行操作
        if (aBoolean){
            //2.1.1 抢到排它锁，对数据进行处理
            redisTemplate.opsForValue().decrement("num");
            //2.1.2 删除之前判断删除的值是否等于当前锁的值
            if (uuid.equals((String)redisTemplate.opsForValue().get("num"))){
                //2.1.3 将锁删除
                redisTemplate.delete("lock");
            }
        }else {
            //2.2.1 没有抢到锁,线程进行休眠
            try {
                Thread.sleep(100);
                //2.2.2 休眠后还要继续去抢占锁,调用这个方法
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return null;
    }
}

3、删除操作缺乏原子性

场景：
    当前已经设置了过期时间，也设置了uuid，但是也会有极端情况
    假如当前index1的业务逻辑执行完，也已经判定了uuid是当前锁的uuid，但是过期时间到了，自动释放锁
    这时index2抢到了锁，执行了赋值操作，lock的value值已经换成了index2的uuid
    此时 index1 执行删除锁的操作，但是lock的值已经换成了index2的uuid值，所以删除的锁是index2的锁
​
解决：使用lua脚本保证删除操作的原子性
​
KEYS[1] 用来表示在redis 中用作键的参数占位，主要用来传递在redis 中的key。
ARGV[1] 用来表示在redis 中用作值的参数占位，主要用来传递在redis 中的value。

eval "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end" 1 aaa bbb
命令解读：
  eval：lua脚本的关键字
  引号中的语言：如果得到的这个key为aaa的value等于bbb，那么就删除aaa这个key并返回1，否则返回0
  1：代表希望返回1
  aaa   bbb：传入的参数，aaa代表key，bbb代表value
​
127.0.0.1:6379[6]> set aaa bbb
OK
127.0.0.1:6379[6]> keys *
1) "aaa"
2) "num"
127.0.0.1:6379[6]> eval "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end" 1 aaa bbb
(integer) 1
127.0.0.1:6379[6]> keys *
1) "num"

controller层代码：

/**
 * 使用redis实现分布式锁
 */
@RestController
public class RedisLock {
    /**
     * 所有操作都封装在 RedisTemplate 中，所以将 RedisTemplate 作为bean注入进来
     */
    @Resource
    private RedisTemplate redisTemplate;
​
    @GetMapping("testLock")
    public String testLock(){
        //获取uuid作为锁的唯一值，防止误删
        String uuid = UUID.randomUUID().toString();
        //1、获取锁，这里的setIfAbsent方法就是setnx方法，
        //设置过期时间为3秒
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent("lock", uuid,3, TimeUnit.SECONDS);
        //2、根绝排它锁返回的结果进行操作
        if (aBoolean){
            //2.1.1 抢到排它锁，对数据进行处理
            redisTemplate.opsForValue().decrement("num");
            /*使用lua脚本进行原子性操作*/
            //2.1.2 定义lua脚本：如果传入的key和value在数据库中是相对应的，那么就删除这个key并返回1，否则返回 0
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            //2.1.3 使用redis执行lua,指定脚本的返回值类型为Long
            DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
            redisScript.setScriptText(script);
            //2.1.4 因为删除判断的时候，返回的0会被其封装为指定的数据类型，如果不封装那么默认返回String 类型，
            //2.1.5 如果不封装那么返回字符串与 0 会有发生错误，所以设置一下返回值类型 为Long。
            redisScript.setResultType(Long.class);
            //2.1.6 第一个要是script 脚本 ，第二个需要判断的key，第三个就是key所对应的值。后面两个就相当于是参数
            redisTemplate.execute(redisScript, Arrays.asList(redisTemplate.opsForValue().get("num")),uuid);
        }else {
            //2.2.1 没有抢到锁,线程进行休眠
            try {
                Thread.sleep(100);
                //2.2.2 休眠后还要继续去抢占锁,调用这个方法
                testLock();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return null;
    }
}

4.6.5 lua脚本

Lua 是一个小巧的[脚本语言]，Lua脚本可以很容易的被C/C++ 代码调用，也可以反过来调用C/C++的函数，Lua并没有提供强大的库，一个完整的Lua解释器不过200k，所以Lua不适合作为开发独立应用程序的语言，而是作为嵌入式脚本语言。

很多应用程序、游戏使用LUA作为自己的嵌入式脚本语言，以此来实现可配置性、可扩展性。

这其中包括魔兽争霸地图、魔兽世界、博德之门、愤怒的小鸟等众多游戏插件或外挂。

lua脚本教程地址：Lua 教程_w3cschool

local userid=KEYS[1];  #定义的变量，值靠参数传递
local prodid=KEYS[2];  #定义的变量，值靠参数传递
​
#相当于 str="sk:"+prodid+":qt" ，这里的..相当于java里面的 “ + ”
local qtkey="sk:"..prodid..":qt"; 
local usersKey="sk:"..prodid..":usr"; 
​
#这里是判定是否存在这样一个key，如果存在这个key则返回1
local userExists=redis.call("sismember",usersKey,userid);
if tonumber(userExists)==1
 then 
  return 2;
end
​
#获取key的value值，必须是数字类型
local num= redis.call("get" ,qtkey);
#如果值<0 则返回0
if tonumber(num)<=0 then 
  return 0; 
else 
  #否则将该key的value值减1
  redis.call("decr",qtkey);
  #将userskey作为key，userid作为value值 添加到set集合中
  redis.call("sadd",usersKey,userid);
end
return 1;

lua脚本在redis中的优势

①将复杂的或者多步的redis操作，写为一个脚本，一次提交给redis执行，减少反复连接redis的次数。提升性能。
②LUA脚本是类似redis事务，有一定的原子性，不会被其他命令插队，可以完成一些redis事务性的操作。
③但是注意redis的lua脚本功能，只有在Redis 2.6以上的版本才可以使用。