深入浅出Redis

1、简介

redis是一种主流的NOSQL，结构化数据库。(免费开源)

redis可以做：

1. 内存存储、持久化（内存断点即失）
2. 高速缓存，效率高
3. 发布订阅系统
4. 地图信息分析
5. 计时器、计数器（浏览量）
6. …

特性：

1. 多样化的数据类型
2. 持久化
3. 集群
4. 事务
5. …

官网：https://redis.io/

中文网：https://www.redis.net.cn/

不推荐在windows环境运行，建议在linux环境运行

对c有更好的支持

2、安装

系统环境：centos7.9

我安装的redis版本是目前的最新版：7.2.0

需要有gcc环境，安装gcc

yum install gcc-c++

1、下载安装包

官网下载地址：https://redis.io/download/

2、解压

.tar.tz文件解压

tar -zxvf /home/local/redis-7.2.4.tar.gz (你的完整路径)

3、编译

进入解压的目录进行编译:

cd xxx
make

4、启动

进入src目录,启动服务

cd src
./redis-server

启动客服端：

./redis-cli

5、测试

可以看到redis默认端口：6379（一个女星的名字）

6、查看进程

ps -ef|grep redis

7、关闭服务

shutdown #关闭
exit #退出

例子：

[root@iZgw02lo2mc5w2j7dgdd5zZ src]# ./redis-cli
127.0.0.1:6379> shutdown
not connected> exit

3、压力测试

官网原文：

Redis includes the redis-benchmark utility that simulates running commands done by N clients while at the same time sending M total queries. The utility provides a default set of tests, or you can supply a custom set of tests.

大意，redis有一个redis-benchmark的工具，可以模拟多个客户端执行命令，同时发送多个查询命令。提供默认测试，也可以自定义测试。

提供以下选项：

Usage: redis-benchmark [-h <host>] [-p <port>] [-c <clients>] [-n <requests]> [-k <boolean>]

 -h <hostname>      Server hostname (default 127.0.0.1)#主机名
 -p <port>          Server port (default 6379)#端口
 -s <socket>        Server socket (overrides host and port)#覆盖主机和端口
 -a <password>      Password for Redis Auth#授权密码
 -c <clients>       Number of parallel connections (default 50)#并发连接数
 -n <requests>      Total number of requests (default 100000)#总请求数
 -d <size>          Data size of SET/GET value in bytes (default 3)#get/set的数据大小
 --dbnum <db>       SELECT the specified db number (default 0)#查询明确的数据库数
 -k <boolean>       1=keep alive 0=reconnect (default 1)
 -r <keyspacelen>   Use random keys for SET/GET/INCR, random values for SADD
  Using this option the benchmark will expand the string __rand_int__
  inside an argument with a 12 digits number in the specified range
  from 0 to keyspacelen-1. The substitution changes every time a command
  is executed. Default tests use this to hit random keys in the
  specified range.#使用随机Key,SADD使用随机量
 -P <numreq>        Pipeline <numreq> requests. Default 1 (no pipeline).#通过管道传输
 -q                 Quiet. Just show query/sec values#强制退出，仅显示query/sec值
 --csv              Output in CSV format#以CSV输出
 -l                 Loop. Run the tests forever#生成循环，永久执行测试
 -t <tests>         Only run the comma separated list of tests. The test
                    names are the same as the ones produced as output.#仅运行以逗号分隔的测试命令列表
 -I                 Idle mode. Just open N idle connections and wait.#Idle模式，仅打开N个Idle链接并等待

测试100个并发连接，100k个请求：

./redis-benchmark -c 100 -n 100000

结果如：

4、基础知识

4.1、切换数据库

查看配置文件redis.conf。可以看到，redis默认16个数据库，从0-15。

使用select切换数据库：

4.2、清除数据库

使用flushdb默认清除0号数据库，flushall清除所有数据库，例子：

127.0.0.1:6379[2]> flushdb
OK
127.0.0.1:6379[2]> flushall
OK

4.3、Redis单线程&多线程

Redis 是从4开始慢慢支持多线程的，直到 Redis6/7 后才稳定，并不严谨，单线程还是多线程需要视版本而定。

版本3.×（最早版本），也就是大家口口相传的Redis是单线程；

版本4.×，严格意义来说也不是单线程，而是负责处理客户端请求的线程单线程，但是开始加了点多线程的东西（异步删除）；

版本6.x 开始，全面支持多线程。

开启多线程：

Redis7将所有的数据放在内存中，内存的响应时长大约为100纳秒，对于小数据，Redis服务器可以处理8W到10W的QPS（实验室数据，极限），但是对于大部分的公司已经够用了，所以在Redis6.0以后，多线程机制默认是关闭的，如果需要使用，则要在redis.conf配置文件中修改，主要改两个地方：

1. 设置io-threads-do-redis配置项为yes，表示启动多线程
2. 设置线程个数，io-threads 个数，关于个数的设置，官方的建议是如果为4核的CPU，建议线程数设置为2或3，如果为8核CPU建议线程数设置为6（线程数一定要小于机器核数，并非越大越好）

4.3.1、单线程

redis单线程主要是指Redis的网络IO和键值对读写是由一个线程来完成的，Redis在处理客户端的请求时包括获取 (socket 读)、解析、执行、内容返回 (socket 写) 等都由一个顺序串行的主线程处理，这就是所谓的“单线程”。这也是Redis对外提供键值存储服务的主要流程。

但Redis的其他功能，比如持久化RDB、AOF、异步删除、集群数据同步等等，其实是由额外的线程执行的。

高性能服务器不一定的多线程的。CPU线程切换是需要大量机器时钟的，redis是内存操作密集型，CPU的操作一般不是Redis的性能瓶颈。Redis的性能瓶颈是根据机器的内存和网络带宽。

4.3.2、多线程

从redis6.x开始采用io多路复用，让单个线程高效的处理多个连接请求（尽量减少网络IO的时间消耗），将最耗时的Socket的读取、请求解析、写入单独外包出去，剩下的命令执行仍然由主线程串行执行并和内存的数据交互。

4.3.3、为什么这么快

1. 基于内存
2. 简单的数据结构
3. 基于io多路复用
4. 避免上下文切换

使用一个线程来处理多个客户端请求，减少线程上下文切换带来的开销，同时避免了I/O阻塞操作。

主要原因是：IO多路复用+epoll函数使用，才是redis为什么这么快的直接原因，而不是仅仅单线程命令+redis安装在内存中。

什么是epoll？（来源百度百科）

epoll是Linux内核为处理大批量文件描述符而作了改进的poll，是Linux下多路复用IO接口select/poll的增强版本，它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。另一点原因就是获取事件的时候，它无须遍历整个被侦听的描述符集，只要遍历那些被内核IO事件异步唤醒而加入Ready队列的描述符集合就行了。epoll除了提供select/poll那种IO事件的水平触发（Level Triggered）外，还提供了边缘触发（Edge Triggered），这就使得用户空间程序有可能缓存IO状态，减少epoll_wait/epoll_pwait的调用，提高应用程序效率。

