Redis知识点总结

Lou.Chen

Redis

Redis

大约 85 分钟

1、Redis简介

1.1 海量用户/高并发带来的问题和现象

关系型数据库带来的问题：
性能瓶颈：磁盘IO性能低下
扩展瓶颈：数据关系复杂，扩展性差，不便于大规模集群

解决思路：
使用NoSQL
降低磁盘IO次数，越低越好 --> 内存存储
去除数据间关系，越简单越好 --> 不存储关系，仅存储数据

1.2 什么是NoSQL

NoSQL：即 Not-Only SQL（泛指非关系型的数据库），作为关系型数据库的补充。

作用：应对基于海量用户和海量数据前提下的数据处理问题。

特征
- 可扩容，可伸缩
- 大数据量下高性能
- 灵活的数据模型
- 高可用
常见NoSQL数据库
- Redis
- memcache
- HBase
- MongoDB

1.3 Redis简介

概念
- Redis (REmote DIctionary Server) 是用 C 语言开发的一个开源的高性能键值对（key-value）数据库。
特征
- 数据间没有必然的关联关系
- 内部采用单线程机制进行工作
- 高性能。官方提供测试数据，50个并发执行100000 个请求,读的速度是110000 次/s,写的速度是81000次/s。
- 多数据类型支持
- 持久化支持。可以进行数据灾难恢复

1.4 Redis的应用

为热点数据加速查询（主要场景），如热点商品、热点新闻、热点资讯、推广类等高访问量信息等
任务队列，如秒杀、抢购、购票排队等
即时信息查询，如各位排行榜、各类网站访问统计、公交到站信息、在线人数信息（聊天室、网站）、设备信号等
时效性信息控制，如验证码控制、投票控制等
分布式数据共享，如分布式集群架构中的 session 分离
消息队列
分布式锁

2、Redis的下载与安装

2.1 在线体验

Redis官网 http://try.redis.io/open in new window

2.2 安装包编译安装

准备gcc环境基于c++
```
yum install gcc-c++
```
redis >=6.0 会出现编译错误： https://blog.csdn.net/xixiyuguang/article/details/106612841open in new window

下载安装Redis

#下载安装包
wget http://download.redis.io/releases/redis-5.0.8.tar.gz
#解压
tar -zxvf redis-5.0.8.tar.gz 
#切换到解压目录
cd redis-5.0.8
#编译
make
#安装
make install

配置文件启动Redis
```
redis-server redis.conf
```
利用redis.config启动多个redis
redis-server redis-6379.config
redis-server redis-6380.config

后台启动Redis

#修改安装目录下的 redis.conf文件
daemonize no 变更为: daemonize yes  
#保存退出，再次启动
redis-server redis.conf

关闭Redis

#第一种方式：切换到Redis文件目录关闭
redis-cli shutdown

#第二种方式：使用kill强行关闭
#搜索到该进程
ps -ef|grep redis 
kill -9 'pid'

2.3 外网访问与密码配置

修改redis.conf配置文件

bind 127.0.0.1
- **解释：**redis 只接收来自于该 IP 地址列表的请求
- 解决办法：注释掉该行语句，即可接受所有IP
- 与protected-mode设置无关,即设置为yes依然指定的ip也可访问
protected-mode yes
保护模式：除本机以外，其他的都无法连接上
- 解释：默认是yes，即开启。
- 若为yes:
  - 即 bind、密码都没配置 --外网不可访问
- 若为no:
  - 即 bind、密码都没配置 --外网可访问
- 配置bind或者设置密码都会使protected-mode失效
```
requirepass 123456
```

进入命令行

进入命令行和关闭服务器：

redis-cli -a 密码 进入命令行

redis-cli -a 密码 shutdown 关闭服务器

进入命令行后的验证:

auth 密码

进入命令行后退出：

exit

2.4 Docker安装Redis

1)拉取redis镜像

docker pull redis

2)运行容器

挂载在宿主机上的文件必须先存在

docker run -d --restart=always --name myredis -p 6379:6379 
-v /opt/myredis/data:/data 
-v /opt/myredis/conf/redis.conf:/usr/local/etc/redis/redis.conf 
redis redis-server /usr/local/etc/redis/redis.conf

宿主机上的aof数据/opt/myredis/data对应容器的 /data (持久化文件)
宿主机上/opt/myredis/conf/redis.conf对应容器的 /usr/local/etc/redis/redis.conf (redis配置文件)
redis-server /usr/local/etc/redis/redis.conf 运行时启动redis

3)进入容器

docker exec -it myredis redis-cli

2.5 redis.config配置文件

2.5.1 服务端配置

daemonize yes/no
- no 以控制台形式启动
  - ⚠️ 若以控制台形式启动，则 logfile 选项必须注释，否则日志将不会打印，也不会写入到日志文件中。
- yes 以后台形式启动
以守护进程方式启动，使用本启动方式，redis将以服务的形式存在，日志将不再打印到命令窗口中
port 6379
设定当前服务启动端口号
dir "/自定义目录/redis/data"
设定当前服务文件保存位置，包含日志文件、持久化文件等
bind 127.0.0.1
绑定主机地址
databases 16
设置数据库数量，默认16

2.5.2 日志配置

logfile "6379.log"
设定日志文件名，便于查阅
loglevel debug|verbose|notice|warning
设置服务器以指定日志记录级别
注意：日志级别开发期设置为verbose即可，生产环境中配置为notice，简化日志输出量，降低写日志IO的频度

2.5.3 客户端配置

maxclients 0
设置同一时间最大客户端连接数，默认无限制。当客户端连接到达上限，Redis会关闭新的连接
timeout 300
客户端闲置等待最大时长，达到最大值后关闭连接。如需关闭该功能，设置为 0

2.5.4 多服务器快捷配置

include /path/server-端口号.conf
即引用通用的一些配置
导入并加载指定配置文件信息，用于快速创建redis公共配置较多的redis实例配置文件，便于维护

3、Redis基本操作

3.1 通用命令

清屏
clear
删除key
del [key]
序列化给定的key
dump [key]
判断指定的key是否存在
0不存在；1存在
exists [key]
查看key的有效期（秒）
-2已过期 -1永久有效
ttl [key]
查看key的有效期（毫秒）
pttl [key]
设置key的有效期
注意：如果key在过期时间被重新set了，那么过期时间会失效
1设置成功 0设置失败
expire [key] [seconds秒]
pexpire [key] [milliseconds毫秒]
expireat [key] [timestamp秒时间戳]
pexpireat [key] [milliseconds-timestamp毫秒时间戳]
移除一个key的过期时间
切换key从时效性转换为永久性
persist [key]
通配符查看所有的key
keys [pattern]
* 匹配任意数量的任意符号
? 配合一个任意符号
[] 匹配一个指定符号
keys * 查看所有key
keys k* 查看以k开头的所有key
keys *wo 查看所有以wo结尾的key
keys ???chen 查询前面三个任意字符开头，结尾chen
keys u[st]er:1 查询以u开头，中间包含s或t字符，并且以er:1结尾的
获取key的类型
type [key]
为key重命名
- rename [key] [newkey]
  若重命名的key存在则会覆盖已存在的key
- renamenx [key] [newkey]
  新的 key 不存在时修改 key 的名称
  修改成功时，返回 1 。如果 newkey 已经存在，返回 0 。

3.2 帮助命令

help [命令]

查看指定命令的用法

127.0.0.1:6379> help get

  GET key
  summary: Get the value of a key
  since: 1.0.0
  group: string

help @组名

help @string

查看所有string类型组下的所有命令

127.0.0.1:6379> help @string

  APPEND key value
  summary: Append a value to a key
  since: 2.0.0

  BITCOUNT key [start end]
  summary: Count set bits in a string
  since: 2.6.0

  BITFIELD key [GET type offset] [SET type offset value] [INCRBY type offset increment] [OVERFLOW WRAP|SAT|FAIL]
  summary: Perform arbitrary bitfield integer operations on strings
  since: 3.2.0

  BITOP operation destkey key [key ...]
  summary: Perform bitwise operations between strings
  since: 2.6.0

  ..........

help 空格 tab键 快速生成提示

3.3 数据操作指令

redis为每个服务提供有16个数据库，编号从0到15
每个数据库之间的数据相互独立

切换数据库
select [index]
查看库的键的数量
dbsize
删除当前数据库中的所有key
flushdb
删除所有数据库的key
flushall
移动键到其他数据库
move [key] [db编号]

4、五种基本数据类型操作

Redis数据存储格式

redis 自身是一个 Map，其中所有的数据都是采用 key : value 的形式存储
数据类型指的是存储的数据的类型，也就是 value 部分的类型，key 部分永远都是字符串

四种数据类型（list/set/zset/hash），在第一次使用时，如果容器不存在，就自动创建一个
四种数据类型（list/set/zset/hash），如果里边没有元素了，那么立即删除容器，释放内存。

key的设置与约定

数据库中的热点数据key命名惯例
表名：主键名：主键值：字段名
order:id：28231121：name
nwes:id：2321345：title

4.1 String

String 是 Redis 里边最最简单的一种数据结构。值通常使用字符串，如果字符串以整数的形式展示，可以作为数字操作使用

Redis 中的字符串是动态字符串，内部是可以修改的，像 Java 中的 StringBuffer，它采用分配冗余空间的方式来减少内存的频繁分配。在 Redis 内部结构中，一般实际分配的内存会大于需要的内存，当字符串小于 1M 的时候，扩容都是在现有的空间基础上加倍，扩容每次扩 1M 空间，最大 512M。

数据最大存储量：
521MB
数值计算最大范围（java中的long的最大值）:
9223372036854775807

**append **

追加操作

append [key] [value]

  append k1 lc
  append k1 .hello
  #获取键值
  get k1  #==>  'lc.hello'

set
给key赋值
set [key] [value]
```
set k2 33
get k2 # ==> '33'
```
get
获取key的value
get [key]
decr
可以实现对 value 的减 1 操作
前提是 value 是一个数字，如果 value 不是数字，会报错，如果 value 不存在，则会给一个默认的值为 0，在默认值的基础上减一。
decr [key]
```
decr k3 # ==> '-1'
```
```
set k2 33
decr k2 #==> '32'
```
incr
给某一个key的value自增
incr [key]
decrby
设置减的步长
decrby [key] [decrement]
```
set k2 100
decrby k2 10 #==>  '90'
```
incrby
给某一个key的value自增，并设置步长
incrby [key] [increment]
getrange
截取指定范围的字符串，相当于java中的substring
start 表示开始的位置(包含此位置)
end 表示结束的位置 -1 表示截取到最后一个(包含)
getrange [key] [start] [end]
```
set lou www.louchen.top
get lou #==> "www.louchen.top"
getrange lou 4 -1 #==> "louchen.top"
getrange lou 4 -5 #==> "louchen"
```
getset
获取并更新某一个key
getset [key] [value]
```
set k2 90
getset k2 100
get k2 #==> "100"
```
incrbyfloat
和incrby类似，步长可以设置为浮点数
incrbyfloat [key] [increment]
```
set k2 100
incrbyfloat k2 0.22
get k2 #==> 100.22
```
mget 和 mset
批量获取与批量设置
mget [key1] [key2] [key3] [key...]
mset [key1] [value1] [key2] [value2] [key...] [value..]
```
mset k1 11 k2 22 k3 33
mget k1 k2 k3 #===>  "11"  "22"  "33"
```
setex
给key设置value并同时设置过期时间(秒)
psetex [key] [seconds] [value]
```
set k1 10 woshi
#十秒后再获取
get k1 #==> "nil(表示不存在)"
```
psetex
给key设置value并同时设置过期时间(毫秒)
psetex [key] [millseconds] [value]
setnx
默认情况下set命令会覆盖已存在的key, setnx不会修改已存在的key
0表示修改失败
setnx [key] [value]
msetnx
批量设置，只要有一个已存在的key都不会修改
msetnx [key1] [value1] [key2] [value2] [key...] [value..]
setrange
覆盖一个已经存在的key的value，指定范围覆盖
offset为偏移量,若偏移量超过本身字符的长度，则超过的位置用0补齐
setrange [key] [offset] [value]
```
set k1 louchen
setrange k1 0 222
get k1 #==> "222chen"

setrange k1 10 333
get k1 #==> "222chen\x00\x00\x00333"
```
strlen
查看字符串总长度
strlen [key]

4.2 List

List集合中的数据可以重复

数据存储需求：存储多个数据，并对数据进入存储空间的顺序进行区分
需要的存储结构：一个存储空间保存多个数据，且通过数据可以体现进入顺序
list类型：保存多个数据，底层使用双向链表存储结构实现
数据总容量是有限的，最多2³² - 1 个元素 (4294967295)。

lrange 返回列表指定区间内的元素
返回列表所有元素 lrange [key] 0 -1
lrange [key] [start] [stop]
lpush 和 rpush
如果 key 不存在，那么在进行 push 操作前会创建一个空列表。如果 key 对应的值不是一个 list 的话，那么会返回一个错误。
- **lpush **
  将所有指定的值插入到存于 key 的列表的头部
  lpush [key] [value1] [value2] [value...]
```
127.0.0.1:6379> lpush k1 1 2 3 4 5
(integer) 5
127.0.0.1:6379> lrange k1 0 0
1)"5"
127.0.0.1:6379> lrange k1 0 -1
1)"5"
2)"4"
3)"3"
4)"2"
5)"1"
```
- rpush
  将所有指定的值插入到存于 key 的列表的尾部。
  rpush [key] [value1] [value2] [value...]