5、五大数据类型

5.1、Redis-Key

基本语法：

redis 127.0.0.1:6379> COMMAND KEY_NAME

实例：

127.0.0.1:6379> keys *
1) "h2"
2) "h1"
3) "hello"
127.0.0.1:6379> del h2
(integer) 1

基本命令：

序号	命令及描述
1	del key 该命令用于在 key 存在是删除 key。
2	dump key 序列化给定 key ，并返回被序列化的值。
3	exists key 检查给定 key 是否存在。
4	expire key seconds 为给定 key 设置过期时间。
5	EXPIREAT key timestamp EXPIREAT 的作用和 EXPIRE 类似，都用于为 key 设置过期时间。 不同在于 EXPIREAT 命令接受的时间参数是 UNIX 时间戳(unix timestamp)。
6	PEXPIRE key milliseconds 设置 key 的过期时间亿以毫秒计。
7	PEXPIREAT key milliseconds-timestamp 设置 key 过期时间的时间戳(unix timestamp) 以毫秒计
8	KEYS pattern 查找所有符合给定模式( pattern)的 key 。
9	MOVE key db 将当前数据库的 key 移动到给定的数据库 db 当中。
10	PERSIST key 移除 key 的过期时间，key 将持久保持。
11	PTTL key 以毫秒为单位返回 key 的剩余的过期时间。
12	TTL key 以秒为单位，返回给定 key 的剩余生存时间(TTL, time to live)。
13	RANDOMKEY 从当前数据库中随机返回一个 key 。
14	RENAME key newkey 修改 key 的名称
15	RENAMENX key newkey 仅当 newkey 不存在时，将 key 改名为 newkey 。
16	TYPE key 返回 key 所储存的值的类型。

5.2、String

常用命令：

• set/get设置获取key的值
• mset/mget 批量设置获取的值
• setnx 在key不存在时设置key的值
• setex 设置过期时间
• incr/decr 数字自增或自减一
• incrby/decrby数字加上或减去给定值
• append 最佳
• strlen 长度
• getset 先get在set

基本命令：

序号	命令及描述
1	set key value 设置指定 key 的值
2	get key 获取指定 key 的值。
3	getrange key start end 返回 key 中字符串值的子字符(左右都包)
4	GETSET key value 将给定 key 的值设为 value ，并返回 key 的旧值(old value)。
5	GETBIT key offset 对 key 所储存的字符串值，获取指定偏移量上的位(bit)。
6	MGET key1 [key2..] 获取所有(一个或多个)给定 key 的值。
7	SETBIT key offset value 对 key 所储存的字符串值，设置或清除指定偏移量上的位(bit)。
8	SETEX key seconds value 将值 value 关联到 key ，并将 key 的过期时间设为 seconds (以秒为单位)。
9	SETNX key value 只有在 key 不存在时设置 key 的值。
10	SETRANGE key offset value 用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始，offset为-1可以获取到结尾。
11	STRLEN key 返回 key 所储存的字符串值的长度。
12	MSET key value key value … 同时设置一个或多个 key-value 对。
13	MSETNX key value key value … 同时设置一个或多个 key-value 对，当且仅当所有给定 key 都不存在。
14	PSETEX key milliseconds value 这个命令和 SETEX 命令相似，但它以毫秒为单位设置 key 的生存时间，而不是像 SETEX 命令那样，以秒为单位。
15	INCR key 将 key 中储存的数字值增一。
16	INCRBY key increment 将 key 所储存的值加上给定的增量值（increment） 。
17	INCRBYFLOAT key increment 将 key 所储存的值加上给定的浮点增量值（increment） 。
18	DECR key 将 key 中储存的数字值减一。
19	DECRBY key decrement key 所储存的值减去给定的减量值（decrement） 。
20	APPEND key value 如果 key 已经存在并且是一个字符串， APPEND 命令将 value 追加到 key 原来的值的末尾。

例子：

127.0.0.1:6379> APPEND hello ,world
(integer) 15
127.0.0.1:6379> get hello
"worldadsa,world"
127.0.0.1:6379> GETRANGE hello 2 -1
"rldadsa,world"

用法：

• set user:1 {name:jiuzhao,age:18}
• mset user:1:name jiuzhao user:1:age 18

推荐使用第二种，不需要解析json

使用场景：

• 计数器
• 统计多单位的数量
• 粉丝数
• 对象缓存存储

5.3、List

List左边为头，右为尾

基本命令：

序号	命令及描述
1	BLPOP key1 [key2 ] timeout移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
2	BRPOP key1 [key2 ] timeout 移出并获取列表的最后一个元素， 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
3	BRPOPLPUSH source destination timeout 从列表中弹出一个值，将弹出的元素插入到另外一个列表中并返回它； 如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止。
4	LINDEX key index 通过索引获取列表中的元素
5	[LINSERT key BEFORE
6	LLEN key 获取列表长度
7	LPOP key 移出并获取列表的第一个元素
8	LPUSH key value1 [value2] 将一个或多个值插入到列表头部
9	LPUSHX key value 将一个或多个值插入到已存在的列表头部
10	LRANGE key start stop 获取列表指定范围内的元素，stop为-1可以获取到结尾
11	LREM key count value 移除列表元素
12	LSET key index value 通过索引设置列表元素的值
13	LTRIM key start stop 对一个列表进行修剪(trim)，就是说，让列表只保留指定区间内的元素，不在指定区间之内的元素都将被删除。
14	RPOP key 移除并获取列表最后一个元素
15	RPOPLPUSH source destination 移除列表的最后一个元素，并将该元素添加到另一个列表并返回
16	RPUSH key value1 [value2] 在列表中添加一个或多个值
17	RPUSHX key value 为已存在的列表添加值

例子：

127.0.0.1:6379> rpush list one two three four
(integer) 4
127.0.0.1:6379> LRANGE list 0 -1
1) "one"
2) "two"
3) "three"
4) "four"
127.0.0.1:6379> rpush list one two three four
(integer) 8
127.0.0.1:6379> LRANGE list 0 -1
1) "one"
2) "two"
3) "three"
4) "four"
5) "one"
6) "two"
7) "three"
8) "four"
127.0.0.1:6379> lrem list 1 one
(integer) 1
127.0.0.1:6379> LRANGE list 0 -1
1) "two"
2) "three"
3) "four"
4) "one"
5) "two"
6) "three"
7) "four"
127.0.0.1:6379> lrem list 2 two
(integer) 2
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "four"
3) "one"
4) "three"
5) "four"
###
127.0.0.1:6379> LINDEX list 1
"four"
127.0.0.1:6379> LINDEX list 2
"one"
127.0.0.1:6379> LLEN list
(integer) 5
###trim
127.0.0.1:6379> LRANGE list 0 -1
1) "three"
2) "four"
3) "one"
4) "three"
5) "four"
127.0.0.1:6379> LTRIM list 2 3
OK
127.0.0.1:6379> LRANGE list 0 -1
1) "one"
2) "three"

使用场景：

• 消息队列
• 栈

5.4、Hash

基本命令：

序号	命令及描述
1	HDEL key field2 [field2] 删除一个或多个哈希表字段
2	HEXISTS key field 查看哈希表 key 中，指定的字段是否存在。
3	HGET key field 获取存储在哈希表中指定字段的值/td>
4	HGETALL key 获取在哈希表中指定 key 的所有字段和值
5	HINCRBY key field increment 为哈希表 key 中的指定字段的整数值加上增量 increment 。
6	HINCRBYFLOAT key field increment 为哈希表 key 中的指定字段的浮点数值加上增量 increment 。
7	HKEYS key 获取所有哈希表中的字段
8	HLEN key 获取哈希表中字段的数量
9	HMGET key field1 [field2] 获取所有给定字段的值
10	HMSET key field1 value1 [field2 value2 ] 同时将多个 field-value (域-值)对设置到哈希表 key 中。
11	HSET key field value 将哈希表 key 中的字段 field 的值设为 value 。
12	HSETNX key field value 只有在字段 field 不存在时，设置哈希表字段的值。
13	HVALS key 获取哈希表中所有值
14	HSCAN key cursor [MATCH pattern] [COUNT count] 迭代哈希表中的键值对。

使用场景：

• 变更数据，尤其是用户信息之类的，经常变动的信息。

5.5、Set

基本命令：

序号	命令及描述
1	SADD key member1 [member2]向集合添加一个或多个成员
2	SCARD key 获取集合的成员数
3	SDIFF key1 [key2]返回给定所有集合的差集
4	SDIFFSTORE destination key1 [key2]返回给定所有集合的差集并存储在 destination 中
5	SINTER key1 [key2]返回给定所有集合的交集
6	SINTERSTORE destination key1 [key2] 返回给定所有集合的交集并存储在 destination 中
7	SISMEMBER key member 判断 member 元素是否是集合 key 的成员
8	SMEMBERS key 返回集合中的所有成员
9	SMOVE source destination member 将 member 元素从 source 集合移动到 destination 集合
10	SPOP key 移除并返回集合中的一个随机元素
11	SRANDMEMBER key [count] 返回集合中一个或多个随机数
12	SREM key member1 [member2] 移除集合中一个或多个成员
13	SUNION key1 [key2] 返回所有给定集合的并集
14	SUNIONSTORE destination key1 [key2] 所有给定集合的并集存储在 destination 集合中
15	[SSCAN key cursor MATCH pattern] [COUNT count] 迭代集合中的元素

例子：

127.0.0.1:6379> sadd set a b c d e f g h i j k
(integer) 11
127.0.0.1:6379> SCARD set
(integer) 11
127.0.0.1:6379> SMEMBERS set
 1) "a"
 2) "b"
 3) "c"
 4) "d"
 5) "e"
 6) "f"
 7) "g"
 8) "h"
 9) "i"
10) "j"
11) "k"
127.0.0.1:6379> SREM set 0
(integer) 0
127.0.0.1:6379> SREM set a
(integer) 1
127.0.0.1:6379> SMEMBERS set
 1) "b"
 2) "c"
 3) "d"
 4) "e"
 5) "f"
 6) "g"
 7) "h"
 8) "i"
 9) "j"
10) "k"
127.0.0.1:6379> SRANDMEMBER set 
"i"
127.0.0.1:6379> SRANDMEMBER set 2
1) "e"
2) "i"
127.0.0.1:6379> SRANDMEMBER set 2
1) "c"
2) "j"

5.6、Zset

有序集合。

基本命令：

序号	命令及描述
1	ZADD key score1 member1 [score2 member2] 向有序集合添加一个或多个成员，或者更新已存在成员的分数
2	ZCARD key 获取有序集合的成员数
3	ZCOUNT key min max 计算在有序集合中指定区间分数的成员数
4	ZINCRBY key increment member 有序集合中对指定成员的分数加上增量 increment
5	ZINTERSTORE destination numkeys key [key …] 计算给定的一个或多个有序集的交集并将结果集存储在新的有序集合 key 中
6	ZLEXCOUNT key min max 在有序集合中计算指定字典区间内成员数量
7	ZRANGE key start stop [WITHSCORES] 通过索引区间返回有序集合成指定区间内的成员
8	ZRANGEBYLEX key min max [LIMIT offset count] 通过字典区间返回有序集合的成员
9	[ZRANGEBYSCORE key min max WITHSCORES] [LIMIT] 通过分数返回有序集合指定区间内的成员，-inf,+inf、-1，0都可表示±∞。使用（表示开闭
10	ZRANK key member 返回有序集合中指定成员的索引
11	ZREM key member [member …] 移除有序集合中的一个或多个成员
12	ZREMRANGEBYLEX key min max 移除有序集合中给定的字典区间的所有成员
13	ZREMRANGEBYRANK key start stop 移除有序集合中给定的排名区间的所有成员
14	ZREMRANGEBYSCORE key min max 移除有序集合中给定的分数区间的所有成员
15	ZREVRANGE key start stop [WITHSCORES] 返回有序集中指定区间内的成员，通过索引，分数从高到底
16	ZREVRANGEBYSCORE key max min [WITHSCORES] 返回有序集中指定分数区间内的成员，分数从高到低排序
17	ZREVRANK key member 返回有序集合中指定成员的排名，有序集成员按分数值递减(从大到小)排序
18	ZSCORE key member 返回有序集中，成员的分数值
19	ZUNIONSTORE destination numkeys key [key …] 计算给定的一个或多个有序集的并集，并存储在新的 key 中
20	[ZSCAN key cursor MATCH pattern] [COUNT count] 迭代有序集合中的元素（包括元素成员和元素分值）

使用场景：

• set排序 存储班级成绩表，工资表排序
• 权重排序，如普通消息，重要消息
• 排行榜应用实现

6、四种特殊数据类型

6.1、Geospatial

定位，附近的人，打车距离计算

Redis3.2推出，底层原理Zset，因此Zset的命令Geospatial可以使用1！

基本命令：

命令	描述
Redis GEOHASH 命令	返回一个或多个位置元素的 Geohash 表示，将二维的经纬度转换为一维的字符串
Redis GEOPOS 命令	从key里返回所有给定位置元素的位置（经度和纬度）
Redis GEODIST 命令	返回两个给定位置之间的距离
Redis GEORADIUS 命令	以给定的经纬度为中心， 找出某一半径内的元素
Redis GEOADD 命令	将指定的地理空间位置（纬度、经度、名称）添加到指定的key中
Redis GEORADIUSBYMEMBER 命令	找出位于指定范围内的元素，中心点是由给定的位置元素决定

6.2、HyperLogLog

什么是基数？

基数是一个数学术语，它表示一个集合中元素的数量或大小。例如，如果一个集合有三个元素，那么它的基数就是3。基数可以用来比较不同集合的大小，判断它们是否有相同的元素个数。

简介：

Redis2.8.9版本更新了HyperLogLog数据结构。

HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定 的、并且是很小的。

在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基 数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。

但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素。

0.81%的错误率！