```
127.0.0.1:6379> rpush k2 1 2 3 4 5
(integer) 5
127.0.0.1:6379> lrange k2 0 0
1) "1"
127.0.0.1:6379> lrange k2 0 -1
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
```
rpop
移除并返回列表的尾元素
rpop [key]
lpop
移除并返回列表的头元素
lpop [key]
lindex
返回列表中下标为index的元素，只是查询
索引从0开始
lindex [key] [index]
llen
返回key中元素的个数
llen [key]
ltrim
对列表进行修剪
从指定start索引位置(包含此位置)到stop位置截取(包含此位置)
截取的范围值会覆盖原来的key
ltrim [key] [start] [stop]
blpop/brpop
阻塞式的弹出，相当于lpop的阻塞版
即当弹出所有元素时，再次弹出会阻塞此操作，等待下一次插入再弹出，阻塞时间设置为秒
blpop [key] [timeout秒]
brpoplpush
阻塞时间设置为秒，用于移除列表的最后一个元素，并将该元素添加到另一个列表的头部并返回。
brpoplpush [source] [destination] [timeout]
lrem
根据参数 COUNT 的值，移除列表中与参数 VALUE 相等的元素。
COUNT 的值可以是以下几种：
- count > 0 : 从表头开始向表尾搜索，移除与 VALUE 相等的元素，数量为 COUNT 。
- count < 0 : 从表尾开始向表头搜索，移除与 VALUE 相等的元素，数量为 COUNT 的绝对值。
- count = 0 : 移除表中所有与 VALUE 相等的值。
LREM [key] [count] [VALUE]

4.3 Set

Set集合中的数据不可重复

新的存储需求：存储大量的数据，在查询方面提供更高的效率
需要的存储结构：能够保存大量的数据，高效的内部存储机制，便于查询
set类型：与hash存储结构完全相同，仅存储键，不存储值（nil），并且值是不允许重复的
set 虽然与hash的存储结构相同，但是无法启用hash中存储值的空间

sadd
添加元素到集合
若有重复的value，则只会添加一个
sadd [key] [value1] [value2] [value...]
smembers
获取该key的所有元素
smembers [key]
srem
移除指定的元素,可以同时移除多个
srem [key] [member1] [member2]
sismember
返回某一个元素是否在集合中
0 代表不存在，1代表存在
sismember [key] [member]
scard
返回集合的数量
scard [key]
srandmember
随机返回一个或多个元素
count表示随机返回元素的个数。没有count则返回一个
srandmember [key] [count]
spop
随机返回并出栈一个元素
spop [key]
smove
把一个元素从一个集合移动到另一个集合
source:源集合
destination：目标集合
member：元素
smove [source] [destination] [member]
```
 sadd k1 a b c
 #将k1中的a元素移动到k2中
 smove k1 k2 a
```
sdiff
返回两个集合的差集
这里注意是key1-key2集合。除去key2在key1中的元素。反之类似
sidff [key1] [key2]
sinter
返回两个集合的交集
sinter [key1] [key2]
sdiffstore
类似于sdiff,只不过，计算出来的结果会保存在一个新的集合中
destination:保存的新集合的key
key1-key2的差集
sdiffstore [destination] [key1] [key2]
sinterstore
类似于sinter,只不过，计算出来的交集会保存在一个新的集合中
destination:保存的新集合的key
key1，key2的交集
sinterstore [destination] [key1] [key2]
sunion
求并集,不存储，只返回
sunion [key1] [key2] [key...]
sunionstore
求并集保存到新的集合中
destination:保存的新集合的key
sunion [destination] [key1] [key2]

4.4 Hash

在 Hash 结构中，key 是一个字符串，value 则是一个 key/value 键值对

底层使用哈希表结构实现数据存储

Hash存储结构优化：
如果field数量较少，存储结构优化为类数组结构
如果field数量较多，存储结构使用HashMap结构
每个 hash 可以存储 2³² - 1 个键值对

hset

添加值

hset [key] [field] [value]

hset k1 name lc
hset k1 age 18
hset k1 gender male

hget
获取值
hget [key] [field]
```
hget k1 name #==>  "lc"
```
hmset
批量设置
hmset [key] [field1] [value1] [field2] [value2]
```
hmset k2 name lcc age 18 gender femle
```

hmget

批量获取

hmget [key] [field1] [field2]

hmget k2 name age gender
------
1) "lcc"
2) "18"
3) "femle"

hdel
删除一个指定的field
hdel [key] [field1] [field2]
hsetnx
默认情况下，如果field存在则会覆盖已有的value。但是hsetn不能覆盖已有的field值
hsetnx [key] [field] [value]
hvals
获取所有的value
hvals [key]
hkeys
获取所有的key
hkeys [key]
hgetall
同时获取所有的key和value
hgetall [key]
hgetall 操作可以获取全部属性，如果内部field过多，遍历整体数据效率就很会低，有可能成为数据访问瓶颈
hexists
返回指定field是否存在
不存在返回0 存在返回1
hexists [key] [field]
hincrby
给指定的value自增1或指定步长
increment:步长
hincrby [key] [field] [increment]
hincrbyfloat
给指定的value自增指定浮点数
hincrby [key] [field] [increment]
hlen
返回某一个key中value的数量
hlen [key]
hstrlen
返回某一个key中的某一个field的字符串长度
hstrlen [key] [filed]

4.5 ZSet

也称之为有序的Set集合

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

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

集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是O(1)。集合中最大的成员数为 2³² - 1 (4294967295, 每个集合可存储40多亿个成员)。

score保存的数据存储空间是64位，如果是整数范围是-9007199254740992~9007199254740992
score保存的数据也可以是一个双精度的double值，基于双精度浮点数的特征，可能会丢失精度，使用时候要慎重
sorted_set 底层存储还是基于set结构的，因此数据不能重复，如果重复添加相同的数据，score值将被反复覆盖，保留最后一次修改的结果

zadd
将指定的元素添加到有序集合中
score：分数 作为值
member： 作为键
zadd [key] [score1] [member1] [score2] [member2]
```
zadd k1 10 m1 20 m2 30 m3 40 m4
```
zscore
返回member的score值
zscore [key] [member]
```
127.0.0.1:6379> zscore k1 m1
"10"
```
zrange
返回指定集合范围中的一组元素
withsocres: 返回score值(可选)
返回k1集合中的所有member和score
zcore k1 0 -1 withsocres
zcore [key] [start] [stop] withsocres
```
127.0.0.1:6379> zrange k1 0 -1 withscores
1) "m1"
2) "10"
3) "m2"
4) "20"
5) "m3"
6) "30"
7) "m4"
8) "40"
127.0.0.1:6379> zrange k1 0 -1
1) "m1"
2) "m2"
3) "m3"
4) "m4"
```
zrevrange
逆序（按照分数从大到小）返回指定集合范围中的一组元素
zrevrange [key] [start] [stop] withscores
zcard
返回元素个数
member键的个数
zcard [key]
zcount
返回score在某一个区间内的元素的个数
默认是闭区间即包含min和max
zcount [key] [min] [max]
```
#包含20，50
ZCOUNT k1 20 50 #==>  4
#在20到50之间 不包含20，50
zocunt k1 (20 (50 #==> 2
```
zrangebyscore
按照score的范围返回元素
withscores: 是否携带score(可选)
zrangebyscore [key] [min] [max] withscores
zrevrangebyscore
逆序按照score的返回元素
withscores: 是否携带score(可选)
zrevrangebyscore [key] [max] [min] WITHSCORES
zrank
返回元素的排名值(从小到大，从0开始)
zrank [key] [member]
```
zrank k1 v1 #==> 0
```
zrevrank
返回元素排名值(从大到小)
zrevrank [key] [memeber]
```
zrevrank k1 v1 #==> 3
```
zincrby
按照指定步长自增
increment: 步长
zincrby [key] [increment] [member]
zinterstore
求两个集合的交集并存入新的集合
destination：新的目标集合
numkeys：相同member的个数
把key1和key2中的members相同的元素把其score累加得到新的score
zinterstore [destination] [numkeys] [key1] [key2]
zunionstore
求两个集合的并集存入新的集合
destination：新的目标集合
numkeys：相同member的个数
把key1和key2中的members相同的元素把其score累加得到新的score
zunionstore [destination] [numkeys] [key1] [key2]
zrem
弹出指定member元素
zrem [key] [member1] [member2]
zlexcount
计算有序集合中成员数量
ZLEXCOUNT k1 - + 返回最小的到最大的成员数量(即全部)
zlexcount [key] [min] [max]
```
#统计v2到v3之间的成功数 包括v2和v3
zlexcount k1 [v2 [v3   #==> 2
```
zrangebylex
返回指定区间内的成员
zrangebylex [key] [min] [max]
```
ZRANGEBYLEX k1 - +
------------------
1) "v2"
2) "v3"
3) "v4"
```
zremrangebyscore
移除有序集中，指定分数（score）区间内的所有成员。
包含min和max
zremrangebyscore [key] [min] [max]
zremrangebyrank
按照排名移除范围数据
zremrangebyrank [key] [start] [stop]

5、使用Java连接Redis

5.1 Jedis连接

pom.xml

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.3.0</version>
            <type>jar</type>
            <scope>compile</scope>
        </dependency>

    public static void main(String[] args) {
//        1、构造一个jedis对象，并设置端口（默认6379）
        Jedis jedis=new Jedis("192.168.56.101",6379);
//        2、密码验证
        jedis.auth("123456");
//        3、测试是否连接成功
        String ping = jedis.ping();
//        输出pong 表示连接成功
        System.out.println(ping);

//        在jedis中，操作键值的方法和在Redis命令行中的api完成一致。这里不再过多赘述
//        jedis.set()
//        jedis.zadd()
//        jedis.hset()
//        jedis.lpush()
//        jedis.sadd()
    }

5.2 Jedis连接池的使用

在实际应用中，Jedis 实例我们一般都是通过连接池来获取，由于 Jedis 对象不是线程安全的，所以，当我们使用 Jedis 对象时，从连接池获取 Jedis，使用完成之后，再还给连接池。

    public static void main(String[] args) {
//        连接池配置对象
        GenericObjectPoolConfig config=new GenericObjectPoolConfig();
//        连接池最大空闲数
        config.setMaxIdle(300);
//        最大连接数
        config.setMaxTotal(1000);
//        连接最大等待时间（毫秒）。-1表示无限制
        config.setMaxWaitMillis(30000);
//        在空闲时检查有效性
        config.setTestOnBorrow(true);
        /**
         * 1、连接池配置
         * 2、redis地址
         * 3、端口
         * 4、超时时间
         * 5、密码
         */
        JedisPool jedisPool= new JedisPool(config, "192.168.56.101",6379,30000,"123456");
//         2、从连接池中获取一个jedis连接
        try(Jedis jedis = jedisPool.getResource()){
//            认证密码
            String ping=jedis.ping();
            System.out.println(ping);
        }
    }

5.3 Lettuce连接

pom.xml

	  <dependency>
            <groupId>io.lettuce</groupId>
            <artifactId>lettuce-core</artifactId>
            <version>5.1.2.RELEASE</version>
        </dependency>

    public static void main(String[] args) {
//        注意这里的redis密码可以写在url地址中
        RedisClient redisClient = RedisClient.create("redis://123456@192.168.56.101");
        StatefulRedisConnection<String, String> connect = redisClient.connect();
        RedisCommands<String, String> sync = connect.sync();
        sync.set("name", "lc");
        String s = sync.get("name");
        System.out.println(s);
    }

6、数据持久化

将当前数据状态进行保存，快照形式，存储数据结果，存储格式简单，关注点在数据
将数据的操作过程进行保存，日志形式，存储操作过程，存储格式复杂，关注点在数据的操作过程

6.1 RDB

默认在启动的时候会去加载dump.rdb，恢复数据

6.1.1 save指令

save
手动执行一次保存操作
save指令相关配置
- dbfilename dump.rdb
  说明：设置本地数据库文件名，默认值为 dump.rdb
  经验：通常设置为dump-端口号.rdb
- dir
  说明：设置存储.rdb文件的路径
  经验：通常设置成存储空间较大的目录中，目录名称data
- rdbcompression yes
  说明：设置存储至本地数据库时是否压缩数据，默认为 yes，采用 LZF 压缩
  经验：通常默认为开启状态，如果设置为no，可以节省 CPU 运行时间，但会使存储的文件变大（巨大）
- rdbchecksum yes
  说明：设置是否进行RDB文件格式校验，该校验过程在写文件和读文件过程均进行
  经验：通常默认为开启状态，如果设置为no，可以节约读写性过程约10%时间消耗，但是存储一定的数据损坏风险
save指令的执行会阻塞当前Redis服务器，直到当前RDB过程完成为止，有可能会造成长时间阻塞，线上环境不建议使用。

6.1.2 bgsave指令

bgsave
后台执行一个保存操作
bgsave指令工作原理
bgsave命令是针对save阻塞问题做的优化。Redis内部所有涉及到RDB操作都采用bgsave的方式，save命令可以放弃使用。
bgsave指令相关配置
- stop-writes-on-bgsave-error yes
  说明：后台存储过程中如果出现错误现象，是否停止保存操作
  经验：通常默认为开启状态

6.1.3 通过redis.conf配置save执行

save [second] [changes]
- **作用：**满足限定时间范围内key的变化数量达到指定数量即进行持久化
- 参数：
  - second：监控时间范围
  - changes：监控key的变化量