使用场景：

网页的UV(独立访客)（一个人访问一个网站多次，但是算作一个人）

传统方式set保存用户id，需要大量存储空间。目的是为了计数，而不是保存用户id。

基本命令：

序号	命令及描述
1	PFADD key element [element …] 添加指定元素到 HyperLogLog 中。
2	PFCOUNT key [key …] 返回给定 HyperLogLog 的基数估算值。
3	PFMERGE destkey sourcekey [sourcekey …] 将多个 HyperLogLog 合并为一个 HyperLogLog

6.3、Bitmaps

位存储

统计用户信息，活跃，不活跃！登入，未登入！打卡，365打卡！

表现：

SETBIT且GETBIT均为 O(1)。 BITOP是 O(n)，其中n是比较中最长字符串的长度。

基本命令：

命令	描述
bitcount	计算字符串中设置位的数量
bitfield	对字符串执行任意位域整数操作
bitfield_ro	对字符串执行只读位域整数操作
bitop	对多个字符串执行按位运算，并存储结果
bitpos	查找字符串中第一个设置的（1）
getbit	按偏移量返回一个位值
setbit	设置或清理字符串值偏移处的位。如果key不存在，创建该key

###################################
#bitfield
#如果偏移量以#字符为前缀，则指定的偏移量将乘以整数编码的宽度
#新玩家以 1000 金币开始教程（偏移量为 0
127.0.0.1:6379> BITFIELD player:1:stats SET u32 #0 1000
1) (integer) 0
#杀死俘虏王子的妖精后，添加获得的 50 金币并增加“被杀”计数器（偏移量 1）。
127.0.0.1:6379> BITFIELD player:1:stats INCRBY u32 #0 50 INCRBY u32 #1 1
1) (integer) 1050
2) (integer) 1
#向铁匠支付 999 金币，即可购买一把传说中的生锈匕首。
127.0.0.1:6379> BITFIELD player:1:stats INCRBY u32 #0 -999
1) (integer) 
#阅读玩家的统计数据：
127.0.0.1:6379> BITFIELD player:1:stats GET u32 #0 GET u32 #1
1) (integer) 51
2) (integer) 1

6.4、Stream

文档介绍：

Redis Streams是一种数据结构，其作用类似于仅附加日志。您可以使用流实时记录和同步事件。Redis 流用例的示例包括：

• 事件溯源（例如，跟踪用户操作、点击等）
• 传感器监控（例如，现场设备的读数）
• 通知（例如，将每个用户的通知记录存储在单独的流中）

Redis 为每个流条目生成一个唯一的 ID。您可以使用这些 ID 稍后检索其关联条目或读取和处理流中的所有后续条目。

Redis 流支持多种修剪策略（以防止流无限增长）和一种以上的消费策略（请参阅 XREAD、 XREADGROUP 和 XRANGE ）。

基本命令：

• XADD 向流中添加新条目。
• XREAD 读取一个或多个条目，从给定位置开始并及时向前移动。
• XRANGE 返回两个提供的条目 ID 之间的条目范围。
• XLEN 返回流的长度。

表现：

向流中添加条目是 O(1)。访问任何单个条目都是 O(n)，其中n是 ID 的长度。由于流 ID 通常很短且具有固定长度，因此这有效地减少了恒定时间查找。有关原因的详细信息，请注意流是作为 基数树 实现的。

7、事务

事务本质：一组命令的集合。一个事务中的所有命令都会被序列化，事务执行过程中，会按照顺序。

一次性、顺序性、排他性。

Redis事务没有隔离级别的概念。

redis可以做事务，但是redis的事务没有原子性，redis单条命令保持原子性。

7.1、multi

redis的事务：

• 开启事务（multi）
• 命令入队（…）
• 执行事务（exec）

例子：

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> keys *
QUEUED
127.0.0.1:6379(TX)> EXEC
1) OK
2) OK
3) 1) "k2"
   2) "k1"
   3) "player:1:stats"

7.2、discard

放弃事务：discard，所有事务都不会被执行

例子：

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> get k1
QUEUED
127.0.0.1:6379(TX)> DISCARD
OK

7.3、异常

编译型异常，事务中所有命令都不会被执行

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> getset k3
(error) ERR wrong number of arguments for 'getset' command
127.0.0.1:6379(TX)> EXEC
(error) EXECABORT Transaction discarded because of previous errors.
127.0.0.1:6379> get k1
(nil)

运行时异常：其他命令可以正常执行，错误命令抛出异常

127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> set k1 v1
QUEUED
127.0.0.1:6379(TX)> INCR k1
QUEUED
127.0.0.1:6379(TX)> set k2 v2
QUEUED
127.0.0.1:6379(TX)> get k1
QUEUED
127.0.0.1:6379(TX)> get k2
QUEUED
127.0.0.1:6379(TX)> exec
1) OK
2) (error) ERR value is not an integer or out of range
3) OK
4) "v1"
5) "v2"
127.0.0.1:6379> get k1
"v1"

7.4、监控

悲观锁：很悲观，认为什么时候都会出问题，无论做什么都会加锁

乐观锁：很乐观，认为什么时候都不会出问题，所以不会上锁！更新数据的时候判断一下，在此期间是否有人修改过这个数据。获取version,更新的时候比较version。

watch监视

exec、unwatch会退出监控

127.0.0.1:6379> watch money
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> DECRBY money 10
QUEUED
127.0.0.1:6379(TX)> INCRBY out 10
QUEUED
127.0.0.1:6379(TX)> exec
1) (integer) 90
2) (integer) 10

修改失败：

#在exec之前，另一个客户端对money进行赋值操作
127.0.0.1:6379> WATCH money
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> DECRBY out 30
QUEUED
127.0.0.1:6379(TX)> INCRBY money 30
QUEUED
127.0.0.1:6379(TX)> exec
(nil)
127.0.0.1:6379> get money
"90"
127.0.0.1:6379> get out
"40"

使用unwatch：

127.0.0.1:6379> unwatch
OK
127.0.0.1:6379> WATCH money
OK
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379(TX)> INCRBY money 10
QUEUED
127.0.0.1:6379(TX)> DECRBY out 10
QUEUED
127.0.0.1:6379(TX)> EXEC
1) (integer) 100
2) (integer) 30

8、Jedis

导入对应依赖

<!-- https://mvnrepository.com/artifact/redis.clients/jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>5.1.0</version>
</dependency>

直接new 一个Jedis对象

9、SpringBoot整合

在springboot2.x之后jedis被替换为了lettuce

jedis：采用直连，多个线程操作的话，是不安全的，如果想要避免不安全的，使用jedis pool连接池。更新BIO模式。

lettuce：采用netty，实例可以在多个线程中进行共享，不存在线程不安全的情况！可以减少线程数量，更像NIO模式。

springboot 所有配置类都有一个自动配置类 RedisAutoConfiguration

自动配置类会绑定一个properties配置文件 RedisProperties

源码分析：

@Bean
	@ConditionalOnMissingBean(name = "redisTemplate")
	@ConditionalOnSingleCandidate(RedisConnectionFactory.class)//可以自己定义一个RedisTemplate来替换这个默认的
	public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
		//默认的RedisTemplate没有过多设置，redis对象都是需要序列化！
        //俩个泛型都是Object,Object的类型，我们使用需要强制类型转换为<String,Object>
		RedisTemplate<Object, Object> template = new RedisTemplate<>();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}

	@Bean
	@ConditionalOnMissingBean
	@ConditionalOnSingleCandidate(RedisConnectionFactory.class)//由于String是redis最常使用的类型，单独使用一个bean
	public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
		StringRedisTemplate template = new StringRedisTemplate();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}

1、导入依赖

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-redis</artifactId>
    <version>3.2.1</version>
</dependency>

2、配置连接

spring:  
  redis:
    # Redis数据库索引（默认为0）
    database: 0
    # Redis服务器地址
    host: 47.122.58.xxx
    # Redis服务器连接端口
    port: 6379
    timeout: 5000
    password: xxxx

3、测试

bitmaps在opsForValue

@Test
public void contextLoads(){
    System.out.println("hello");
    //opsForValue String,Bitmaps
    //opsForList List
    //opsForSet set
    //opsForHash hash
    //opsForZSet ZSet
    //opsForGeo geo
    //opsForHyperLogLog
    //...
    redisTemplate.opsForValue().set("test1","t1");
    
    Object test1 = redisTemplate.opsForValue().get("test1");
    System.out.println(test1);
}

自定义redis配置类

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {

        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(redisConnectionFactory);

        // 使用Jackson2JsonRedisSerializer来序列化和反序列化redis的value值
        // 解决查询缓存转换异常的问题
        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper objectMapper = new ObjectMapper();
        // 指定要序列化的域，field,get和set,以及修饰符范围，ANY是都有包括private和public
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 指定序列化输入的类型，类必须是非final修饰的，final修饰的类，比如String,Integer等会跑出异常
        //objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL);
        objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
        objectMapper.setTimeZone(TimeZone.getDefault());
        objectMapper.setPropertyNamingStrategy(PropertyNamingStrategies.LOWER_CAMEL_CASE);
        // 确定遇到未知属性（未映射到属性，并且没有“任何setter”或处理程序可以处理它）是否会导致失败（通过抛出JsonMappingException）的功能。
        // 此设置仅在尝试了未知属性的所有其他处理方法后生效，并且属性保持未处理状态
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        //序列号key value
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(jackson2JsonRedisSerializer);

        template.afterPropertiesSet();

        return template;
    }
}

RedisUtil工具类：

@Component
public class RedisUtil {

    @Autowired
    //@Qualifier("myRedisTemplate") //一定要保证是我们自己写的RedisTemplate
    private RedisTemplate<String, Object> redisTemplate;

    // =============================common============================