- 范例：
  save 900 1 900s内 1个key变化
  save 300 10 300s内 10个key变化
  save 60 10000 60s内 10000key变化
配置原理
即对key的增删改才会记录为一次key的变化。
save配置要根据实际业务情况进行设置，频度过高或过低都会出现性能问题，结果可能是灾难性的
save配置中对于second与changes设置通常具有互补对应关系，尽量不要设置成包含性关系
save配置启动后执行的是bgsave操作

RDB三种启动方式对比

RDB特殊启动方式

服务器运行过程中重启
debug reload
关闭服务器时指定保存数据
shutdown save
默认情况下执行shutdown命令时，自动执行bgsave(如果没有开启AOF持久化功能)

RDB的

RDB优缺点

优点
- RDB是一个紧凑压缩的二进制文件，存储效率较高
- RDB内部存储的是redis在某个时间点的数据快照，非常适合用于数据备份，全量复制等场景
- RDB恢复数据的速度要比AOF快很多
- 应用：服务器中每X小时执行bgsave备份，并将RDB文件拷贝到远程机器中，用于灾难恢复。
缺点
- RDB方式无论是执行指令还是利用配置，无法做到实时持久化，具有较大的可能性丢失数据
- bgsave指令每次运行要执行fork操作创建子进程，要牺牲掉一些性能
- Redis的众多版本中未进行RDB文件格式的版本统一，有可能出现各版本服务之间数据格式无法兼容现象

6.2 AOF

RDB存储的弊端
存储数据量较大，效率较低基于快照思想，每次读写都是全部数据，当数据量巨大时，效率非常低
大数据量下的IO性能较低
基于fork创建子进程，内存产生额外消耗
宕机带来的数据丢失风险
解决思路
不写全数据，仅记录部分数据
降低区分数据是否改变的难度，改记录数据为记录操作过程
对所有操作均进行记录，排除丢失数据的风险

6.2.1 AOF概念

AOF(append only file)持久化：以独立日志的方式记录每次写命令，重启时再重新执行AOF文件中命令达到恢复数据的目的。与RDB相比可以简单描述为改记录数据为记录数据产生的过程
AOF的主要作用是解决了数据持久化的实时性，目前已经是Redis持久化的主流方式

6.2.2 AOF写数据的三种策略appendfsync

always(每次）
每次写入操作均同步到AOF文件中，数据零误差，性能较低
everysec（每秒）
推荐使用，也是默认配置
每秒将缓冲区中的指令同步到AOF文件中，数据准确性较高，性能较高 在系统突然宕机的情况下丢失1秒内的数据
no（系统控制）
由操作系统控制每次同步到AOF文件的周期，整体过程不可控

6.2.3 redis.conf中AOF相关配置

appendonly yes|no
是否开启AOF持久化功能，默认为不开启状态
appendfsync always|everysec|no
AOF写数据策略
appendfilename [filename]
AOF持久化文件名，默认文件名未appendonly.aof，建议配置为appendonly-端口号.aof
dir [目录]
AOF持久化文件保存路径，与RDB持久化文件保持一致即可

6.2.4 AOF重写

AOF写数据时遇到的问题

什么是AOF重写

随着命令不断写入AOF，文件会越来越大，为了解决这个问题，Redis引入了AOF重写机制压缩文件体积。AOF文件重写是将Redis进程内的数据转化为写命令同步到新AOF文件的过程。简单说就是将对同一个数据的若干个条命令执行结果转化成最终结果数据对应的指令进行记录。

AOF重写作用

降低磁盘占用量，提高磁盘利用率
提高持久化效率，降低持久化写时间，提高IO性能
降低数据恢复用时，提高数据恢复效率

AOF重写规则

进程内已超时的数据不再写入文件
忽略无效指令，重写时使用进程内数据直接生成，这样新的AOF文件只保留最终数据的写入命令如del key1、 hdel key2、srem key3、set key4 111、set key4 222等
对同一数据的多条写命令合并为一条命令如lpush list1 a、lpush list1 b、 lpush list1 c 可以转化为：lpush list1 a b c。为防止数据量过大造成客户端缓冲区溢出，对list、set、hash、zset等类型，每条指令最多写入64个元素

AOF重写方式

手动重写
bgrewriteaof
- bgrewriteaof重写原理
自动重写
- 自动重写触发条件设置
  auto-aof-rewrite-min-size [size]
  auto-aof-rewrite-percentage [percentage]
- 自动重写触发比对参数
  aof_current_size
  aof_base_size
  - 运行指令**info [Persistence]**获取具体信息
- 自动重写触发条件

AOF重写流程

6.3 RDB和AOF的区别

RDB和AOF的选择

对数据非常敏感，建议使用默认的AOF持久化方案
- AOF持久化策略使用everysecond，每秒钟fsync一次。该策略redis仍可以保持很好的处理性能，当出现问题时，最多丢失0-1秒内的数据。
- 注意：由于AOF文件存储体积较大，且恢复速度较慢
数据呈现阶段有效性，建议使用RDB持久化方案
- 数据可以良好的做到阶段内无丢失（该阶段是开发者或运维人员手工维护的），且恢复速度较快，阶段点数据恢复通常采用RDB方案
- 注意：利用RDB实现紧凑的数据持久化会使Redis降的很低，慎重总结：
综合比对
- RDB与AOF的选择实际上是在做一种权衡，每种都有利有弊
- 如不能承受数分钟以内的数据丢失，对业务数据非常敏感，选用AOF
- 如能承受数分钟以内的数据丢失，且追求大数据集的恢复速度，选用RDB
- 灾难恢复选用RDB
- 双保险策略，同时开启 RDB 和 AOF，重启后，Redis优先使用 AOF 来恢复数据，降低丢失数据的量

7、Redis事务

7.1 基本概念

redis事务就是一个命令执行的队列，将一系列预定义命令包装成一个整体（一个队列）。当执行时，一次性按照添加顺序依次执行，中间不会被打断或者干扰。Redis 事务的本质是一组命令的集合。事务支持一次执行多个命令，一个事务中所有命令都会被序列化。在事务执行过程，会按照顺序串行化执行队列中的命令，其他客户端提交的命令请求不会插入到事务执行命令序列中。

总结说：redis事务就是一次性、顺序性、排他性的执行一个队列中的一系列命令。

7.2 事务的基本操作

开启事务
multi
设定事务的开启位置，此指令执行后，后续的所有指令均加入到事务中
执行事务
exec
设定事务的结束位置，同时执行事务。与multi成对出现，成对使用

注意：加入事务的命令暂时进入到任务队列中，并没有立即执行，只有执行exec命令才开始执行

取消事务
discard
终止当前事务的定义，发生在multi之后，exec之前

Redis事务不保证原子性

Redis中，单条命令是原子性执行的，但事务不保证原子性，且没有回滚。事务中任意命令执行失败，其余的命令仍会被执行。注意，Redis 中的事务并不能算作原子性。它仅仅具备隔离性，也就是说当前的事务可以不被其他事务打断。
由于每一次事务操作涉及到的指令还是比较多的，为了提高执行效率，我们在使用客户端的时候，可以通过 pipeline 来优化指令的执行。

127.0.0.1:6379> multi	#开启事务
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> incr k1	 #自增1，操作失败
QUEUED
127.0.0.1:6379> exec  #提交事务
1) OK
2) OK
3) (error) ERR value is not an integer or out of range
127.0.0.1:6379> get k1  
"v1"
127.0.0.1:6379> get k2
"v2"

已经执行完毕的命令对应的数据不会自动回滚，需要程序员自己在代码中实现回滚。

7.3 事务的工作流程

7.4 事务监听watch

因为Redis没有事务的隔离级别的概念，所以会带来以下问题：

银行转账案例：
初始化余额
set money 100
A线程开启事务进行取钱操作
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set money 0  #还未提交事务
B线程开启事务进行存钱操作
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set money 900
QUEUED
127.0.0.1:6379> exec #已提交事务
1) OK
B线程提交完事务之后，A线程提交事务
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set money 0
QUEUED
127.0.0.1:6379> exec  #提交事务
1) OK
A线程和B线程分别获取值，结果均为0

解决方案

watch [key1] [key2]
对 key 添加监视锁，在执行exec前如果key发生了变化，终止事务执行
一但执行 EXEC 开启事务的执行后，无论事务使用执行成功， WARCH 对变量的监控都将被取消。
故当事务执行失败后，需重新执行WATCH命令对变量进行监控，并开启新的事务进行操作。
unwatch
取消对所有key的监视

银行转账案例改造
初始化余额
set money 100
开启对money的监听(必须在multi开启事务之前)
watch money
A线程开启事务进行取钱操作
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set money 0  #还未提交事务
B线程开启事务进行存钱操作
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set money 900
QUEUED
127.0.0.1:6379> exec #已提交事务
1) OK
B线程提交完事务之后，A线程提交事务
127.0.0.1:6379> multi #开启事务
OK 
127.0.0.1:6379> set money 0  
QUEUED
127.0.0.1:6379> exec   #提交事务
(nil)   #修改失败 自动退出当前事务
A线程和B线程分别获取值，结果均为900

7.5 分布式锁

7.5.1 问题场景

场景：例如一个简单的用户操作，一个线城去修改用户的状态，首先从数据库中读出用户的状态，然后在内存中进行修改，修改完成后，再存回去。在单线程中，这个操作没有问题，但是在多线程中，由于读取、修改、存这是三个操作，不是原子操作，所以在多线程中，这样会出问题。

对于这种问题，我们可以使用分布式锁来限制程序的并发执行。

✔️解决方案：

设置一个公共锁，获取到锁才去执行修改操作，否则阻塞等待

setnx [key] [value]
使用 setnx 设置一个公共锁
利用setnx命令的返回值特征，有值则返回设置失败，无值则返回设置成功
- 对于返回设置成功的，拥有控制权，进行下一步的具体业务操作
- 对于返回设置失败的，不具有控制权，排队或等待
执行完事务操作后，然后del释放该锁

⚠️以上方案会出现的问题：

如果我们的业务代码在执行的过程中抛出异常或者挂了，这样会导致del指令没有被调用(没有释放锁)，那么后面的请求全部堵塞在这，锁永远得不到释放
这里我们的解决方式就是在获得锁后，给锁添加过期时间，确保锁在一定的时间之后，能够得到释放。

✔️解决方案：

设置一个公共锁，给锁添加一个过期时间，获取到锁才去执行修改操作，否则阻塞等待。若时间过期，则自动释放该锁

expire [key] [second]
pexpire [lock-key] [milliseconds]
使用 expire 为锁key添加时间限定，到时不释放，放弃锁
由于操作通常都是微秒或毫秒级，因此该锁定时间不宜设置过大。具体时间需要业务测试后确认。
- 例如：持有锁的操作最长执行时间127ms，最短执行时间7ms。
- 测试百万次最长执行时间对应命令的最大耗时，测试百万次网络延迟平均耗时
- 锁时间设定推荐：最大耗时120%+平均网络延迟110%
- 如果业务最大耗时<<网络平均延迟，通常为2个数量级，取其中单个耗时较长即可

⚠️上述方案的问题：获得锁和设置过期时间不能保证原子性

在获得锁和设置锁的过期时间如果服务器挂掉，这个时候锁被占用，过期时间也没有设置，那么也会发生死锁，因为获取锁和设置过期时间为两个操作，不具备原子性

✔️最终解决方式：让setnx和exprie同时执行，即在获得锁的同时，同时设置锁的过期时间，保证原子性

7.5.2 实现方法

❓获得锁同时设置锁过期时间

为了解决这个问题，从 Redis2.8 开始，setnx 和 expire 可以通过一个命令一起来执行了

String lock = jedis.set("lock", "1", new SetParams().nx().ex(3));

public static void main(String[] args) {
//        连接池配置对象
        GenericObjectPoolConfig config=new GenericObjectPoolConfig();
//        连接池最大空闲数
        config.setMaxIdle(300);
//        最大连接数
        config.setMaxTotal(1000);
//        连接最大等待时间（毫秒）。-1表示无限制
        config.setMaxWaitMillis(30000);
//        在空闲时检查有效性
        config.setTestOnBorrow(true);
        /**
         * 1、连接池配置
         * 2、redis地址
         * 3、端口
         * 4、超时时间
         * 5、密码
         */
        JedisPool jedisPool= new JedisPool(config, "192.168.56.101",6379,30000,"123456");
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        //模拟多用户请求
        for (int i = 0; i < 20; i++) {
            int finalI = i;
            executorService.execute(()->{
                try(Jedis jedis = jedisPool.getResource()){
                    //money初始值为10
                    if (Integer.parseInt(jedis.get("money"))>0) {
                        //这里我们可以在set操作的同时，同时加入其他参数：nx不存在时操作，ex设置过期时间
                        //这样我们可以防止在获得锁和设置过期时间，如果服务器挂掉，则这个时候锁被占用，无法及时得到释放，从而造成死锁。因为获得锁和设置过期时间是两个操作，不具备原子性。
                        String lock = jedis.set("lock", "1", new SetParams().nx().ex(3));
                        if (lock != null && lock.equals("OK")) {
                            //给锁添加一个过期时间，防止应用在运行过程中抛出异常导致锁无法及时得到释放
                            jedis.expire("lock", 3);
                            Long money = jedis.decr("money");
                            System.out.println("客户"+ finalI +"进行了取钱1的操作,剩余"+money);
                            //释放锁
                            jedis.del("lock");
                        }else{
                            System.out.println(finalI+"正在等待..");
                        }
                    }else{
                        System.out.println("余额为0");
                    }
                }
            });
        }

    }