    /**
     * 指定缓存失效时间
     *
     * @param key  键
     * @param time 时间(秒)
     */
    public boolean expire(String key, long time) {
        try {
            if (time > 0) {
                redisTemplate.expire(key, time, TimeUnit.SECONDS);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 根据key 获取过期时间
     *
     * @param key 键 不能为null
     * @return 时间(秒) 返回0代表为永久有效
     */
    public long getExpire(String key) {
        return redisTemplate.getExpire(key, TimeUnit.SECONDS);
    }


    /**
     * 判断key是否存在
     *
     * @param key 键
     * @return true 存在 false不存在
     */
    public boolean hasKey(String key) {
        try {
            return redisTemplate.hasKey(key);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除缓存
     *
     * @param key 可以传一个值 或多个
     */
    @SuppressWarnings("unchecked")
    public void del(String... key) {
        if (key != null && key.length > 0) {
            if (key.length == 1) {
                redisTemplate.delete(key[0]);
            } else {
                redisTemplate.delete(String.valueOf(CollectionUtils.arrayToList(key)));
            }
        }
    }


    // ============================String=============================

    /**
     * 普通缓存获取
     *
     * @param key 键
     * @return 值
     */
    public Object get(String key) {
        return key == null ? null : redisTemplate.opsForValue().get(key);
    }

    /**
     * 普通缓存放入
     *
     * @param key   键
     * @param value 值
     * @return true成功 false失败
     */

    public boolean set(String key, Object value) {
        try {
            redisTemplate.opsForValue().set(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 普通缓存放入并设置时间
     *
     * @param key   键
     * @param value 值
     * @param time  时间(秒) time要大于0 如果time小于等于0 将设置无限期
     * @return true成功 false 失败
     */

    public boolean set(String key, Object value, long time) {
        try {
            if (time > 0) {
                redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
            } else {
                set(key, value);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 递增
     *
     * @param key   键
     * @param delta 要增加几(大于0)
     */
    public long incr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, delta);
    }


    /**
     * 递减
     *
     * @param key   键
     * @param delta 要减少几(小于0)
     */
    public long decr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, -delta);
    }


    // ================================Map=================================

    /**
     * HashGet
     *
     * @param key  键 不能为null
     * @param item 项 不能为null
     */
    public Object hget(String key, String item) {
        return redisTemplate.opsForHash().get(key, item);
    }

    /**
     * 获取hashKey对应的所有键值
     *
     * @param key 键
     * @return 对应的多个键值
     */
    public Map<Object, Object> hmget(String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * HashSet
     *
     * @param key 键
     * @param map 对应多个键值
     */
    public boolean hmset(String key, Map<String, Object> map) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * HashSet 并设置时间
     *
     * @param key  键
     * @param map  对应多个键值
     * @param time 时间(秒)
     * @return true成功 false失败
     */
    public boolean hmset(String key, Map<String, Object> map, long time) {
        try {
            redisTemplate.opsForHash().putAll(key, map);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 向一张hash表中放入数据,如果不存在将创建
     *
     * @param key   键
     * @param item  项
     * @param value 值
     * @return true 成功 false失败
     */
    public boolean hset(String key, String item, Object value) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 向一张hash表中放入数据,如果不存在将创建
     *
     * @param key   键
     * @param item  项
     * @param value 值
     * @param time  时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
     * @return true 成功 false失败
     */
    public boolean hset(String key, String item, Object value, long time) {
        try {
            redisTemplate.opsForHash().put(key, item, value);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 删除hash表中的值
     *
     * @param key  键 不能为null
     * @param item 项 可以使多个 不能为null
     */
    public void hdel(String key, Object... item) {
        redisTemplate.opsForHash().delete(key, item);
    }


    /**
     * 判断hash表中是否有该项的值
     *
     * @param key  键 不能为null
     * @param item 项 不能为null
     * @return true 存在 false不存在
     */
    public boolean hHasKey(String key, String item) {
        return redisTemplate.opsForHash().hasKey(key, item);
    }


    /**
     * hash递增 如果不存在,就会创建一个 并把新增后的值返回
     *
     * @param key  键
     * @param item 项
     * @param by   要增加几(大于0)
     */
    public double hincr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, by);
    }


    /**
     * hash递减
     *
     * @param key  键
     * @param item 项
     * @param by   要减少记(小于0)
     */
    public double hdecr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, -by);
    }


    // ============================set=============================

    /**
     * 根据key获取Set中的所有值
     *
     * @param key 键
     */
    public Set<Object> sGet(String key) {
        try {
            return redisTemplate.opsForSet().members(key);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 根据value从一个set中查询,是否存在
     *
     * @param key   键
     * @param value 值
     * @return true 存在 false不存在
     */
    public boolean sHasKey(String key, Object value) {
        try {
            return redisTemplate.opsForSet().isMember(key, value);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 将数据放入set缓存
     *
     * @param key    键
     * @param values 值 可以是多个
     * @return 成功个数
     */
    public long sSet(String key, Object... values) {
        try {
            return redisTemplate.opsForSet().add(key, values);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 将set数据放入缓存
     *
     * @param key    键
     * @param time   时间(秒)
     * @param values 值 可以是多个
     * @return 成功个数
     */
    public long sSetAndTime(String key, long time, Object... values) {
        try {
            Long count = redisTemplate.opsForSet().add(key, values);
            if (time > 0) {
                expire(key, time);
            }
            return count;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 获取set缓存的长度
     *
     * @param key 键
     */
    public long sGetSetSize(String key) {
        try {
            return redisTemplate.opsForSet().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 移除值为value的
     *
     * @param key    键
     * @param values 值 可以是多个
     * @return 移除的个数
     */

    public long setRemove(String key, Object... values) {
        try {
            Long count = redisTemplate.opsForSet().remove(key, values);
            return count;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }

    // ===============================list=================================

    /**
     * 获取list缓存的内容
     *
     * @param key   键
     * @param start 开始
     * @param end   结束 0 到 -1代表所有值
     */
    public List<Object> lGet(String key, long start, long end) {
        try {
            return redisTemplate.opsForList().range(key, start, end);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 获取list缓存的长度
     *
     * @param key 键
     */
    public long lGetListSize(String key) {
        try {
            return redisTemplate.opsForList().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 通过索引 获取list中的值
     *
     * @param key   键
     * @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
     */
    public Object lGetIndex(String key, long index) {
        try {
            return redisTemplate.opsForList().index(key, index);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     */
    public boolean lSet(String key, Object value) {
        try {
            redisTemplate.opsForList().rightPush(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @param time  时间(秒)
     */
    public boolean lSet(String key, Object value, long time) {
        try {
            redisTemplate.opsForList().rightPush(key, value);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @return
     */
    public boolean lSet(String key, List<Object> value) {
        try {
            redisTemplate.opsForList().rightPushAll(key, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

    }


    /**
     * 将list放入缓存
     *
     * @param key   键
     * @param value 值
     * @param time  时间(秒)
     * @return
     */
    public boolean lSet(String key, List<Object> value, long time) {
        try {
            redisTemplate.opsForList().rightPushAll(key, value);
            if (time > 0) {
                expire(key, time);
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 根据索引修改list中的某条数据
     *
     * @param key   键
     * @param index 索引
     * @param value 值
     * @return
     */

    public boolean lUpdateIndex(String key, long index, Object value) {
        try {
            redisTemplate.opsForList().set(key, index, value);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }


    /**
     * 移除N个值为value
     *
     * @param key   键
     * @param count 移除多少个
     * @param value 值
     * @return 移除的个数
     */

    public long lRemove(String key, long count, Object value) {
        try {
            Long remove = redisTemplate.opsForList().remove(key, count, value);
            return remove;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }
}

10、Redis.conf详解

redis-server启动时读取配置文件

./redis-server redis.conf

查看配置文件

单位配置

不区分大小写

包含

导入一些子配置

模型

在启动时加载模块。如果服务器无法加载模块，它将中止。可以使用多个loadmodule指令。

网络

绑定主机ip，除此之外不许访问，

保护模式是否开启

端口设置

TLS/SSL

加密协议

general 通用设置

后台运行，默认no

若以后台方式运行，我们需要指定一个pid文件，进程文件。

日志生成规则，debug，verbose，notice，warning，nothing

日志文件名：

数据库数量设置：

是否总是显示logo，7.2默认no：

快照 SNAPSHOTTING

持久化，在规定时间内，执行多少次操作，则会持久化到文件.rdb.aof

redis是内存数据库，如果没有持久化，那么数据断电即失。

翻译：

save <seconds> <changes> [<seconds> <changes> ...]

如果给定的秒数已过，并且已对数据库执行的写操作超过给定的数量，
Redis 将保存数据库。

快照可以通过单个空字符串参数完全禁用，例如下面的示例：

save ""

除非另有说明，否则默认情况下，Redis 将在以下情况保存数据库：

在3600秒（一小时）之后，如果至少执行了1次更改
在300秒（5分钟）之后，如果至少执行了100次更改
在60秒之后，如果至少执行了10000次更改
您可以通过取消注释以下行来显式设置这些值。

#持久化出错了是否继续工作
stop-writes-on-bgsave-error yes
#是否压缩rdb文件，需要消耗cpu资源
rdbcompression yes
#保持rdb文件进行错误校验
rdbchecksum yes
#转存的文件名
dbfilename dump.rdb
#rdb文件保存的目录
dir ./

复制 REPLICATION

安全 SECURITY

设置密码:

也可以在redis-cli中：

限制 CLIENTS MEMORY MANAGEMENT

#限制最大登入数
maxclients 10000
#最大内存大小
maxmemory <bytes>
#内存上线后处理策略
maxmemory-policy noeviction
#noeviction: 不删除策略, 达到最大内存限制时, 如果需要更多内存, 直接返回错误信息。
#allkeys-lru: 所有key通用; 优先删除最近最少使用(less recently used ,LRU) 的 key。
#volatile-lru: 只限于设置了 expire 的部分; 优先删除最近最少使用又过期(less recently used ,LRU) 的 key。
#allkeys-random: 所有key通用; 随机删除一部分 key。
#volatile-random: 只限于设置了 expire 的部分; 随机删除一部分过期 key。
#volatile-ttl: 只限于设置了 expire 的部分; 优先删除剩余时间(time to live,TTL) 短的key。

aof配置 APPEND ONLY MODE 模式

#默认不开启aof模式，默认使用rdb方式持久化的，在大部分的情况下，rdb完全可用！
appendonly no
#持久化文件名
appendfilename "appendonly.aof"
# 同步策略时间间隔
# appendfsync always #一直执行
appendfsync everysec #每秒执行一次
# appendfsync no #从不执行

11、Redis持久化

Redis是内存数据库，如果不将内存中的数据库保存到磁盘，那么一旦服务进程退出，服务器中的数据库数据也会消失。所以Redis推出了持久化功能。

Redis持久化选择有四种：

1. RDB
2. AOF
3. RDB与AOF混合
4. 无持久化

官方文档地址：https://redis.io/docs/management/persistence/

11.1、RDB，快照snapshotting

RDB（Redis DataBase）

什么是RDB？

Redis 可以通过创建快照来获得存储在内存里面的数据在 某个时间点 上的副本。Redis 创建快照之后，可以对快照进行备份，可以将快照复制到其他服务器从而创建具有相同数据的服务器副本（Redis 主从结构，主要用来提高 Redis 性能），还可以将快照留在原地以便重启服务器的时候使用。

快照持久化是 Redis 默认采用的持久化方式，在 redis.conf 配置文件中默认有此下配置：

save 3600 1 300 100 60 10000

子进程做数据持久化，它不会修改现有的内存数据结构，它只是对数据结构进行遍历读取，然后序列化写到磁盘中。

留一倍内存空间较为安全，redis数据占8G，还需预留8G，也就是说内存需求是16G。

什么时候产生rdb文件？

1. save规则满足的情况下，自动触发rdb规则
2. 执行flushdb
3. 推出redis

优点：

1. 适合大规模的数据恢复
2. 如果对数据完整性要求不高

缺点：

1. 需要一定的时间间隔，最后一次修改数据可能丢失
2. 会占用一定的内存空间

redis提供修复备份工具redis-check-rdb

11.2、AOF

AOF是什么？

AOF（Append Only File）

将我们的命令都记录下来，恢复的时候就把这个文件全部都执行一遍。

与快照持久化相比，AOF 持久化的实时性更好。默认情况下 Redis 没有开启 AOF（append only file）方式的持久化（Redis 6.0 之后已经默认是开启了），可以通过 appendonly 参数开启：

appendonly yes

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

AOF重写

如果aof文件太大了，fork一个新进程来将我们的文件进行重写。

AOF优点：

1. 每一次修改都同步，文件完整性更好。
2. 默认每秒同步一次
3. 不开启，从不同步，效率最高

缺点

1. 对于数据文件大小来说，aof远大于rdb，修复的速度也比rdb慢
2. aof运行效率比rdb慢，所以默认rdb

备份出问题，redis提供了一个工具redis-check-aof进行修复

redis-check-aof --fix appendonly.aof

11.3、混合模式

如果保存的数据要求安全性比较高的话，建议同时开启 RDB 和 AOF 持久化或者开启 RDB 和 AOF 混合持久化。

俩个都开的情况下，使用aof的备份。

12、Redis发布订阅

官网文档：

SUBSCRIBE，UNSUBSCRIBE并PUBLISH实现发布/订阅消息传递范式，其中（引用维基百科）发送者（发布者）没有被编程为将其消息发送到特定接收者（订阅者）。相反，发布的消息被表征为通道，而不知道可能有哪些订阅者（如果有）。订阅者表达对一个或多个频道的兴趣，并且仅接收感兴趣的消息，而不知道有哪些发布者（如果有）。发布者和订阅者的这种解耦允许更大的可扩展性和更动态的网络拓扑。

频道channel1订阅了三个客户端

担忧新消息publish发送给channel1时，会发送给三个客户端。

使用场景：

1. 实时消息系统
2. 实时聊天
3. 订阅，关注

基本命令：

序号	命令及描述
1	PSUBSCRIBE pattern [pattern …]订阅一个或多个符合给定模式的频道。
2	[PUBSUB subcommand argument [argument …]] 查看订阅与发布系统状态。
3	PUBLISH channel message 将信息发送到指定的频道。
4	[PUNSUBSCRIBE pattern [pattern …]] 退订所有给定模式的频道。
5	SUBSCRIBE channel [channel …] 订阅给定的一个或多个频道的信息。
6	[UNSUBSCRIBE channel [channel …]] 指退订给定的频道。

13、主从复制

13.1、简介

概念：
主从复制, 是指将一台Redis服务器的数据, 复制到其他的Redis服务器. 前者称为主节点(master/leader) , 后者称为从节点(slave/follower); 数据的复制是单向的, 只能由主节点到从节点, Master以写为主, Slave以读为主。

默认情况下, 每台Redis服务器都是主节点; 且一个主节点可以有多个从节点 (或没有从节点),但一个从节点只能有一个主节点。

例如：电商网站的商品，读多写少。

13.2、作用

主从复制的主要作用：

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

一般来说，要将Redis运用于工程项目中，只使用一台Redis是万万不能的，原因如下：

1. 从结构上，单给redis服务器会发生单点故障，并且一台服务器需要处理所有请求负载压力较大。
2. 从容量上，单个redis服务器内存容量有限，就算一台redis服务器内存容量为256G，也不能将所有内存用于Redis存储内存，一般来说，单台redis最大使用内存不应该超过20G。

13.3、单实例内存设置

Redis内存过大会怎么样？

在主库宕机的时候，我们最常见的容灾策略为“切主”。具体为从该集群剩余从库中选出一个从库并将其升级为主库，该从库升级为主库后再将剩余从库挂载至其下成为其从库，最终恢复整个主从集群结构。

redis集群中一旦从库换主，redis的做法是将更换主库的从库清空然后从新主库完整同步一份数据再进行续传。

当Redis服务器内存约为20G时，一个从库恢复时间约为18分钟（来源网上），10G约为8分钟，5G约为4分钟。因此单实例Redis内存大小一般最大设置10-20G，最好10G以下。

13.4、配置

只要配置从库，不用配置主库。

使用info replication查看当前库信息，如下：

127.0.0.1:6379> info replication#查看当前库信息
# Replication
role:master#角色
connected_slaves:0#没有从机
master_failover_state:no-failover
master_replid:a21472456872230d1d19bf4504c395f15e3955c7
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

修改对应信息：

1. 端口
2. pid名字
3. log文件名字
4. dump.rdb名字

命令配置：从库使用slaveof 127.0.0.1 6379

配置文件配置，修改从库redis.conf：

# replicaof <masterip> <masterport>
# masterauth <master-password>
...

13.5、细节

主机可以写，从机只能读！从机保存主机的数据。

主机断开连接，从机依旧连接到主机，但是没有写操作，这时候主机重连，从机可以直接获取主机的信息。

复制原理：

Slave启动成功，连接到一个Master后，会发送一个sync命令。

Master接到命令，启动后台的存盘进程，同时收集所有接收到的用于修改数据集命令，在后台执行完毕后，Master将传输整个数据文件到Salve，并完成一次完全同步。

• 全量复制：Slave服务在接收到数据库文件后，将其存盘并加载到内存中。
• 增量复制：Master继续将所有新的收集到的修改命令依次传递给Slave，完成同步。

只要重新连接Master，就自动执行一次完全同步。

手动配置俩种模型：

主机断开，使用slaveof no one，自己变为主机。

13.6、哨兵模式

哨兵模式是什么？

Redis哨兵模式是是一个管理多个 Redis 实例的工具，它可以实现对 Redis 的监控、通知、自动故障转移，是Redis实现高可用 的实现方案。

单哨兵模式：

在上面模式中，哨兵主要有几个作用：

• 监控状态：会向所有监控对象每秒发送ping命令，通过是否有响应来判断Master和所有Slave节点状态。
• 故障转移：当一旦发现Master节点异常，它将尝试进行故障转移，选择新的Slave节点为Master节点，并通过发布订阅的方式通知其他Slave节点修改配置。

但在上面的模式中，哨兵节点也存在单点故障。因此，为防止Sentinel发生意外，Sentinel也需要实现集群高可用。

上图就构成了多哨兵模式，实现了哨兵节点的高可用。

• Sentinel不只是监控Redis节点，各Sentinel节点之间也会互相监控

哨兵模式特点：

• 自动检查及故障转移：
• 主观下线和客观下线：
• 投票选举

13.6.1、自动检查及故障转移

当主节点宕机时，哨兵模式可以自动检测到宕机事件，并从从节点中选举出新的主节点，确保系统的持续可用性。

13.6.2、主观下线和客观下线

主观下线是指一个哨兵节点认为主节点不可用，但它并不确定其他哨兵节点是否也认为主节点不可用。当一个哨兵节点在一定时间(配置参数：down-after-milliseconds)内无法与主节点通信（比如发送PING命令没有收到响应），它会认为主节点下线。但在这个阶段，其他哨兵节点并不知道这个节点的状态，仅有一个哨兵主观地认为主节点宕机。

客观下线是指一个主节点被多数哨兵节点认定为不可用。当一个哨兵节点认为主节点宕机后，它会向其他哨兵节点询问对主节点的状态，并请求其他哨兵进行确认。如果多数（大多数至少需要半数加1）的哨兵节点都认为主节点不可用，那么主节点就会被判定为客观下线。客观下线意味着主节点的状态在整个哨兵集群中得到了确认。

主客观一致才能下线Master。

主客观是为了避免误判。只有少部分认为主节点下线可能是网络波动等原因造成，此时不应进行故障转移。只有多数哨兵节点确认主节点下线，才能确保1故障转移的正确性。

13.6.3、投票选举

在多Sentinel模式下，各节点会相互监控主从节点的健康状态。当主节点发生故障时，首先由Sentinel节点之间基于Raft算法进行投票选举，按照谁发现主节点故障谁去处理的原则，选举出一个领头Sentinel节点（Leader Sentinel）。这个领头Sentinel节点负责进行故障转移操作。

什么是Raft算法？

Raft是一种共识算法，只有在大多数(多于半数， ⌊n/2⌋+1 )服务器同意某一个命令并保存了该命令之后，这个命令才会被提交，然后服务器再对客户端进行回应，这样做是为了保证当主服务器宕机时，能够找出一台和原来主服务器具有相同状态机的替代服务器。

为了实现上述的“大多数”原则，Raft给服务器设置了三种状态，分别是领导者(leader)、跟随者(follower)和候选者(candidate)。跟随者通过投票选出领导者，只有得到“大多数”跟随者投票的服务器能成为领导者；领导者负责将命令同步给跟随者，只有被“大多数”跟随者确认的命令才能提交。

• sentinel monitor <master-name> <ip> <redis-port> <quorum>：让 sentinel 监控地址为 ip:port 的Master，master-name 可以自定义； 表示当有多少个 sentinel 认为主服务器宕机时，它才算真正的宕机掉，通常数量为半数或半数以上数量设置。
• sentinel down-after-milliseconds <master-name> <milliseconds>：在指定的毫秒数内，若主节点没有应答哨兵的 PING 命令，此时哨兵认为服务器主观下线，默认时间为 30 秒。
• parallel-syncs <master-name> <number> ：指定可以有多少个 Redis 服务同步新的主机，一般而言，这个数字越小同步时间越长，而越大，则对网络资源要求就越高。
• failover-timeout <master-name> <milliseconds>：指定故障转移允许的毫秒数，若超过这个时间，就认为故障转移执行失败，默认为 3 分钟。

启动命令：

./redis-sentinel sentinel.conf

14、Redis缓存穿透和雪崩

服务器高可用问题。

14.1、缓存穿透

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

解决方法有俩种：

1. 布隆过滤器
2. 缓存空对象

14.1.1、布隆过滤器

原理：利用一个很长的二进制数组，通过一系列的hash函数来确定该数据是否存在。

减少误判的方式：

1. 增加二进制数组位数
2. 增加Hash次数

布隆过滤器是1970年由布隆提出的。它实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都比一般的算法要好的多，缺点是有一定的误识别率和删除困难。

14.1.2、缓存空对象

持久层不命中时，仍然将空对象保留到缓存层中，之后再访问这个数据将会从缓存中获取，这样就保护了后端数据源。

这是一种空间换时间的算法，如果这是一次攻击，问题将会很严重，近期多次的缓存穿透，内存消耗严重。

14.2、缓存击穿

缓存击穿（量太大，缓存过期）

缓存击穿, 是指一个key非常热点, 在不停的扛着大并发, 大并发集中对着一个点进行访问, 当这个key在失效的瞬间, 持续的大并发就穿破缓存, 直接请求数据库, 就像在墙壁上凿开了一个洞。

解决方案：

1. 设置热点数据永不过期
2. 加互斥锁

14.2.1、设置热点数据永不过期

将热点数据设置为永不过期，从而避免缓存失效的问题。但是这种方法存在一个缺点，就是热点数据可能会被修改，如果不及时更新缓存，可能会导致缓存中的数据与实际数据不一致。

14.2.2、互斥加锁

保证缓存与数据库的一致性，但是如果缓存重建时间过长，性能会有极大影响，甚至有死锁的风险，牺牲了可用性。

当业务线程在处理用户请求时，如果发现访问的数据不在 Redis 里，就加个互斥锁，保证同一时间内只有一个请求来构建缓存（从数据库读取数据，再将数据更新到 Redis 里），当缓存构建完成后，再释放锁。未能获取互斥锁的请求，要么等待锁释放后重新读取缓存，要么就返回空值或者默认值。

14.3、缓存雪崩

缓存雪崩是指当缓存中的大量数据在同一时间失效，导致大量请求直接访问数据库，从而引起数据库的压力过大，严重影响系统的性能。

解决方案：

1. 限流降级
2. 数据预热
3. 缓存数据的随机过期时间

14.3.1、限流降级

使用分布式锁，从而避免大量请求同时访问数据库的情况。在SpringBoot中，我们可以通过Redisson来实现分布式锁的解决方案。

先在缓存中查询数据，如果不存在，则尝试获取分布式锁，如果获取到了锁，则查询数据库并将数据缓存到Redis中。如果没有获取到锁，则等待一段时间再尝试获取锁，这样即使大量请求同时访问系统，也能够保证只有一个请求去查询数据库并缓存数据，从而避免了缓存雪崩的问题。

14.3.2、数据预热

在系统启动时预热缓存，将系统中的热点数据提前加载到缓存中，从而避免了大量请求同时访问数据库的情况。在SpringBoot中，我们可以通过编写一个启动时执行的方法，来实现预热缓存的解决方案。

14.3.3、随机过期时间

在缓存数据的过期时间上增加随机因素，从而避免大量数据在同一时间失效的情况。在SpringBoot中，我们可以通过设置Redis缓存的过期时间和一个随机值来实现这个解决方案。