⚠️上述代码存在超时问题，若业务代码执行的时间超过了锁的过期时间，那么会出现线程紊乱

第一个线程首先获得锁，然后开始执行业务代码，这是执行业务代码的时间超过了锁设定的过期时间，那么该线程还未执行完业务代码该锁就被释放了。此时第二个线程获取到锁开始执行，此时该线程执行了几秒之后，第一个线程也执行完了，那么该第一个线程就会释放锁(del)，但是注意，它释放的这个锁是第二个线程的锁，第三个第四个线程也如此，那么就会造成线程紊乱，修改紊乱。

7.6.3 解决超时问题

尽量避免在获取锁的时候，执行耗时操作
可以在锁上考虑，将锁的value设置为一个随机字符串，每次释放锁的时候，都去比较随机字符串是否一致，如果一致，再去释放锁，否则，不释放。

对于第二种方案，由于释放锁的时候，要去查看锁的 value，第二个比较 value 的值是否正确，第三步释放锁，有三个步骤，很明显三个步骤不具备原子性，为了解决这个问题，得引入Lua脚本

Lua 脚本的优势：

使用方便，Redis 中内置了对 Lua 脚本的支持。
Lua 脚本可以在 Redis 服务端原子的执行多个 Redis 命令。
由于网络在很大程度上会影响到 Redis 性能，而使用 Lua 脚本可以让多个命令一次执行，可以有效解决网络给 Redis 带来的性能问题。

在 Redis 中，使用 Lua 脚本，大致上两种思路：

提前在 Redis 服务端写好 Lua 脚本，然后在 Java 客户端去调用脚本（推荐）。
可以直接在 Java 端去写 Lua 脚本，写好之后，需要执行时，每次将脚本发送到 Redis 上去执行。

1️⃣在Redis服务器中创建Lua脚本

切换到redis安装目录，创建存储lua的文件夹
创建vi dustrlock.lua文件
添加内容：
- ```
if redis.call("get",KEYS[1])==ARGV[1] then
   return redis.call("del",KEYS[1])
else
   return 0
end
```
  reids.call 表示调用get操作方式，KEYS[1]表示传过来要操作的key(可以有多个key，索引从1开始)，ARGV表示传过来的其他参数。
给Lua脚本求一个SHA1和(相当于给该lua文件算出一个标识符)
redis-cli script load "$(cat lua/dustrlock.lua)"
返回的标识符：
9d0abd0b3b3bfd1b5294f957dcab483e58b97c84
script load相当于这个命令会在 Redis 服务器中缓存 Lua 脚本，并返回脚本内容的 SHA1 校验和，然后在 Java 端调用时，传入 SHA1 校验和作为参数，这样 Redis 服务端就知道执行哪个脚本了。

    public static void main(String[] args) {
//        连接池配置对象
        GenericObjectPoolConfig config=new GenericObjectPoolConfig();
//        连接池最大空闲数
        config.setMaxIdle(300);
//        最大连接数
        config.setMaxTotal(1000);
//        连接最大等待时间（毫秒）。-1表示无限制
        config.setMaxWaitMillis(30000);
//        在空闲时检查有效性
        config.setTestOnBorrow(true);
        /**
         * 1、连接池配置
         * 2、redis地址
         * 3、端口
         * 4、超时时间
         * 5、密码
         */
        JedisPool jedisPool= new JedisPool(config, "192.168.56.101",6379,30000,"123456");
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        //模拟多用户请求
        for (int i = 0; i < 20; i++) {
            int finalI = i;
            executorService.execute(()-> {
                try (Jedis jedis = jedisPool.getResource()) {
                    //money初始值为10
                    if (Integer.parseInt(jedis.get("money")) > 0) {
                        //获取随机字符串作为锁的值
                        String lockValue = UUID.randomUUID().toString();
                        //这里我们可以在set操作的同时，同时加入其他参数：nx不存在时操作，ex设置过期时间
                        //这样我们可以防止在获得锁和设置过期时间，如果服务器挂掉，则这个时候锁被占用，无法及时得到释放，从而造成死锁。因为获得锁和设置过期时间是两个操作，不具备原子性。
                        String lock = jedis.set("lock", lockValue, new SetParams().nx().ex(3));
                        if (lock != null && lock.equals("OK")) {
                            //给锁添加一个过期时间，防止应用在运行过程中抛出异常导致锁无法及时得到释放
                            jedis.expire("lock", 3);
                            Long money = jedis.decr("money");
                            System.out.println("客户" + finalI + "进行了取钱1的操作,剩余" + money);
                            //释放锁的操作由Lua代替
                            jedis.evalsha("9d0abd0b3b3bfd1b5294f957dcab483e58b97c84", Arrays.asList("lock"), Arrays.asList(lockValue));
                        } else {
                            System.out.println(finalI + "正在等待..");
                        }
                    }
                }
            });
        }

2️⃣直接在Java代码中编写Lua脚本

    public static void main(String[] args) {
//        连接池配置对象
        GenericObjectPoolConfig config = new GenericObjectPoolConfig();
//        连接池最大空闲数
        config.setMaxIdle(300);
//        最大连接数
        config.setMaxTotal(1000);
//        连接最大等待时间（毫秒）。-1表示无限制
        config.setMaxWaitMillis(30000);
//        在空闲时检查有效性
        config.setTestOnBorrow(true);
        /**
         * 1、连接池配置
         * 2、redis地址
         * 3、端口
         * 4、超时时间
         * 5、密码
         */
        JedisPool jedisPool = new JedisPool(config, "192.168.56.101", 6379, 30000, "123456");
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        //模拟多用户请求
        for (int i = 0; i < 20; i++) {
            int finalI = i;
            executorService.execute(() -> {
                try (Jedis jedis = jedisPool.getResource()) {
                    //money初始值为10
                    if (Integer.parseInt(jedis.get("money")) > 0) {
                        //获取随机字符串作为锁的值
                        String lockValue = UUID.randomUUID().toString();
                        //这里我们可以在set操作的同时，同时加入其他参数：nx不存在时操作，ex设置过期时间
                        //这样我们可以防止在获得锁和设置过期时间，如果服务器挂掉，则这个时候锁被占用，无法及时得到释放，从而造成死锁。因为获得锁和设置过期时间是两个操作，不具备原子性。
                        String lock = jedis.set("lock", lockValue, new SetParams().nx().ex(3));
                        if (lock != null && lock.equals("OK")) {
                            //给锁添加一个过期时间，防止应用在运行过程中抛出异常导致锁无法及时得到释放
                            jedis.expire("lock", 3);
                            Long money = jedis.decr("money");
                            System.out.println("客户" + finalI + "进行了取钱1的操作,剩余" + money);
                            //释放锁的操作由Lua代替
                            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
                            jedis.eval(script, Arrays.asList("lock"), Arrays.asList(lockValue));
                        } else {
                            System.out.println(finalI + "正在等待..");
                        }
                    }
                }
            });
        }
    }

7.6 Redisson实现分布式锁

官网：https://github.com/redisson/redisson/wiki/目录open in new window

参考：https://www.cnblogs.com/qdhxhz/p/11046905.htmlopen in new window

8、删除策略

Redis中的数据特征

Redis是一种内存级数据库，所有数据均存放在内存中，内存中的数据可以通过TTL指令获取其状态
XX ：具有时效性的数据
-1 ：永久有效的数据
-2 ：已经过期的数据或被删除的数据或未定义的数据

8.1 定时删除

创建一个定时器，当key设置有过期时间，且过期时间到达时，由定时器任务立即执行对键的删除操作
优点：节约内存，到时就删除，快速释放掉不必要的内存占用
**缺点：**CPU压力很大，无论CPU此时负载量多高，均占用CPU，会影响redis服务器响应时间和指令吞吐量
总结：用处理器性能换取存储空间（拿时间换空间）

8.2 惰性删除

数据到达过期时间，不做处理。等下次访问该数据时
- 如果未过期，返回数据
- 发现已过期，删除，返回不存在
优点：节约CPU性能，发现必须删除的时候才删除
缺点：内存压力很大，出现长期占用内存的数据
总结：用存储空间换取处理器性能（拿时间换空间）

8.3 定期删除

原理：

Redis启动服务器初始化时，读取配置server.hz的值，默认为10
每秒钟执行server.hz次serverCron()
--> databasesCron()
--> activeExpireCycle()
activeExpireCycle()对每个expires[*]逐一进行检测，每次执行250ms/server.hz
对某个expires[*]检测时，随机挑选W个key检测
- 如果key超时，删除key
- 如果一轮中删除的key的数量>W*25%，循环该过程
- 如果一轮中删除的key的数量≤W*25%，检查下一个expires[*]，0-15循环
- W取值=ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP属性值
参数current_db用于记录activeExpireCycle() 进入哪个expires[*] 执行
如果activeExpireCycle()执行时间到期，下次从current_db继续向下执行

总结：

周期性轮询redis库中的时效性数据，采用随机抽取的策略，利用过期数据占比的方式控制删除频度
特点：
- CPU性能占用设置有峰值，检测频度可自定义设置
- 内存压力不是很大，长期占用内存的冷数据会被持续清理
周期性抽查存储空间（随机抽查，重点抽查）

8.4 三种删除策略对比

删除策略	内存占用情况	CPU利用情况	性能
定时删除	节约内存，无占用	不分时段占用CPU资源，频度高	拿时间换空间
惰性删除	内存占用严重	延时执行，CPU利用率高	拿空间换时间
定期删除	内存定期随机清理	每秒花费固定的CPU资源维护内存	随机抽查，重点抽查

8.5 逐出算法

❓ 当新数据进入redis时，如果内存不足怎么办？

Redis使用内存存储数据，在执行每一个命令前，会调用freeMemoryIfNeeded()检测内存是否充足。如果内存不满足新加入数据的最低存储要求，redis要临时删除一些数据为当前指令清理存储空间。清理数据的策略称为逐出算法。
注意：逐出数据的过程不是100%能够清理出足够的可使用的内存空间，如果不成功则反复执行。当对所有数据尝试完毕后，如果不能达到内存清理的要求，将出现错误信息。
- (error) OOM command not allowed when used memory >'maxmemory'`

相关配置：

最大可使用内存
maxmemory
占用物理内存的比例，默认值为0，表示不限制。生产环境中根据需求设定，通常设置在50%以上。
每次选取待删除数据的个数
maxmemory-samples
选取数据时并不会全库扫描，导致严重的性能消耗，降低读写性能。因此采用随机获取数据的方式作为待检测删除数据
逐出策略
maxmemory-policy
达到最大内存后的，对被挑选出来的数据进行删除的策略

影响数据逐出的相关配置:

检测易失数据（可能会过期的数据集server.db[i].expires ）
- volatile-lru：挑选最近最少使用的数据淘汰 volatile-lfu：挑选最近使用次数最少的数据淘汰 volatile-ttl：挑选将要过期的数据淘汰 volatile-random：任意选择数据淘汰
检测全库数据（所有数据集server.db[i].dict ）
- allkeys-lru：挑选最近最少使用的数据淘汰 allkeys-lfu：挑选最近使用次数最少的数据淘汰 allkeys-random：任意选择数据淘汰
放弃数据驱逐
- no-enviction（驱逐）：禁止驱逐数据（redis4.0中默认策略），会引发错误OOM（Out Of Memory）

使用INFO命令输出监控信息，查询缓存 hit 和 miss 的次数，根据业务需求调优Redis配置

9、高级数据类型

9.1 Bitmaps

9.1.1 基本介绍：

用户一年的签到记录，如果你用 string 类型来存储，那你需要 365 个 key/value，操作起来麻烦。通过位图可以有效的简化这个操作。

它的统计很简单：01111000111

每天的记录占一个位，365 天就是 365 个位，大概 46 个字节，这样可以有效的节省存储空间，如果有一天想要统计用户一共签到了多少天，统计 1 的个数即可。

对于位图的操作，可以直接操作对应的字符串（get/set），可以直接操作位（getbit/setbit）.

在 Redis 中，字符串都是以二进制的方式来存储的。例如 set k1 a，a 对应的 ASCII 码是 97，97 转为二进制是 01100001，BIT 相关的命令就是对二进制进行操作的。

零存整取
存储的方式以二进制形式，取出的方式以字符串
例如：存储字符串Java
字符 ASCII 十进制
J 74 01001010
a 97 01100001
v 118 01110110
在redis中使用set [key] [offset] [value]的形式设置当前字母的bit位，设置1的位置即可，0无需设置
修改 key 对应的 value 在 offset(偏移) 处的 bit 值
设置J:
- set name 1 1
- set name 4 1
- set name 6 1
设置a:
- set name 9 1
- set name 10 1
- set name 15 1
设置v:
- set name 17 1
- set name 18 1
- set name 19 1
- set name 21 1
- set name 22 1
设置a:
- set name 25 1
- set name 26 1
- set name 31 1
取出字符串:
```
get name  #==> "Java"
```
整存零取
存一个字符串进去，但是通过位操作获取字符串
设置值：
- set k1 louchen.top
获取二进制位：
key 对应的 value 在 offset 处的 bit 值。
getbit [key] [offset]
```
getbit k1 0
getbit k1 1
......
getbit k1 87
```

字符	ASCII	十进制
J	74	01001010
a	97	01100001
v	118	01110110

9.1.2 命令介绍：

getbit
key 对应的 value 在 offset 处的 bit 值。
getbit [key] [offset]
setbit
修改 key 对应的 value 在 offset 处的 bit 值
setbit [key] [offset] [value]
bitcount
统计二进制数据中 1 的个数。
这里的start和end表示字符的起始（包含start,end）
bitcount [key] [start] [end]
bitpos
统计在0或1第一次出现的位置
bit ：0或1
start/end: 字符出现的位置(包含start，end)
bitpos [key] [bit] [start] [end]
bitop
对指定key按位进行交、并、非、异或操作，并将结果保存到destKey中
bitop op destKey key1 [key2...]
- and：交
- or：并
- not：非
- xor：异或
bitfield
在 Redis 3.2 之后，新加了一个功能叫做 bitfiled ，可以对 bit 进行批量操作。
1️⃣ 单个获取
BITFIELD name get u4 0
表示获取 name 中的位，从 0 开始获取，获取 4 个位，返回一个无符号数字。
- u 表示无符号数字
- i 表示有符号数字，有符号的话，第一个符号就表示符号位，1 表示是一个负数。
BITFIELD name get u4 0
- 即 0100 ==> 6
BITFIELD name get i4 0
- 即0100 ==> 6
bitfield name get u4 1
- 即 1001==>9
bitfield name get i4 1
- 即 1001 ，首位为1那么这个二进制位负数，则 -8+1=-7
2️⃣ 批量获取：
bitfield name get u4 1 get i4 1 get u4 0 get i4 0
3️⃣ incrby
对指定范围进行自增操作，自增操作可能会出现溢出，既可能是向上溢出，也可能是向下溢出。Redis 中对于溢出的处理方案是折返。
8位无符号数 255 加 1 溢出变为 0；8 位有符号数 127，加 1 变为 - 128.
默认的的溢出策略：
在Jbva基础上操作
- bitfield name incrby u2 6 1
  - 设置第6位开始，将2个bit位后面加1 ，即10 加1，得 11 ==> 01001011
  - get name ==> "Kbva"
  bitfield name incrby u2 6 1
  - 还是从第六位开始，在两个bit位后面加1，即 11 加 1，溢出。则将这两位全部置为0 ==> 01001000
  - get name ==> "Hbva"
修改默认的溢出策略：
在我们对Hbva基础上操作
H: 010010 00
- - fail策略：
    - bitfield name overflow fail incrby u2 6 1
      执行上述命令三次
      get name ==> "Kbva" 此时K: 010010 11
    - 再次执行 bitfield name overflow fail incrby u2 6 1
      (nil) 操作失败！此时的值还是为 “Kbva”
  - sat策略:
    - bitfield name overflow fail incrby u2 6 1
      无论执行多少次，值还是不会变。保持原来的最大或最小值
      (integer) 3
setfield
bitfield name set u8 8 98
表示从第8位开始获取8个无符号二进制位大小的数替换为98
此时的Java==>Jbva

9.2 HyperLogLog

特点
高级不精确去重的数据结构.(一般是超过一百个就开始不准确了) 占用空间小(一个键最多12k,可以计算2^64个元素) 没有contains操作因此有些他就支持不了,拓展到布隆过滤器
适用于一个热点页面的去重访问次数. 不适合单个用户的数据统计
背景
一般我们评估一个网站的访问量，有几个主要的参数：
- pv，Page View，网页的浏览量
- uv，User View，访问的用户
一般来说，pv 或者 uv 的统计，可以自己来做，也可以借助一些第三方的工具，比如 cnzz，友盟等。
如果自己实现，pv 比较简单，可以直接通过 Redis 计数器就能实现。但是 uv 就不一样，uv 涉及到另外一个问题，去重。
我们首先需要在前端给每一个用户生成一个唯一 id，无论是登录用户还是未登录用户，都要有一个唯一 id，这个 id 伴随着请求一起到达后端，在后端我们通过 set 集合中的 sadd 命令来存储这个 id，最后通过 scard 统计集合大小，进而得出 uv 数据。
如果是千万级别的 UV，需要的存储空间就非常惊人。而且，像 UV 统计这种，一般也不需要特别精确，800w 的 uv 和 803w 的 uv，其实差别不大。所以，我们要介绍今天的主角---HyperLogLog
Redis 中提供的 HyperLogLog 就是专门用来解决这个问题的，HyperLogLog 提供了一套不怎么精确但是够用的去重方案，会有误差，官方给出的误差数据是 0.81%，这个精确度，统计 UV 够用了。

😅命令介绍

pfadd
用来添加记录，类似于 sadd ，添加过程中，重复的记录会自动去重。
element：用户的标识
pfadd [key] [element...]
pfcount
统计一个或多个key的值的个数。取并集（去重），这种操作不会改变原key的大小
pfcount [key] [key...]
```
pfadd k1 u1 u2 u3 u4
pfcount k1  #==> 4
pfadd k2 u4 u5
pfcount k2  #==> 2
pfcount k1 k2 #===> 5
```
pfmerge
合并多个key到某一个key中，会去重。会改变目标key的大小
destkey :目标key
sourcekey 源key
pfmerge destkey sourcekey [sourcekey...]
```
pfcount uv #==> 994
pfadd u1 aa bb #==>添加不在uv的两个值 aa bb
pfmerge u1 uv  #==>合并uv到u1中
pfcount u1 #==> 996
```

测试误差

通过java代码来模拟大数据量的用户访问

public class Redis {

 private JedisPool jedisPool;

 public Redis() {
//        连接池配置对象
     GenericObjectPoolConfig config=new GenericObjectPoolConfig();
//        连接池最大空闲数
     config.setMaxIdle(300);
//        最大连接数
     config.setMaxTotal(1000);
//        连接最大等待时间（毫秒）。-1表示无限制
     config.setMaxWaitMillis(30000);
//        在空闲时检查有效性
     config.setTestOnBorrow(true);
     /**
         * 1、连接池配置
         * 2、redis地址
         * 3、端口
         * 4、超时时间
         * 5、密码
         */
        jedisPool = new JedisPool(config, "主机",6379,30000,"密码");
    }

    public void execute(CallWithJedis callWithJedis) {
        try (Jedis jedis=jedisPool.getResource()) {
            callWithJedis.call(jedis);
        }
    }
}

public interface CallWithJedis {
    void call(Jedis jedis);
}

public class HyperLogLogTest {
    public static void main(String[] args) {
        Redis redis = new Redis();
        redis.execute(jedis -> {
            for (int i = 0; i < 1000; i++) {
//                理论上我们累加的为1001次
                jedis.pfadd("uv", "u"+i,"u"+(i+1));
            }
            long uv = jedis.pfcount("uv");
            //实际输出 994
            System.out.println(uv);
        });
    }
}

🌖理论值是 1001，实际打印出来 994，有误差，但是在可以接受的范围内。

9.3 GEO

Redis3.2 开始提供了 GEO 模块。该模块也使用了 GeoHash 算法。

核心思想：GeoHash 是一种地址编码方法，使用这种方式，能够将二维的空间经纬度数据编码成一个一维字符串。

以经过伦敦格林尼治天文台旧址的经线为 0 度经线，向东就是东经，向西就是西经。如果我们将西经定义负，经度的范围就是 [-180,180]。纬度北纬 90 度到南纬 90 度，如果我们将南纬定义负，则纬度的范围就是 [-90,90]。

接下来，以本初子午线和赤道为界，我们可以将地球上的点分配到一个二维坐标中：

GeoHash 算法就是基于这样的思想，划分的次数越多，区域越多，每个区域中的面积就更小了，精确度就会提高。 GeoHash 具体算法：以北京天安门广场为例（39.9053908600,116.3980007200）：

①.纬度的范围在 (-90,90) 之间，中间值为 0，对于 39.9053908600 值落在 (0,90),因此得到的值为 1
②.(0,90) 的中间值为 45，39.9053908600 落在 (0,45) 之间，因此得到一个 0
③.(0,45) 的中间值为 22.5，39.9053908600 落在 (22.5,45)之间，因此得到一个 1
④. .... 这样，我们得到的纬度二进制是 101 按照同样的步骤，我们可以算出来经度的二进制是 110

接下来将经纬度合并（经度占偶数位，纬度占奇数位）： 111001

按照 Base32 （0-9,b-z,去掉 a i l 0）对合并后的二进制数据进行编码，编码的时候，先将二进制转换为十进制，然后进行编码。

将编码得到的字符串，可以拿去 geohash.orgopen in new window 网站上解析。

GeoHash 特点：

1、用一个字符串表示经纬度

2、 GeoHash 表示的是一个区域，而不是一个点。

3、编码格式有规律，例如一个地址编码之后的格式是 123，另一个地址编码之后的格式是 123456，从字符串上就可以看出来，123456 处于 123 之中。

**经纬度查询：**http://www.gpsspg.com/maps.htmopen in new window

命令介绍

添加地址

longitude:经度

latitude:纬度

geoadd [key] [longitude] [latitude] [member]

geoadd city 121.4737000000 31.2303700000 shanghai ##添加上海的位置
geoadd city 114.3052500000 30.5927600000 wuhan ##添加武汉位置

查看两个地址之间的距离
unit: 默认为 m(米)，可选为 m , km , ft (英尺)，mi(英里)
geodist [key] [member1] [member2] [unit]
```
geoadd city shanghai wuhan km 

"687.6116"
```

获取元素的位置

geopos [key] [member...]

geopos city shanghai

1) 1) "121.4736977219581604"
   2) "31.23036910904709629"

获取元素hash值
geohash [key] [member...]
```
geohash city wuhan

1) "wt3q114x9r0"
```
通过解析的哈希地址，我们可以查看其定位
http://geohash.org/wt3q114x9r0open in new window
30.59276 114.30525
查看附近的地址(通过成员)
key:指定key
member:以该地点为中心
radius:距离该中心的半径距离
unit:距离单位
withcoord:经纬度
withhash：hash值
withdist:半径距离
count:显示的数量
asc/desc:升序或者降序
georadiusbymeber [key] [member] [radius] [unit] [withcoord] [withhash] [withdist] [Count count] [asc|desc]
查询以武汉为中心，距离武汉3000km的3个地点（包括武汉本身），并且显示经纬度，hash值，半径距离，结果以距离的降序排列
GEORADIUSBYMEMBER city wuhan 3000 km withcoord withhash withdist count 3 desc
```
1) 1) "beijing"
   2) "1049.6601"
   3) (integer) 4069152897912916
   4) 1) "116.49902611970901489"
      2) "39.85915559490933191"
2) 1) "guangzhou"
   2) "858.8102"
   3) (integer) 4046518432493551
   4) 1) "113.35693091154098511"
      2) "22.91792342803383775"
3) 1) "jinan"
   2) "703.7194"
   3) (integer) 4065887515903236
   4) 1) "117.15820580720901489"
      2) "36.45663716057446635"
```
查看附近的地址(通过经纬度)
GEORADIUSBYMEMBER city wuhan 3000 km withcoord withhash withdist count 3 desc
把上面的wuhan元素替换成经经纬度即可：
GEORADIUS city 114.3052500000 30.5927600000 3000 km withcoord withhash withdist count 3 desc

10、主从复制

10.1 主从复制简介

互联网“三高”架构：

高并发
高性能
高可用

你的“Redis”是否高可用

单机redis的风险与问题：

问题1.机器故障
- 现象：硬盘故障、系统崩溃
- 本质：数据丢失，很可能对业务造成灾难性打击
- 结论：基本上会放弃使用redis.
问题2.容量瓶颈
- 现象：内存不足，从16G升级到64G，从64G升级到128G，无限升级内存
- 本质：穷，硬件条件跟不上
- 结论：放弃使用redis
结论：为了避免单点Redis服务器故障，准备多台服务器，互相连通。将数据复制多个副本保存在不同的服务器上，连接在一起，并保证数据是同步的。即使有其中一台服务器宕机，其他服务器依然可以继续提供服务，实现Redis的高可用，同时实现数据冗余备份。

多台服务器连接方案：

提供数据方：master
- 主服务器，主节点，主库
- 主客户端
接收数据方：slave
- 从服务器，从节点，从库
- 从客户端
需要解决的问题：
- 数据同步
核心工作：
- master的数据复制到slave中

主从复制:

主从复制即将master中的数据即时、有效的复制到slave中

**特征：**一个master可以拥有多个slave，一个slave只对应一个master
职责：
- master:
  - 写数据
  - 执行写操作时，将出现变化的数据自动同步到slave
  - 读数据（可忽略）
- slave:
  - 读数据
  - 写数据（禁止）

主从复制的作用：

读写分离：master写、slave读，提高服务器的读写负载能力
负载均衡：基于主从结构，配合读写分离，由slave分担master负载，并根据需求的变化，改变slave的数量，通过多个从节点分担数据读取负载，大大提高Redis服务器并发量与数据吞吐量
故障恢复：当master出现问题时，由slave提供服务，实现快速的故障恢复
数据冗余：实现数据热备份，是持久化之外的一种数据冗余方式
高可用基石：基于主从复制，构建哨兵模式与集群，实现Redis的高可用方案

高可用集群：

10.2 主从复制工作流程

总述：

主从复制过程大体可以分为3个阶段
- 建立连接阶段（即准备阶段）
- 数据同步阶段
- 命令传播阶段

10.2.1 阶段一：建立连接阶段

建立slave到master的连接，使master能够识别slave，并保存slave端口号

建立连接阶段工作流程：

主从连接（slave连接master）：

方式一：客户端发送命令
- slaveof <masterip> <masterport>
方式二：启动服务器参数
- redis-server --slaveof <masterip> <masterport>
方式三：服务器配置
- slaveof <masterip> <masterport>

主从断开连接:

客户端发送命令
- slaveof no one
  slave断开连接后，不会删除已有数据，只是不再接受master发送的数据

授权访问：：

案例：

准备工作：
这里选择使用两个配置文件来模拟两个不同的机器,分别代表主机master和从机slave
这里为了方便观察，主机和从机都以控制台形式启动
master redis-6379.conf
注释掉日志文件，改为前台启动，修改rdb,aof等文件名,访问密码等避免冲突
slave reids-6380.conf
注释掉日志文件，改为前台启动，修改rdb,aof等文件名,访问密码等避免冲突

其中上门四个标签分别代表 主机服务器，从机服务器，连接主机的客户端，连接从机的客户端

第一种方式:

从机slave客户端：

1️⃣ 从机连接客户端：

redis-cli -a 123456 -p 6380

2️⃣ 从机连接主机：
slaveof 127.0.0.1 6379
- 若主机设置密码，则需要在从机的配置文件中添加主机的密码：
  - masterauth 123456
若连接成功，则可以看到在主机控制台看到连接成功的信息：
3️⃣ 测试值
- 进入主机客户端设置值
  - 进入主机客户端
    - reids-cli -a 123456
      - 默认端口 6379
    - set k1 1
- 从机客户端取值 get k1
- 若从机取值成功，即可代表数据同步成功！

第二种方式：

从机启动服务时直接连接主机
- redis-server redis-6380.conf --slaveof 127.0.0.1 6379

第三种方式(推荐)：

通过配置从机的配置文件来连接主机
- 在redis-6380.conf中添加
  - slaveof 127.0.0.1 6379

10.2.2 阶段二：数据同步阶段工作流程

在slave初次连接master后，复制master中的所有数据到slave
将slave的数据库状态更新成master当前的数据库状态

数据同步阶段master说明：

如果master数据量巨大，数据同步阶段应避开流量高峰期，避免造成master阻塞，影响业务正常执行
复制缓冲区大小设定不合理，会导致数据溢出。如进行全量复制周期太长，进行部分复制时发现数据已经存在丢失的情况，必须进行第二次全量复制，致使slave陷入死循环状态。
repl-backlog-size 1mb 默认为1mb
master单机内存占用主机内存的比例不应过大，建议使用50%-70%的内存，留下30%-50%的内存用于执行bgsave命令和创建复制缓冲区

数据同步阶段slave说明:

为避免slave进行全量复制、部分复制时服务器响应阻塞或数据不同步，建议关闭此期间的对外服务
slave-serve-stale-data yes|no
数据同步阶段，master发送给slave信息可以理解master是slave的一个客户端，主动向slave发送命令
多个slave同时对master请求数据同步，master发送的RDB文件增多，会对带宽造成巨大冲击，如果master带宽不足，因此数据同步需要根据业务需求，适量错峰
slave过多时，建议调整拓扑结构，由一主多从结构变为树状结构，中间的节点既是master，也是slave。注意使用树状结构时，由于层级深度，导致深度越高的slave与最顶层master间数据同步延迟较大，数据一致性变差，应谨慎选择

10.2.3 阶段三：命令传播阶段

当master数据库状态被修改后，导致主从服务器数据库状态不一致，此时需要让主从数据同步到一致的状态，同步的动作称为命令传播
master将接收到的数据变更命令发送给slave，slave接收命令后执行命令

命令传播阶段的部分复制：

命令传播阶段出现了断网现象
- 网络闪断闪连 忽略
- 短时间网络中断 部分复制
- 长时间网络中断 全量复制
部分复制的三个核心要素
- 服务器的运行 id（run id）
- 主服务器的复制积压缓冲区
- 主从服务器的复制偏移量

1️⃣服务器运行ID（runid）

概念：服务器运行ID是每一台服务器每次运行的身份识别码，一台服务器多次运行可以生成多个运行id
组成：运行id由40位字符组成，是一个随机的十六进制字符例如：fdc9ff13b9bbaab28db42b3d50f852bb5e3fcdce
作用：运行id被用于在服务器间进行传输，识别身份如果想两次操作均对同一台服务器进行，必须每次操作携带对应的运行id，用于对方识别
实现方式：运行id在每台服务器启动时自动生成的，master在首次连接slave时，会将自己的运行ID发送给slave，slave保存此ID，通过info Server命令，可以查看节点的runid

2️⃣复制缓冲区

概念：复制缓冲区，又名复制积压缓冲区，是一个先进先出（FIFO）的队列，用于存储服务器执行过的命令，每次传播命令，master都会将传播的命令记录下来，并存储在复制缓冲区
- 复制缓冲区默认数据存储空间大小是1M，由于存储空间大小是固定的，当入队元素的数量大于队列长度时，最先入队的元素会被弹出，而新元素会被放入队列
由来：每台服务器启动时，如果开启有AOF或被连接成为master节点，即创建复制缓冲区
作用：用于保存master收到的所有指令（仅影响数据变更的指令，例如set，select）
数据来源：当master接收到主客户端的指令时，除了将指令执行，会将该指令存储到缓冲区中

3️⃣ 主从服务器复制偏移量（offset）

概念：一个数字，描述复制缓冲区中的指令字节位置
分类：
- master复制偏移量：记录发送给所有slave的指令字节对应的位置（多个）
- slave复制偏移量：记录slave接收master发送过来的指令字节对应的位置（一个）
数据来源：
- master端：发送一次记录一次
- slave端：接收一次记录一次
作用：同步信息，比对master与slave的差异，当slave断线后，恢复数据使用

心跳机制：

进入命令传播阶段候，master与slave间需要进行信息交换，使用心跳机制进行维护，实现双方连接保持在线
master心跳：
- 指令：PING
- 周期：由repl-ping-slave-period决定，默认10秒
- 作用：判断slave是否在线
- 查询：INFO replication 获取slave最后一次连接时间间隔，lag项维持在0或1视为正常、
slave心跳任务
- 指令：REPLCONF ACK
- 周期：1秒
- 作用1：汇报slave自己的复制偏移量，获取最新的数据变更指令
- 作用2：判断master是否在线
心跳阶段注意事项
- 当slave多数掉线，或延迟过高时，master为保障数据稳定性，将拒绝所有信息同步操作
  min-slaves-to-write 2 min-slaves-max-lag 8
  slave数量少于2个，或者所有slave的延迟都大于等于10秒时，强制关闭master写功能，停止数据同步
- slave数量由slave发送REPLCONF ACK命令做确认
- slave延迟由slave发送REPLCONF ACK命令做确认

10.2.4 主从复制全部工作流程

10.3 主从复制常见问题

10.3.1 频繁的全量复制（1）

伴随着系统的运行，master的数据量会越来越大，一旦master重启，runid将发生变化，会导致全部slave的全量复制操作

内部优化调整方案：

master内部创建master_replid变量，使用runid相同的策略生成，长度41位，并发送给所有slave
在master关闭时执行命令 shutdown save，进行RDB持久化,将runid与offset保存到RDB文件中
- repl-id repl-offset
- 通过redis-check-rdb命令可以查看该信息
master重启后加载RDB文件，恢复数据
- 重启后，将RDB文件中保存的repl-id与repl-offset加载到内存中
  - master_repl_id = repl master_repl_offset = repl-offset
  - 通过info命令可以查看该信息

作用：本机保存上次runid，重启后恢复该值，使所有slave认为还是之前的master

10.3.2 频繁的全量复制（2）

问题现象
- 网络环境不佳，出现网络中断，slave不提供服务
问题原因
- 复制缓冲区过小，断网后slave的offset越界，触发全量复制
最终结果
- slave反复进行全量复制
解决方案
- 修改复制缓冲区大小
  repl-backlog-size
建议设置如下：
- 测算从master到slave的重连平均时长second
- 获取master平均每秒产生写命令数据总量write_size_per_second
- 最优复制缓冲区空间 = 2 * second * write_size_per_second

10.3.3 频繁的网络中断（1）

问题现象
- master的CPU占用过高或 slave频繁断开连接
问题原因
- slave每1秒发送REPLCONF ACK命令到master
- 当slave接到了慢查询时（keys * ，hgetall等），会大量占用CPU性能
- master每1秒调用复制定时函数replicationCron()，比对slave发现长时间没有进行响应
最终结果
- master各种资源（输出缓冲区、带宽、连接等）被严重占用
解决方案
- 通过设置合理的超时时间，确认是否释放slave
  repl-timeout
  该参数定义了超时时间的阈值（默认60秒），超过该值，释放slave

10.3.4 频繁的网络中断（2）

问题现象
- slave与master连接断开
问题原因
- master发送ping指令频度较低
- master设定超时时间较短
- ping指令在网络中存在丢包
解决方案
- 提高ping指令发送的频度
  repl-ping-slave-period
  超时时间repl-time的时间至少是ping指令频度的5到10倍，否则slave很容易判定超时

10.3.5 数据不一致

问题现象
- 多个slave获取相同数据不同步
问题原因
- 网络信息不同步，数据发送有延迟
解决方案
- 优化主从间的网络环境，通常放置在同一个机房部署，如使用阿里云等云服务器时要注意此现象
- 监控主从节点延迟（通过offset）判断，如果slave延迟过大，暂时屏蔽程序对该slave的数据访问
  slave-serve-stale-data yes|no
  开启后仅响应info、slaveof等少数命令（慎用，除非对数据一致性要求很高）

11、哨兵

11.1 哨兵简介

问题：主机"宕机":

关闭master和所有slave
找一个slave作为master
修改其他slave的配置，连接新的主
启动新的master与slave
全量复制N+部分复制N
关闭期间的数据服务谁来承接？
找一个主？怎么找法？
修改配置后，原始的主恢复了怎么办？

什么是哨兵？

哨兵(sentinel) 是一个分布式系统，用于对主从结构中的每台服务器进行监控，当出现故障时通过投票机制选择新的master并将所有slave连接到新的master。

哨兵的作用

监控
- 不断的检查master和slave是否正常运行。 master存活检测、master与slave运行情况检测
通知（提醒）
- 当被监控的服务器出现问题时，向其他（哨兵间，客户端）发送通知。
自动故障转移
- 断开master与slave连接，选取一个slave作为master，将其他slave连接到新的master，并告知客户端新的服务器地址

注意：
哨兵也是一台redis服务器，只是不提供数据服务
通常哨兵配置数量为单数

11.2 启用哨兵模式

准备工作：
配置一主二从的结构
创建三个配置不同的配置文件 redis-6379.conf、redis-6380.conf、reids-6381.conf, 端口分别对应6379,6380,6381，并修改配置文件中的其他必要信息
一主机服务器配置为：redis-6379.conf
二从服务器配置为:redis-6380.conf、reids-6381.conf
注意增加，主从连接的地址以及密码验证方式
masterauth 123456slaveof 127.0.0.1 6379
配置三个哨兵(配置相同，端口不同)
分别新建三个sentinel-26379.conf、sentinel-26380.conf、sentinel-26391.conf 文件，端口分别对应26379,26380,26381，并修改配置文件中的其他必要信息
哨兵配置文件详解 sentinel.conf
port 26379 #服务端口 daemonize no #设定守护进程启动(是否后台启动) pidfile /var/run/redis-sentinel.pid logfile "" dir /tmp #工作存储地址信息 sentinel monitor mymaster 127.0.0.1 6379 2 #【mymaster】为自定义监控主机的名称，【127.0.0.1 6379】为主机的ip和端口，【2】代表有2个哨兵认为该主机挂了，则代表改主机挂了，通过设置哨兵的数量的一半加1 sentinel down-after-milliseconds mymaster 30000 #哨兵认为连接主机的时间超过30s没响应则认为主机挂了，【mymaster】注意和上面一行名称一致 sentinel parallel-syncs mymaster 1 #发生故障时主备切换时最多可以有多少个slave同时对新的master进行同步 sentinel failover-timeout mymaster 180000 #同步超时时间3min sentinel deny-scripts-reconfig yes sentinel auth-pass mymaster 123456 #设置主机访问密码
- 若主机设置了**访问密码**，则哨兵也需要设置访问密码

`sentinel auth-pass mymaster 123456`
其它的一些配置信息
详细参考 https://blog.csdn.net/a1282379904/article/details/52335051open in new window
配置项范例说明
sentinel auth-pass <服务器名称> <password> sentinel auth-pass mymaster 123456 连接服务器口令
sentinel down-after-milliseconds <自定义服务名称><主机地址><端口><主从服务器总量> sentinel monitor mymaster 192.168.194.131 6381 1 设置哨兵监听的主服务器信息，最后的参数决定了最终参与选举的服务器数量（-1）
sentinel down-after-milliseconds <服务名称><毫秒数（整数）> sentinel down-after-milliseconds mymaster 3000 指定哨兵在监控Redis服务时，判定服务器挂掉的时间周期，默认30秒（30000），也是主从切换的启动条件之一
sentinel parallel-syncs <服务名称><服务器数（整数）> sentinel parallel-syncs mymaster 1 指定同时进行主从的slave数量，数值越大，要求网络资源越高，要求约小，同步时间约长
sentinel failover-timeout <服务名称><毫秒数（整数）> sentinel failover-timeout mymaster 9000 指定出现故障后，故障切换的最大超时时间，超过该值，认定切换失败，默认3分钟
sentinel notification-script <服务名称><脚本路径> 服务器无法正常联通时，设定的执行脚本，通常调试使用。

配置项	范例	说明
sentinel auth-pass <服务器名称> <password>	sentinel auth-pass mymaster 123456	连接服务器口令
sentinel down-after-milliseconds <自定义服务名称><主机地址><端口><主从服务器总量>	sentinel monitor mymaster 192.168.194.131 6381 1	设置哨兵监听的主服务器信息，最后的参数决定了最终参与选举的服务器数量（-1）
sentinel down-after-milliseconds <服务名称><毫秒数（整数）>	sentinel down-after-milliseconds mymaster 3000	指定哨兵在监控Redis服务时，判定服务器挂掉的时间周期，默认30秒（30000），也是主从切换的启动条件之一
sentinel parallel-syncs <服务名称><服务器数（整数）>	sentinel parallel-syncs mymaster 1	指定同时进行主从的slave数量，数值越大，要求网络资源越高，要求约小，同步时间约长
sentinel failover-timeout <服务名称><毫秒数（整数）>	sentinel failover-timeout mymaster 9000	指定出现故障后，故障切换的最大超时时间，超过该值，认定切换失败，默认3分钟
sentinel notification-script <服务名称><脚本路径>		服务器无法正常联通时，设定的执行脚本，通常调试使用。

启动步骤：

先启动主机，
- redis-server redis-6379.conf
再启动从机
- redis-server redis-6380.conf
- redis-server redis-6381.conf
接着启动三个哨兵
- redis-sentinel sentinel-26379.conf
- redis-sentinel sentinel-26380.conf
- redis-sentinel sentinel-26381.conf

关键启动信息：

哨兵服务器的信息：
该哨兵的sentinel-26379.conf配置文件中的信息
进入哨兵1客户端：
- redis-cli -p 26379
- 使用info命令
  - - **可以看到哨兵的一些连接状态信息，从机的数量，哨兵的数量等**

主机断开，哨兵在从机中重新选举出新的主机：
手动断开主机，30s后查看哨兵和从机的信息
从机1的信息：
哨兵1的信息
注意：
若下线的主机6379重新上线，那么哨兵会取消认定该主机的下线行为，但是该6379将不再担任主机，而是以从机的身份连接。

11.3 哨兵工作原理

主从切换：

哨兵在进行主从切换过程中经历三个阶段：

监控
通知
故障转移

11.3.1 阶段一：监控阶段

11.3.2 阶段二：通知阶段

11.3.3 阶段三：故障转移阶段

服务器列表中挑选备选master
- 在线的
- 响应慢的
- 与原master断开时间久的
- 优先原则
  - 优先级
  - offset
  - runid
发送指令（ sentinel ）
- 向新的master发送slaveof no one
- 向其他slave发送slaveof 新masterIP端口

11.3.4 总结

监控
- 同步信息
通知
- 保持联通
故障转移
- 发现问题
- 竞选负责人
- 优选新master
- 新master上任，其他slave切换master，原master作为slave故障回复后连接

12、集群cluster

12.1 集群简介

业务发展过程中遇到的峰值瓶颈 ❓
redis提供的服务OPS可以达到10万/秒，当前业务OPS已经达到10万/秒
内存单机容量达到256G，当前业务需求内存容量1T
使用集群的方式可以快速解决上述问题

集群架构

集群就是使用网络将若干台计算机联通起来，并提供统一的管理方式，使其对外呈现单机的服务效果

集群作用

分散单台服务器的访问压力，实现负载均衡
分散单台服务器的存储压力，实现可扩展性
降低单台服务器宕机带来的业务灾难

12.2 Redis集群结构设计

数据存储设计

通过算法设计，计算出key应该保存的位置
将所有的存储空间计划切割成16384份，每台主机保存一部分每份代表的是一个存储空间，不是一个key的保存空间
将key按照计算出的结果放到对应的存储空间
客户端与Redis节点直连,不需要中间Proxy层，直接连接任意一个Master节点
根据公式HASH_SLOT=CRC16(key) mod 16384，计算出映射到哪个分片上，然后Redis会去相应的节点进行操作

增强可扩展性

集群内部通讯设计

各个数据库相互通信，保存各个库中槽的编号数据
一次命中，直接返回
一次未命中，告知具体位置

12.3 cluster集群结构搭建

为了配置一个redis cluster,我们需要准备至少6台redis，为啥至少6台呢？我们可以在redis的官方文档中找到如下一句话：

Note that the minimal cluster that works as expected requires to contain at least three master nodes.

因为最小的redis集群，需要至少3个主节点，既然有3个主节点，而一个主节点搭配至少一个从节点，因此至少得6台redis。然而对我来说，就是复制6个redis配置文件。本实验的redis集群搭建依然在一台电脑上模拟。

准备工作

准备6个配置文件，分别为（⚠️ 这里不需要配置主从）
redis-6379.conf 主1 redis-6382.conf 从1
redis-6380.conf 主2 redis-6383.conf 从2
redis-6381.conf 主3 redis-6384.conf 从3
配置文件大致和前面内容相同，只是需要在各配置文件添加如下内容：
cluster-enabled yes #标记为集群
cluster-config-file nodes-6379.conf  #集群节点生成配置文件
cluster-node-timeout 10000  #下线超时时间为10s
cluster配置详解
添加节点
cluster-enabled yes|no
cluster配置文件名，该文件属于自动生成，仅用于快速查找文件并查询文件内容
cluster-config-file <filename>
节点服务响应超时时间，用于判定该节点是否下线或切换为从节点
cluster-node-timeout <milliseconds>
master连接的slave最小数量
cluster-migration-barrier <count>

启动流程

1️⃣先启动主机，再启动从机
- redis-server redis-6379.conf
- redis-server redis-6380.conf
- redis-server redis-6381.conf
- redis-server redis-6382.conf
- redis-server redis-6383.conf
- redis-server redis-6384.conf

2️⃣创建集群

Redis Cluster 在5.0之后取消了ruby脚本 redis-trib.rb的支持（手动命令行添加集群的方式不变），集合到redis-cli里，避免了再安装ruby的相关环境。直接使用redis-clit的参数--cluster 来取代。为方便自己后面查询就说明下如何使用该命令进行Cluster的创建和管理，关于Cluster的相关说明可以查看官网open in new window或则Redis Cluster部署、管理和测试open in new window。

参考：https://www.cnblogs.com/zhoujinyi/p/11606935.htmlopen in new window

查看集群帮助命令

redis-cli --cluster help

redis-cli --cluster help
Cluster Manager Commands:
  create         host1:port1 ... hostN:portN   #创建集群
                 --cluster-replicas <arg>      #从节点个数
  check          host:port                     #检查集群
                 --cluster-search-multiple-owners #检查是否有槽同时被分配给了多个节点
  info           host:port                     #查看集群状态
  fix            host:port                     #修复集群
                 --cluster-search-multiple-owners #修复槽的重复分配问题
  reshard        host:port                     #指定集群的任意一节点进行迁移slot，重新分slots
                 --cluster-from <arg>          #需要从哪些源节点上迁移slot，可从多个源节点完成迁移，以逗号隔开，传递的是节点的node id，还可以直接传递--from all，这样源节点就是集群的所有节点，不传递该参数的话，则会在迁移过程中提示用户输入
                 --cluster-to <arg>            #slot需要迁移的目的节点的node id，目的节点只能填写一个，不传递该参数的话，则会在迁移过程中提示用户输入
                 --cluster-slots <arg>         #需要迁移的slot数量，不传递该参数的话，则会在迁移过程中提示用户输入。
                 --cluster-yes                 #指定迁移时的确认输入
                 --cluster-timeout <arg>       #设置migrate命令的超时时间
                 --cluster-pipeline <arg>      #定义cluster getkeysinslot命令一次取出的key数量，不传的话使用默认值为10
                 --cluster-replace             #是否直接replace到目标节点
  rebalance      host:port                                      #指定集群的任意一节点进行平衡集群节点slot数量 
                 --cluster-weight <node1=w1...nodeN=wN>         #指定集群节点的权重
                 --cluster-use-empty-masters                    #设置可以让没有分配slot的主节点参与，默认不允许
                 --cluster-timeout <arg>                        #设置migrate命令的超时时间
                 --cluster-simulate                             #模拟rebalance操作，不会真正执行迁移操作
                 --cluster-pipeline <arg>                       #定义cluster getkeysinslot命令一次取出的key数量，默认值为10
                 --cluster-threshold <arg>                      #迁移的slot阈值超过threshold，执行rebalance操作
                 --cluster-replace                              #是否直接replace到目标节点
  add-node       new_host:new_port existing_host:existing_port  #添加节点，把新节点加入到指定的集群，默认添加主节点
                 --cluster-slave                                #新节点作为从节点，默认随机一个主节点
                 --cluster-master-id <arg>                      #给新节点指定主节点
  del-node       host:port node_id                              #删除给定的一个节点，成功后关闭该节点服务
  call           host:port command arg arg .. arg               #在集群的所有节点执行相关命令
  set-timeout    host:port milliseconds                         #设置cluster-node-timeout
  import         host:port                                      #将外部redis数据导入集群
                 --cluster-from <arg>                           #将指定实例的数据导入到集群
                 --cluster-copy                                 #migrate时指定copy
                 --cluster-replace                              #migrate时指定replace
  help           

For check, fix, reshard, del-node, set-timeout you can specify the host and port of any working node in the cluster.

⚠️注意：Redis Cluster最低要求是3个主节点，如果需要集群需要认证，则在最后加入 -a xx 即可。

创建集群主从节点

redis-cli --cluster create 127.0.0.1:6379 127.0.0.1:6380 127.0.0.1:6381 127.0.0.1:6382 127.0.0.1:6383 127.0.0.1:6384 --cluster-replicas 1

说明：--cluster-replicas 参数为数字，1表示每个主节点需要1个从节点。并且前面的地址都为mater，剩余的为slave
若要配置 1主2从，那么 --cluster-replicas 2 ，配置的地址为9个，前三个为主机，后六个为从机

创建后的日志信息：

[root@localhost src]# redis-cli --cluster create 127.0.0.1:6379 127.0.0.1:6380 127.0.0.1:6381 127.0.0.1:6382 127.0.0.1:6383 127.0.0.1:6384 --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:6383 to 127.0.0.1:6379
Adding replica 127.0.0.1:6384 to 127.0.0.1:6380
Adding replica 127.0.0.1:6382 to 127.0.0.1:6381
>>> Trying to optimize slaves allocation for anti-affinity
[WARNING] Some slaves are in the same host as their master
M: 7698aeb50b47fed20b0c254b644f1b25ebb8a49b 127.0.0.1:6379
   slots:[0-5460] (5461 slots) master
M: 6a05218cd747f8609ef1a8ab7a56eb769d140240 127.0.0.1:6380
   slots:[5461-10922] (5462 slots) master
M: daf3b434b5ad195b10c4e931c1767eb3b5da17a7 127.0.0.1:6381
   slots:[10923-16383] (5461 slots) master
S: 84bc4b534065cf33f13c92457ac39e8a92562810 127.0.0.1:6382
   replicates daf3b434b5ad195b10c4e931c1767eb3b5da17a7
S: 64ffbe55e885506a1cdccad8b077a0539811cc49 127.0.0.1:6383
   replicates 7698aeb50b47fed20b0c254b644f1b25ebb8a49b
S: b05f6345b9141f2adc7f1be59fc13b1cb54c62fb 127.0.0.1:6384
   replicates 6a05218cd747f8609ef1a8ab7a56eb769d140240
Can I set the above configuration? (type 'yes' to accept): yes
>>> Nodes configuration updated
>>> Assign a different config epoch to each node
>>> Sending CLUSTER MEET messages to join the cluster
Waiting for the cluster to join
....
>>> Performing Cluster Check (using node 127.0.0.1:6379)
M: 7698aeb50b47fed20b0c254b644f1b25ebb8a49b 127.0.0.1:6379
   slots:[0-5460] (5461 slots) master
   1 additional replica(s)
S: b05f6345b9141f2adc7f1be59fc13b1cb54c62fb 127.0.0.1:6384
   slots: (0 slots) slave
   replicates 6a05218cd747f8609ef1a8ab7a56eb769d140240
S: 64ffbe55e885506a1cdccad8b077a0539811cc49 127.0.0.1:6383
   slots: (0 slots) slave
   replicates 7698aeb50b47fed20b0c254b644f1b25ebb8a49b
M: 6a05218cd747f8609ef1a8ab7a56eb769d140240 127.0.0.1:6380
   slots:[5461-10922] (5462 slots) master
   1 additional replica(s)
M: daf3b434b5ad195b10c4e931c1767eb3b5da17a7 127.0.0.1:6381
   slots:[10923-16383] (5461 slots) master
   1 additional replica(s)
S: 84bc4b534065cf33f13c92457ac39e8a92562810 127.0.0.1:6382
   slots: (0 slots) slave
   replicates daf3b434b5ad195b10c4e931c1767eb3b5da17a7
[OK] All nodes agree about slots configuration.
>>> Check for open slots...
>>> Check slots coverage...
[OK] All 16384 slots covered.

设置与获取数据

redis-cli -c 进入集群客户端
redis-cli -c -p 6381 进入指定的集群端口客户端

[root@localhost conf]# redis-cli
127.0.0.1:6379> set name hhaha   --在6379主节点上取不到值
(error) MOVED 5798 127.0.0.1:6380
127.0.0.1:6379> 
[root@localhost conf]# redis-cli -c  
127.0.0.1:6379> set name haha
-> Redirected to slot [5798] located at 127.0.0.1:6380  --直接重定向到在6380主节点的[5798]槽中
OK
127.0.0.1:6380> 
[root@localhost conf]# redis-cli -c -p 6381  
127.0.0.1:6381> get name
-> Redirected to slot [5798] located at 127.0.0.1:6380  --直接在6380主节点上取值，取值成功
"haha"
127.0.0.1:6380> 
[root@localhost conf]# redis-cli -c -p 6380
127.0.0.1:6380> get name
"haha"

主从下线与主从切换

以6379主节点对应的6382从节点为例
当6382从节点上下线对整体功能暂无影响
当6379主节点下线，则从节点6382从节点重试连接失败后会自动转变为主节点。而当6379节点再次上线时，此时6379变为6382的从节点

Cluster节点操作命令

查看集群节点信息

cluster nodes

进入一个从节点 redis，切换其主节点

cluster replicate <master-id>

发现一个新节点，新增主节点

cluster meet ip:port

忽略一个没有solt的节点

cluster forget <id>

手动故障转移

cluster failover

13、场景问题

13.1 缓存预热

❓宕机的出现

服务器启动后迅速宕机

问题排查

请求数量较高
主从之间数据吞吐量较大，数据同步操作频度较高

解决方案

前置准备工作
- 日常例行统计数据访问记录，统计访问频度较高的热点数据
- 利用LRU数据删除策略，构建数据留存队列例如：storm与kafka配合
准备工作
- 将统计结果中的数据分类，根据级别，redis优先加载级别较高的热点数据
- 利用分布式多服务器同时进行数据读取，提速数据加载过程
- 热点数据主从同时预热
实施
- 使用脚本程序固定触发数据预热过程
- 如果条件允许，使用了CDN（内容分发网络），效果会更好

总结

缓存预热就是系统启动前，提前将相关的缓存数据直接加载到缓存系统。避免在用户请求的时候，先查询数据库，然后再将数据缓存的问题！用户直接查询事先被预热的缓存数据！

13.2 缓存雪崩

数据库服务器崩溃

1.系统平稳运行过程中，忽然数据库连接量激增 2.应用服务器无法及时处理请求 3.大量408，500错误页面出现 4.客户反复刷新页面获取数据 5.数据库崩溃 6.应用服务器崩溃 7.重启应用服务器无效 8.Redis服务器崩溃 9.Redis集群崩溃 10.重启数据库后再次被瞬间流量放倒

问题排查

1.在一个较短的时间内，缓存中较多的key集中过期 2.此周期内请求访问过期的数据，redis未命中，redis向数据库获取数据 3.数据库同时接收到大量的请求无法及时处理 4.Redis大量请求被积压，开始出现超时现象 5.数据库流量激增，数据库崩溃 6.重启后仍然面对缓存中无数据可用 7.Redis服务器资源被严重占用，Redis服务器崩溃 8.Redis集群呈现崩塌，集群瓦解 9.应用服务器无法及时得到数据响应请求，来自客户端的请求数量越来越多，应用服务器崩溃 10.应用服务器，redis，数据库全部重启，效果不理想

问题分析

短时间范围内大量key集中过期

解决方案（设计层面）

更多的页面静态化处理
构建多级缓存架构
- Nginx缓存+redis缓存+ehcache缓存
检测Mysql严重耗时业务进行优化
- 对数据库的瓶颈排查：例如超时查询、耗时较高事务等
灾难预警机制
- 监控redis服务器性能指标
  - CPU占用、CPU使用率内存容量查询平均响应时间线程数
限流、降级
- 短时间范围内牺牲一些客户体验，限制一部分请求访问，降低应用服务器压力，待业务低速运转后再逐步放开访问

解决方案（实现方案）

LRU与LFU切换
数据有效期策略调整
- 根据业务数据有效期进行分类错峰，A类90分钟，B类80分钟，C类70分钟
- 过期时间使用固定时间+随机值的形式，稀释集中到期的key的数量
超热数据使用永久key
定期维护（自动+人工）
- 对即将过期数据做访问量分析，确认是否延时，配合访问量统计，做热点数据的延时
加锁慎用！

总结

缓存雪崩就是瞬间过期数据量太大，导致对数据库服务器造成压力。如能够有效避免过期时间集中，可以有效解决雪崩现象的出现（约40%），配合其他策略一起使用，并监控服务器的运行数据，根据运行记录做快速调整。

13.3 缓存击穿

数据库服务器崩溃

1.系统平稳运行过程中 2.数据库连接量瞬间激增 3.Redis服务器无大量key过期 4.Redis内存平稳，无波动 5.Redis服务器CPU正常 6.数据库崩溃

问题排查

1.Redis中某个key过期，该key访问量巨大 2.多个数据请求从服务器直接压到Redis后，均未命中 3.Redis在短时间内发起了大量对数据库中同一数据的访问

问题分析

单个key高热数据 key过期

解决方案

1.预先设定
- 以电商为例，每个商家根据店铺等级，指定若干款主打商品，在购物节期间，加大此类信息key的过期时长
- 注意：购物节不仅仅指当天，以及后续若干天，访问峰值呈现逐渐降低的趋势
2.现场调整
- 监控访问量，对自然流量激增的数据延长过期时间或设置为永久性key
3.后台刷新数据
- 启动定时任务，高峰期来临之前，刷新数据有效期，确保不丢失
4.二级缓存
- 设置不同的失效时间，保障不会被同时淘汰就行
5.加锁
- 分布式锁，防止被击穿，但是要注意也是性能瓶颈，慎重！

总结

缓存击穿就是单个高热数据过期的瞬间，数据访问量较大，未命中redis后，发起了大量对同一数据的数据库访问，导致对数据库服务器造成压力。应对策略应该在业务数据分析与预防方面进行，配合运行监控测试与即时调整策略，毕竟单个key的过期监控难度较高，配合雪崩处理策略即可。

13.4 缓存穿透

数据库服务器崩溃

1.系统平稳运行过程中 2.应用服务器流量随时间增量较大 3.Redis服务器命中率随时间逐步降低 4.Redis内存平稳，内存无压力 5.Redis服务器CPU占用激增 6.数据库服务器压力激增 7.数据库崩溃

问题排查

1.Redis中大面积出现未命中 2.出现非正常URL访问

问题分析

获取的数据在数据库中也不存在，数据库查询未得到对应数据
Redis获取到null数据未进行持久化，直接返回
下次此类数据到达重复上述过程
出现黑客攻击服务器

解决方案

1.缓存null

对查询结果为null的数据进行缓存（长期使用，定期清理），设定短时限，例如30-60秒，最高5分钟

2.白名单策略

提前预热各种分类数据id对应的bitmaps，id作为bitmaps的offset，相当于设置了数据白名单。当加载正常数据时，放行，加载异常数据时直接拦截（效率偏低）
使用布隆过滤器（有关布隆过滤器的命中问题对当前状况可以忽略）

3.实施监控

实时监控redis命中率（业务正常范围时，通常会有一个波动值）与null数据的占比

非活动时段波动：通常检测3-5倍，超过5倍纳入重点排查对象
活动时段波动：通常检测10-50倍，超过50倍纳入重点排查对象
根据倍数不同，启动不同的排查流程。然后使用黑名单进行防控（运营）

4.key加密

问题出现后，临时启动防灾业务key，对key进行业务层传输加密服务，设定校验程序，过来的key校验
例如每天随机分配60个加密串，挑选2到3个，混淆到页面数据id中，发现访问key不满足规则，驳回数据访问

总结

缓存击穿访问了不存在的数据，跳过了合法数据的redis数据缓存阶段，每次访问数据库，导致对数据库服务器造成压力。通常此类数据的出现量是一个较低的值，当出现此类情况以毒攻毒，并及时报警。应对策略应该在临时预案防范方面多做文章。无论是黑名单还是白名单，都是对整体系统的压力，警报解除后尽快移除。

13.5 性能监控指标

监控方式

工具
- Cloud Insight Redis
- Prometheus
- Redis-stat
- Redis-faina
- RedisLive
- zabbix
命令
- benchmark
- redis cli
  - monitor
  - showlog

benchmark

命令
- redis-benchmark [-h ] [-p ] [-c ] [-n <requests]> [-k ]
范例1
- redis-benchmark
  说明：50个连接，10000次请求对应的性能
范例2
- redis-benchmark -c 100 -n 5000
  说明：100个连接，5000次请求对应的性能

monitor

命令
- monitor
  打印服务器调试信息

showlong

命令
- showlong [operator]
  get ：获取慢查询日志 len ：获取慢查询日志条目数 reset ：重置慢查询日志

Redis知识点总结

# 1、Redis简介

# 1.1 海量用户/高并发带来的问题和现象

# 1.2 什么是NoSQL

# 1.3 Redis简介

# 1.4 Redis的应用

# 2、Redis的下载与安装

# 2.1 在线体验

# 2.2 安装包编译安装

# 2.3 外网访问与密码配置

# 进入命令行

# 2.4 Docker安装Redis

# 2.5 redis.config配置文件

# 2.5.1 服务端配置

# 2.5.2 日志配置

# 2.5.3 客户端配置

# 2.5.4 多服务器快捷配置

# 3、Redis基本操作

# 3.1 通用命令

# 3.2 帮助命令

# 3.3 数据操作指令

# 4、五种基本数据类型操作

# Redis数据存储格式

# key的设置与约定

# 4.1 String

# 获取并更新某一个key

# 和incrby类似，步长可以设置为浮点数

# 4.2 List

# 4.3 Set

# 4.4 Hash

# 4.5 ZSet

# 5、使用Java连接Redis

# 5.1 Jedis连接

# 5.2 Jedis连接池的使用

# 5.3 Lettuce连接

# 6、数据持久化

# 6.1 RDB

# 6.1.1 save指令

# 6.1.2 bgsave指令

# 6.1.3 通过redis.conf配置save执行

# RDB三种启动方式对比

# RDB特殊启动方式

# RDB优缺点

# 6.2 AOF

# 6.2.1 AOF概念

# 6.2.2 AOF写数据的三种策略appendfsync

# 6.2.3 redis.conf中AOF相关配置

# 6.2.4 AOF重写

# 6.3 RDB和AOF的区别

# RDB和AOF的选择

# 7、Redis事务

# 7.1 基本概念

# 7.2 事务的基本操作

# Redis事务不保证原子性

# 7.3 事务的工作流程

# 7.4 事务监听watch

# 7.5 分布式锁

# 7.5.1 问题场景

# 7.5.2 实现方法

# 7.6.3 解决超时问题

# 7.6 Redisson实现分布式锁

# 8、删除策略

# 8.1 定时删除

# 8.2 惰性删除

# 8.3 定期删除

# 8.4 三种删除策略对比

# 8.5 逐出算法

# 9、高级数据类型

# 9.1 Bitmaps

# 9.1.1 基本介绍：

# 9.1.2 命令介绍：

# 9.2 HyperLogLog

# 9.3 GEO

# 查看附近的地址(通过经纬度)

# 10、主从复制

# 10.1 主从复制简介

# 10.2 主从复制工作流程

# 10.2.1 阶段一：建立连接阶段

# 10.2.2 阶段二：数据同步阶段工作流程

# 10.2.3 阶段三：命令传播阶段

1、Redis简介

1.1 海量用户/高并发带来的问题和现象

1.2 什么是NoSQL

1.3 Redis简介

1.4 Redis的应用

2、Redis的下载与安装

2.1 在线体验

2.2 安装包编译安装

2.3 外网访问与密码配置

进入命令行

2.4 Docker安装Redis

2.5 redis.config配置文件

2.5.1 服务端配置

2.5.2 日志配置

2.5.3 客户端配置

2.5.4 多服务器快捷配置

3、Redis基本操作

3.1 通用命令

3.2 帮助命令

3.3 数据操作指令

4、五种基本数据类型操作

Redis数据存储格式

key的设置与约定

4.1 String

获取并更新某一个key

和incrby类似，步长可以设置为浮点数

4.2 List

4.3 Set

4.4 Hash

4.5 ZSet

5、使用Java连接Redis

5.1 Jedis连接

5.2 Jedis连接池的使用

5.3 Lettuce连接

6、数据持久化

6.1 RDB

6.1.1 save指令

6.1.2 bgsave指令

6.1.3 通过redis.conf配置save执行

RDB三种启动方式对比

RDB特殊启动方式

RDB优缺点

6.2 AOF

6.2.1 AOF概念

6.2.2 AOF写数据的三种策略appendfsync

6.2.3 redis.conf中AOF相关配置

6.2.4 AOF重写

6.3 RDB和AOF的区别

RDB和AOF的选择

7、Redis事务

7.1 基本概念

7.2 事务的基本操作

Redis事务不保证原子性

7.3 事务的工作流程

7.4 事务监听watch

7.5 分布式锁

7.5.1 问题场景

7.5.2 实现方法

7.6.3 解决超时问题

7.6 Redisson实现分布式锁

8、删除策略

8.1 定时删除

8.2 惰性删除

8.3 定期删除

8.4 三种删除策略对比

8.5 逐出算法

9、高级数据类型

9.1 Bitmaps

9.1.1 基本介绍：

9.1.2 命令介绍：

9.2 HyperLogLog

9.3 GEO

查看附近的地址(通过经纬度)

10、主从复制

10.1 主从复制简介

10.2 主从复制工作流程

10.2.1 阶段一：建立连接阶段

10.2.2 阶段二：数据同步阶段工作流程

10.2.3 阶段三：命令传播阶段

10.2.4 主从复制全部工作流程