Redis-基础篇
Redis入门概述
1、是什么
Remote Dictionary Server(远程字典服务)是完全开源的，使用ANSIC语言编写遵守BSD协议，是一个高性能的Key-Value数据库提供了丰富的数据结构，例如String、Hash、 List、Set.SortedSet等等。数据是存在内存中的，同时Redis支持事务、持久化、LUA脚本、发布/订阅、缓存淘汰、流技术等多种功能特性提供了主从模式、Redis Sentinel和Redis Cluster集群架构方案
基于内存的KV键值对内存数据库
Redis官网
https://www.redis.com.cn/documentation.html中文说明
2、能干什么
主流功能和应用
1.分布式缓存，挡在mysql数据库之前的带刀护卫
与传统数据库关系(mysql)相比Redis是key-value数据库(NoSQL一种)，mysql是关系数据库。Redis数据操作主要在内存，而mysql主要存储在磁盘，Redis在某一些场景使用中要明显优于mysql，比如计数器、排行榜等方面。Redis通常用于一些特定场景，需要与Mysql一起配合使用，两者并不是相互替换和竞争关系,而是共用和配合使用
2.内存存储和持久化(RDB+AOF) redis支持异步将内存中的数据写到硬盘上，同时不影响继续服务
3.高可用架构搭配 -->单机-->主从-->哨兵--->集群(一台崩了多搞几台顶住)
4.缓存穿透、击穿、雪崩
5.分布式锁
6 队列
Reids提供list和set操作，这使得Redis能作为一个很好的消息队列平台来使用。我们常通过Reids的队列功能做购买限制。比如到节假日或者推广期间，进行一些活动，对用户购买行为进行限制，限制今天只能购买几次商品或者一段时间内只能购买一次。也比较适合适用。
7.排行版＋占体
在互联网应用中，有各种各样的排行榜，如电商网站的月度销量排行榜、社交APP的礼物排行榜、小程序的投票排行榜等等。Redis提供的zset数据类型能够快速实现这些复杂的排行榜。
比如小说网站对小说进行排名，根据排名，将排名靠前的小说推荐给用户。
整体功能概述
优势
性能极高- Redis能读的速度是110000次/秒,写的速度是81000次/秒
Redis数据类型丰富，不仅仅支持简单的key-value类型的数据，同时还提供list.set，zset，hash等数据结构的存储
Redis支持数据的持久化，可以将内存中的数据保持在磁盘中,重启的时候可以再次加载进行使用
Redis支持数据的备份，即master-slave模式的数据备份
小结
3、下载
进官网下载
Redis源码地址https://qithub.com/redis/redis
Redis在线测试：https://try.redis.io/
Redis命令参考：http://doc.redisfans.com/
redis-7.0.0.tar.gzRedis-7.0.0
历史版本https://download.redis.io/releases/
4、Redis迭代演化和Redis7新特性浅谈
https://github.com/redis/redis/releases新特性官方说明
Redis7.0 核心特性简介-阿里云开发者社区 (aliyun.com)阿里云关于Redis7.0核心特性简介，作为本文参照学习的资料。
2022年4-月正式发布的Redis 7.0是目前Redis 历史版本中变化最大的版本。首先，它有超过50个以上新增命令;其次，它有大量核心特性的新增和改进。
Redis Functions
Client-eviction
Multi-part AOF
ACL V2
新增命令
listpack替代ziplist
底层性能提升(和编码关系不大)
5、Redis安装配置
官网上推荐的是使用Linux系统去安装Redis，由于企业里面做Redis开发，99%都是Linux版的运用和安装，所以我们使用CentOS-64位的。
首先需要gcc编译环境，使用命令：rpm -qa | grep gcc看自己本机上有没有安装.。如果没有就使用：yum -y install gcc安装，使用gcc -v查看版本号，大于等于4.8.5就行。
安装步骤
新建目录/opt然后进入该目录使用命令下载软件包
wget https://download.redis.io/releases/redis-7.0.0.tar.gz
然后解压该压缩包
tar -zxf redis-7.0.0.tar.gz
进入解压后的目录执行make && make install源码安装
cd redis-7.0.0 make && make install
查看默认安装路径 ：/usr/local/bin
redis-benchmark: 性能测试工具，服务启动后运行该命令，看看自己本子性能如何；
redis-check-aof:修复有问题的AOF文件；
redis-check-dump:修复有问题的dump.rdb文件
redis-cli:客户端，操作入口
redis-sentinel: redis集群使用
redis-server: Redis服务器启动命令
将默认的redis.conf拷贝到自己定义好的路径下，比如 /myredis
mkdir /myredis cp /opt/redis-7.0.0/redis.conf /myredis vim /myredis/redis.conf
修改/myredis目录下redis.conf配置文件做初始化设置
启动服务
redis-server /myredis/redis.conf 启动服务并指定配置文件的路径 ps -ef |grep redis 查看redis服务的端口号：6379
连接服务
redis-cli -a 123456 -p 6379 6379端口号写不写都行，默认就是
再次查看端口使用情况,可以看到有一个server端有一个clin端
ps - ef| grep redis
ping一下会出现PONE就代表成功了，quit就退出
helloword
set k1 helloWorld get k1
关闭，quit只是退出
单实例关闭：redis-cli -a 1223456 shutdown，也可以直接在服务器操作命令行中直接shutdown
多实例关闭，指定端口关闭：redis-cli -p 6379 shutdown
6、Redis10大数据类型
这里说的数据类型是value的数据类型，key的类型都是字符串
http: //www.redis.cn/commands.html命令查询中文文档
命令不区分大小写，但key是区分大小写的
使用help @数据类型即可看到使用方法
Redis键key
keys * 查看当前所有的key
set k1 s1 set k2 s2 set k3 s3 keys *
exists key 判断某个key是否存在,如果返回结果为1则存在，为0不存在
EXISTS k1 (integer) 1 EXISTS k1 k2 (integer) 2 EXISTS k1 k2 k3 (integer) 3 EXISTS k1 k2 k32 (integer) 2type key 查看你的key是什么类型的
127.0.0.1:6379> type k1 string 127.0.0.1:6379> lpush list 1 2 3 4 (integer) 4 127.0.0.1:6379> type list list 127.0.0.1:6379>
del key 删除指定的key数据
127.0.0.1:6379> del k3 (integer) 1 127.0.0.1:6379> del k4 (integer) 0 127.0.0.1:6379>
unlink key 非阻塞删除，仅仅将keys从keyspace元数据中删除，真正的删除会在后续异步中进行。
127.0.0.1:6379> unlink k2 (integer) 1 127.0.0.1:6379>
ttl key 查看还有多少秒过期，-1表示永远不过，-2表示已经过期，默认的设置key的时候都是永不过时
127.0.0.1:6379> EXPIRE k1 10 (integer) 1 127.0.0.1:6379> ttl k1 (integer) 6 127.0.0.1:6379> ttl k1 (integer) 4 127.0.0.1:6379> ttl k1 (integer) 3 127.0.0.1:6379> ttl k1 (integer) 2 127.0.0.1:6379> ttl k1 (integer) 1 127.0.0.1:6379> ttl k1 (integer) -2 127.0.0.1:6379>
expire key 秒钟 为给定的key设置过期时间
move key dbindex[0-15] 将当前数据的key移动到给定的数据库db当中，redis一个实例默认的16个库，默认的使用0号库
127.0.0.1:6379> keys * 1) "list" 2) "k2" 127.0.0.1:6379> move list 2 (integer) 1 127.0.0.1:6379> keys * 1) "k2" 127.0.0.1:6379> select 2 OK 127.0.0.1:6379[2]> keys * 1) "list" 127.0.0.1:6379[2]>
select dbindex 切换数据库[0-15] 默认为0
验证它有16个库，0-15切换16就报错超出范围
127.0.0.1:6379> select 15 OK 127.0.0.1:6379[15]> select 16 (error) ERR DB index is out of range 127.0.0.1:6379[15]>
配置文件中写的也有
dbsize 查看当前数据库key的数量
127.0.0.1:6379> DBSIZE (integer) 2 127.0.0.1:6379> keys * 1) "k1" 2) "k2" 127.0.0.1:6379>
flushdb 清空当前库
127.0.0.1:6379> keys * 1) "k1" 2) "k2" 127.0.0.1:6379> FLUSHdb OK 127.0.0.1:6379> keys * (empty array) 127.0.0.1:6379> select 2 OK 127.0.0.1:6379[2]> keys * 1) "list" 127.0.0.1:6379[2]> select 0 OK 127.0.0.1:6379> FLUSHALL OK 127.0.0.1:6379> select 2 OK 127.0.0.1:6379[2]> ksys * (error) ERR unknown command 'ksys', with args beginning with: '*'
flushall 通杀全部库
10大数据类型
Strings（字符串）：最简单的类型，可以包含任何数据，例如数字、字符串或二进制数据。这是Redis中最基本的数据类型，也是其他数据类型的基础。
经典的就是单值单value
最常用：
set key value和get key各种选项
set key value [NX|XX] [GET] [EX seconds|PX milliseconds|EXAT unix-time-EX seconds:以秒为单位设置过期时间
127.0.0.1:6379> set k1 v1 ex 10 OK 127.0.0.1:6379> ttl k1 (integer) 2 127.0.0.1:6379> ttl k1 (integer) -2 127.0.0.1:6379>
PX milliseconds:以毫秒为单位设置过期时间
EXAT timestamp:设置以秒为单位的UNIX时间戳所对应的时间为过期时间
时间戳(Unix timestamp)转换工具 - 在线工具 (tool.lu)uinx时间戳在线生成
127.0.0.1:6379> set k1 v1 exat 1691307150 OK 127.0.0.1:6379> ttl k1 (integer) 71
PXAT milliseconds-timestamp:设置以毫秒为单位的UTNTX时间戳所对应的时间为过期时间
NX:键不存在的时候设置键值
XX:键存在的时候设置键值
可以先将老值取出来输出一下再将新值设置上
set k1 v1 get
KEEPTTL:保留设置前指定键的生存时间
原本设置的k1的生存时间再次给k1设置新值时就会被新的时间覆盖掉，但是设置新值时加上keepttl就会继续使用原来的时间
127.0.0.1:6379> set k1 v1 ex 60 OK 127.0.0.1:6379> ttl k1 (integer) 56 127.0.0.1:6379> ttl k1 (integer) 55 127.0.0.1:6379> set k1 v1keepttl keepttl OK 127.0.0.1:6379> ttl k1 (integer) 48 127.0.0.1:6379> ttl k1 (integer) 47 127.0.0.1:6379>
GET:返回指定键原本的值，若键不存在时返回nil
SET命令使用Ex、Px、NK参数，其效果等同于sETEx、PSETEX、SETNX命令。根据官方文档的描述未来版本中SETEX、PSETEx、SETNX命令可能会被淘汰。EXAT、PxAT以及GET为Redis 6.2新增的可选参数。
同时设置/获取多个键值
MSET key value [key value ....]
mset k1 v1 k2 v2 k3 v3 mget k1 k2 k3
MGET key [key ....]
mset/mget/msetnx
msetnx要全部执行成功才能过
获取指定区间范围内的值：getrange/ setrange 就相当于字符串截取和设置值
127.0.0.1:6379> set k1 abcd1234 OK 127.0.0.1:6379> GETRANGE k1 0 -1 "abcd1234" 127.0.0.1:6379> GETRANGE k1 1 -1 "bcd1234" 127.0.0.1:6379> GETRANGE k1 1 3 "bcd" 127.0.0.1:6379> SETRANGE k1 1 xyxy (integer) 8 127.0.0.1:6379> get k1 "axyxy234" 127.0.0.1:6379>
数值增减，一定要是数字才能加减
INCR key 递增数字
127.0.0.1:6379> set k1 100 OK 127.0.0.1:6379> get k1 "100" 127.0.0.1:6379> INCR k1 (integer) 101 127.0.0.1:6379> INCR k1 (integer) 102 127.0.0.1:6379> INCR k1 (integer) 103
增加指定的整数 INCRBY key increment
127.0.0.1:6379> INCRBY k1 10 (integer) 113 127.0.0.1:6379> INCRBY k1 10 (integer) 123 127.0.0.1:6379> INCRBY k1 10 (integer) 133 127.0.0.1:6379>
递减数值 DECR key
127.0.0.1:6379> decr k1 (integer) 132 127.0.0.1:6379> decr k1 (integer) 131 127.0.0.1:6379> decr k1 (integer) 130 127.0.0.1:6379> decr k1 (integer) 129
减少指定的整数 DECRBY key decrement
127.0.0.1:6379> decrby k1 10 (integer) 118 127.0.0.1:6379> decrby k1 10 (integer) 108 127.0.0.1:6379> decrby k1 10 (integer) 98 127.0.0.1:6379>
获取字符串长度和内容追加 STRLEN key APPEND key value
127.0.0.1:6379> set k1 1234 OK 127.0.0.1:6379> STRLEN k1 (integer) 4 127.0.0.1:6379> APPEND k1 aaaa (integer) 8 127.0.0.1:6379> get k1 "1234aaaa" 127.0.0.1:6379>
分布式锁
setnx key value
setex(set with expire)键秒值/setnx(set if not exist)
127.0.0.1:6379> set k1 v1 OK 127.0.0.1:6379> expire k1 10 (integer) 1 127.0.0.1:6379> ttl k1 (integer) 3 127.0.0.1:6379> setex k1 10 v1 OK 127.0.0.1:6379> ttl k1 (integer) 6 setnx k1 v111 (integer) 1 127.0.0.1:6379> get k1 "v111" 127.0.0.1:6379> setnx k1 v111 (integer) 0 127.0.0.1:6379> setnx k1 v11 (integer) 0 127.0.0.1:6379>
getset（先get再set）
Sets（集合）：字符串的无序集合，每个元素都是唯一的。你可以添加、删除元素，检查元素是否存在，或者获取集合的大小。left、right都可以插入添加;
SADD key member [member ...] 添加元素自动去重
SMEMBERS key 遍历集合中的所有元素
SISMEMBER key member 判断元素是否在集合中
127.0.0.1:6379> SADD set1 1 1 1 2 2 2 3 4 5 (integer) 5 127.0.0.1:6379> smembers set1 1) "1" 2) "2" 3) "3" 4) "4" 5) "5" 127.0.0.1:6379> sismember set1 d (integer) 0 127.0.0.1:6379> sismember set1 1 (integer) 1 127.0.0.1:6379> sismember set1 2 (integer) 1 127.0.0.1:6379>
SREM key member [member ...] 删除元素
127.0.0.1:6379> srem set1 s (integer) 0 127.0.0.1:6379> srem set1 1 (integer) 1 127.0.0.1:6379> smembers set1 1) "2" 2) "3" 3) "4" 4) "5"
scard 获取集合里面的元素个数
127.0.0.1:6379> scard set1 (integer) 4
SRANDMEMBER key[数字] 从集合中随机展现设置的数字个数元素，元素不删除
SPOS key[数字] 从集合中随机弹出一个元素，出一个删一个
127.0.0.1:6379> srandmember set1 1 1) "2" 127.0.0.1:6379> srandmember set1 2 1) "4" 2) "2" 127.0.0.1:6379> srandmember set1 3 1) "1" 2) "7" 3) "9" 127.0.0.1:6379> smembers set1 1) "1" 2) "2" 3) "3" 4) "4" 5) "5" 6) "6" 7) "7" 8) "8" 9) "9" 127.0.0.1:6379> spop set1 1 1) "1" 127.0.0.1:6379> spop set1 2 1) "2" 2) "3"
smove key1 key2在key1里已存在的某个值 将key 1里已存在的某个值赋给key2
127.0.0.1:6379> sadd set2 a b c (integer) 3 127.0.0.1:6379> smebers set1 (error) ERR unknown command 'smebers', with args beginning with: 'set1' 127.0.0.1:6379> smembers set1 1) "4" 2) "5" 3) "6" 4) "7" 5) "8" 6) "9" 127.0.0.1:6379> smove set1 set2 4 (integer) 1 127.0.0.1:6379> smembers set1 1) "5" 2) "6" 3) "7" 4) "8" 5) "9" 127.0.0.1:6379> smembers set2 1) "c" 2) "b" 3) "a" 4) "4"
集合运算
A,B A----->abc12 B----->123ax
集合的差集运算A-B SDIFF key [key ...]
127.0.0.1:6379> sadd A a b c 1 2 (integer) 5 127.0.0.1:6379> sadd B 1 2 3 a x (integer) 5 127.0.0.1:6379> SDIFF A B 1) "b" 2) "c" 127.0.0.1:6379> SDIFF B A 1) "x" 2) "3"集合的并集运算AUB SUNION key [key ...]
127.0.0.1:6379> SUNION A B 1) "x" 2) "b" 3) "a" 4) "3" 5) "c" 6) "1" 7) "2"
集合的交集运算A∩B
SINTER key [key ..]
SINTERCARD numkeys key [key ...] [LIMIT limit] (Redis7.0新出的)它不返回结果集,而只返回结果的基数（去除重复之后的剩余个数）。返回由所有给定集合的交集产生的集合的基数
127.0.0.1:6379> sinter A B 1) "a" 2) "1" 3) "2" 127.0.0.1:6379> sintercard 2 A B (integer) 3 127.0.0.1:6379> sintercard 2 A B (integer) 3 127.0.0.1:6379> sintercard 2 A B limit 1 (integer) 1 127.0.0.1:6379> sintercard 2 A B limit 2 (integer) 2 127.0.0.1:6379> sintercard 2 A B limit 3 (integer) 3 127.0.0.1:6379> sintercard 2 A B limit 4 (integer) 3
Lists（列表）：字符串的集合，按插入顺序排序。你可以在列表的头部或尾部添加元素，也可以获取列表的一个片段，最多可以包含2^32- 1个元素(4294967295,每个列表超过40亿个元素)。
一个双端链表的结构，主要功能有push/pop等。一般用在粮、队列、消息队列等场景。
如果键不存在，创建新的链表;如果键已存在，新增内容;
如果值全移除，对应的键也就消失了。
它的底层实际是个双向链表，对两端的操作性能很高，通过索引下标的操作中间的节点性能会较差。
案例
lpush/rpush/lrange 就是先进后出
127.0.0.1:6379> lpush list1 1 2 3 4 5 (integer) 5 127.0.0.1:6379> rpush list2 11 22 33 44 55 (integer) 5 127.0.0.1:6379> type list1 list 127.0.0.1:6379> type list2 list 127.0.0.1:6379> lrange list1 0 -1 1) "5" 2) "4" 3) "3" 4) "2" 5) "1" 127.0.0.1:6379> lrange list2 0 -1 1) "11" 2) "22" 3) "33" 4) "44" 5) "55" 127.0.0.1:6379>
lpop/rpop 弹出
掐头去尾
127.0.0.1:6379> lpop list1 "5" 127.0.0.1:6379> lrange list1 0 -1 1) "4" 2) "3" 3) "2" 4) "1" 127.0.0.1:6379> rpop list1 "1" 127.0.0.1:6379> lrange list1 0 -1 1) "4" 2) "3" 3) "2" 127.0.0.1:6379>
lindex，按照索引下标获得元素(从上到下)
127.0.0.1:6379> lindex list1 0 "4" 127.0.0.1:6379> lindex list1 1 "3" 127.0.0.1:6379> lindex list2 0 "11" 127.0.0.1:6379>
llen 获取列表中元素的个数 list.size()
127.0.0.1:6379> llen list1 (integer) 3 127.0.0.1:6379> llen list2 (integer) 5
lrem key 数字N给定值v1解释(删除N个值等于v1的元素)
从left往right删除2个值等于v1的元素，返回的值为实际删除的数量LREM list3 0值，表示删除全部给定的值。零个就是全部值。
127.0.0.1:6379> lpush list3 v1 v1 v1 v2 v3 v3 v3 v4 v5 (integer) 9 127.0.0.1:6379> lrange list3 0 -1 1) "v5" 2) "v4" 3) "v3" 4) "v3" 5) "v3" 6) "v2" 7) "v1" 8) "v1" 9) "v1" 127.0.0.1:6379> lrem list3 2 v1 (integer) 2 127.0.0.1:6379> lrem list3 3 v3 (integer) 3 127.0.0.1:6379> lrange list3 0 -1 1) "v5" 2) "v4" 3) "v2" 4) "v1"
ltrim key开始index结束index，截取指定范围的值后再赋值给key，只要那一截其他的不要了
127.0.0.1:6379> lrange list1 0 -1 1) "9" 2) "7" 3) "6" 4) "5" 5) "4" 6) "3" 7) "2" 8) "1" 9) "4" 10) "3" 11) "2" 127.0.0.1:6379> ltrim list1 2 4 OK 127.0.0.1:6379> lrange list1 0 -1 1) "6" 2) "5" 3) "4" 127.0.0.1:6379>
rpoplpush 源列表目的列表 移除列表的最后一个元素，并将该元素添加到另一个列表并返回
127.0.0.1:6379> lrange list1 0 -1 1) "6" 2) "5" 3) "4" 127.0.0.1:6379> lrange list2 0 -1 1) "11" 2) "22" 3) "33" 4) "44" 5) "55" 127.0.0.1:6379> rpoplpush list1 list2 "4" 127.0.0.1:6379> lrange list1 0 -1 1) "6" 2) "5" 127.0.0.1:6379> lrange list2 0 -1 1) "4" 2) "11" 3) "22" 4) "33" 5) "44" 6) "55"
lset key index value 根据下标设置值
127.0.0.1:6379> lset list1 1 hello OK 127.0.0.1:6379> lrange list1 0 -1 1) "6" 2) "hello" 127.0.0.1:6379>
linsert key before/after 已有值插入的新值
在list1表中的hello后面插入java
127.0.0.1:6379> linsert list1 after hello java (integer) 3 127.0.0.1:6379> lrange list1 0 -1 1) "6" 2) "hello" 3) "java"
Sorted Sets（有序集合zset）：字符串的集合，每个元素都关联着一个浮点数分数，用于排序。你可以添加、删除元素，获取元素的排名，或者获取分数在某个范围内的元素。与Sets不同的是每个元素都会关联一个double类型的分数，redis正是通过分数来为集合中的成员进行从小到大的排序。zset的成员是唯一的,但分数(score)却可以重复。zset集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是O(1)。
在set基础上，每个val值前加一个score分数值。
之前set是k1 v1 v2 v3，现在zset是k1 score1 v1 score2 v2
ZADD key score member [score member ...] 添加元素向有序集合中加入一个元素和该元素的分数
ZRANGE key start stop [WITHSCORES] 按照元素分数从小到大的顺序返回索引从start到stop之间的所有元素
zrevrange上面一个从小到大的反转，即从大到小排
127.0.0.1:6379> zadd zset1 60 v1 70 v2 80 v3 90 v4 100 v5 (integer) 5 127.0.0.1:6379> zrange zset1 0 -1 1) "v1" 2) "v2" 3) "v3" 4) "v4" 5) "v5" 127.0.0.1:6379> zrange zset1 0 -1 withscore (error) ERR syntax error 127.0.0.1:6379> zrange zset1 0 -1 withscores 1) "v1" 2) "60" 3) "v2" 4) "70" 5) "v3" 6) "80" 7) "v4" 8) "90" 9) "v5" 10) "100" 127.0.0.1:6379> zrange zset1 0 2 withscores 1) "v1" 2) "60" 3) "v2" 4) "70" 5) "v3" 6) "80" 127.0.0.1:6379> zrevrange zset1 0 2 withscores 1) "v5" 2) "100" 3) "v4" 4) "90" 5) "v3" 6) "80"
ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count] 获取指定分数范围的元素，
(的意思是不包含 相当于大于但没有等于 limit的作用是返回限制像是分页从60到90的值
127.0.0.1:6379> zrangebyscore zset1 60 90 1) "v1" 2) "v2" 3) "v3" 4) "v4" 127.0.0.1:6379> zrangebyscore zset1 60 90 withscores 1) "v1" 2) "60" 3) "v2" 4) "70" 5) "v3" 6) "80" 7) "v4" 8) "90" 127.0.0.1:6379> zrangebyscore zset1 (60 90 withscores 1) "v2" 2) "70" 3) "v3" 4) "80" 5) "v4" 6) "90" 127.0.0.1:6379> zrangebyscore zset1 (60 90 withscores limit 0 1 1) "v2" 2) "70" 127.0.0.1:6379> zrangebyscore zset1 (60 90 withscores limit 0 2 1) "v2" 2) "70" 3) "v3" 4) "80"
ZSCORE key member 获取元素的分数
ZCARD key 获取集合中元素的数量
zrem key某score下对应的value值，作用是删除元素
ZINCRBY key increment member 增加某个元素的分数
127.0.0.1:6379> zscore zset1 v1 "60" 127.0.0.1:6379> zscore zset1 v2 "70" 127.0.0.1:6379> zcard zset1 (integer) 5 127.0.0.1:6379> zrem zset1 v5 (integer) 1 127.0.0.1:6379> zrem zset1 v5 (integer) 0 127.0.0.1:6379> zrange zset1 (error) ERR wrong number of arguments for 'zrange' command 127.0.0.1:6379> zrange zset1 0 -1 withscores 1) "v1" 2) "60" 3) "v2" 4) "70" 5) "v3" 6) "80" 7) "v4" 8) "90" 127.0.0.1:6379> ZINCRBY zset1 3 v1 "63" 127.0.0.1:6379> zcount zset1 60 80 (integer) 3 127.0.0.1:6379> zcount zset1 60 70 (integer) 2
ZMPOP 从键名列表中的第一个非空排序集中弹出一个或多个元素，它们是成员分数对
把1个键zset1的最小的弹出去1个
127.0.0.1:6379> zmpop 1 zset1 min count 1 1) "zset1" 2) 1) 1) "v1" 2) "63"
zrank key values值，作用是获得下标值
zrevrank key values值，作用是逆序获得下标值
127.0.0.1:6379> zrange zset1 0 -1 1) "v2" 2) "v3" 3) "v4" 127.0.0.1:6379> zrank zset1 v2 (integer) 0 127.0.0.1:6379> zrevrank zset1 v2 (integer) 2
Hashes（哈希）：键值对的集合。它是字符串到字符串的映射，可以用来存储对象，Redis中每个hash可以存储2^32-1键值对（40多亿)。KV模式不变，但V是一个键值对。
Map<String, Map<0bject,object>>案例
hset/hget/ hmset/ hmget/ hgetall/hdel
127.0.0.1:6379> hset user:001 id 1 name z3 age 21 (integer) 3 127.0.0.1:6379> hget user:001 id "1" 127.0.0.1:6379> hget user:001 * (nil) 127.0.0.1:6379> hmset user:02 id 2 name lisi age 99 OK 127.0.0.1:6379> hmget user:02 id name age 1) "2" 2) "lisi" 3) "99" 127.0.0.1:6379> hgetall user:02 1) "id" 2) "2" 3) "name" 4) "lisi" 5) "age" 6) "99" 127.0.0.1:6379> hdel user:02 age (integer) 1 127.0.0.1:6379> hgetall user:02 1) "id" 2) "2" 3) "name" 4) "lisi" 127.0.0.1:6379>
hlen 获取某个key内的全部数量
127.0.0.1:6379> hlen user:02 (integer) 2 127.0.0.1:6379>
hexists key 在key里面的某个值的key存不存在
127.0.0.1:6379> hexists user:02 id (integer) 1 127.0.0.1:6379> hexists user:02 hhh (integer) 0
hkeys/hvals 单独列出key或者单独列出value
127.0.0.1:6379> hkeys user:02 1) "id" 2) "name" 127.0.0.1:6379> hvals user:02 1) "2" 2) "lisi"
hincrby/ hincrbyfloat 整数加指定值和小数加指定值
127.0.0.1:6379> hset user3 id 12 age 87 score 99 (integer) 3 127.0.0.1:6379> hgetall user3 1) "id" 2) "12" 3) "age" 4) "87" 5) "score" 6) "99" 127.0.0.1:6379> hincrby user3 age 1 (integer) 88 127.0.0.1:6379> hincrby user3 age 1 (integer) 89 127.0.0.1:6379> hincrby user3 age 1 (integer) 90 127.0.0.1:6379> hincrby user3 age 3 (integer) 93 127.0.0.1:6379> hset user3 id 12 age 87 score 99.5 (integer) 0 127.0.0.1:6379> hgetall user3 1) "id" 2) "12" 3) "age" 4) "87" 5) "score" 6) "99.5" 127.0.0.1:6379> hincrbyfloat user3 score 0.5 "100" 127.0.0.1:6379> hincrbyfloat user3 score 0.5 "100.5" 127.0.0.1:6379> hgetall user3 1) "id" 2) "12" 3) "age" 4) "87" 5) "score" 6) "100.5"
hsetnx 不存在赋值，存在了无效。
127.0.0.1:6379> hsetnx user3 email abcd@163.com (integer) 1 127.0.0.1:6379> hsetnx user3 email abcd@163.com (integer) 0 127.0.0.1:6379>
Bitmaps（位图）：使用字符串作为位数组。你可以设置、清除单个位，或者统计设置为1的位的数量和状态。由0和1状态表现的二进制位的bit数组。说明:用String类型作为底层数据结构实现的一种统计二值状态的数据类型。
位图本质是数组，它是基于String数据类型的按位的操作。该数组由多个二进制位组成，每个二进制位都对应一个偏移量(我们称之为一个索引)。
命令
setbit key offset val 给指定key的值的第offset赋值val,setbit健偏移位只能零或者1，Bitmap的偏移量是从零开始算的
setbit k1 1 1 保存到redis的数据就是这样
getbit key offset 获取指定key的第offet位
127.0.0.1:6379> SETBIT k1 1 1 (integer) 0 127.0.0.1:6379> SETBIT k1 2 0 (integer) 0 127.0.0.1:6379> SETBIT k1 3 1 (integer) 0 127.0.0.1:6379> type k1 string 127.0.0.1:6379> getbit k1 1 (integer) 1 127.0.0.1:6379> getbit k1 2 (integer) 0 127.0.0.1:6379> getbit k1 3 (integer) 1 127.0.0.1:6379> getbit k1 34 (integer) 0
bitcount key start end 返回指定key中[start,end]中为1的数量
127.0.0.1:6379> setbit uid:login123 1 1 (integer) 0 127.0.0.1:6379> setbit uid:login123 2 1 (integer) 0 127.0.0.1:6379> setbit uid:login123 3 0 (integer) 0 127.0.0.1:6379> bitcount uid:login123 (integer) 2
strlen ，统计字节占用多少，不是字符串长度而是占据几个字节，超过8位后自己按照8位一组一byte再扩容
127.0.0.1:6379> setbit k2 0 1 (integer) 0 127.0.0.1:6379> setbit k2 7 1 (integer) 0 127.0.0.1:6379> strlen k2 (integer) 1 127.0.0.1:6379> setbit k2 8 1 (integer) 0 127.0.0.1:6379> strlen k2 (integer) 2
bitop operation destkey key 对不同的二进制存储数据进行位运算(AND、OR、NOT.XOR)
连续2天都签到的用户，加入某个网站或者系统，它的用户有100ow，做个用户id和位置的映射
比如0号位对应用户id: uid-092iok-lkj
比如1号位对应用户id: uid-7388c-xxx
0和1分别代表一个用户，20230801这一天0、1和2号 都打卡签到了
20230802这一天0和2号来了，1没来，然后使用bitop and将第一天和第二天都来的用户做交集会有两个用户0和2
127.0.0.1:6379> hset uidmap 0 uid-092iok-lkj 1 uid-7388c-xxx (integer) 2 127.0.0.1:6379> hgetall uidmap 1) "0" 2) "uid-092iok-lkj" 3) "1" 4) "uid-7388c-xxx" 127.0.0.1:6379> setbit 20230801 0 1 (integer) 0 127.0.0.1:6379> setbit 20230801 1 1 (integer) 0 127.0.0.1:6379> getbit 20230801 0 (integer) 1 127.0.0.1:6379> getbit 20230801 1 (integer) 1 127.0.0.1:6379> setbit 20230801 2 1 (integer) 0 127.0.0.1:6379> setbit 20230801 3 1 (integer) 0 127.0.0.1:6379> setbit 20230802 0 1 (integer) 0 127.0.0.1:6379> setbit 20230802 2 1 (integer) 0 127.0.0.1:6379> bitcount 20230801 (integer) 4 127.0.0.1:6379> bitcount 20230802 (integer) 2 127.0.0.1:6379> bitop and k3 20230801 20230802 (integer) 1 127.0.0.1:6379> bitcount k3 (integer) 2
HyperLogLogs（超日志，基数统计，属于string类型）：用于估计集合的基数（即不同元素的数量）。虽然它的精度有限，但是无论集合的大小，HyperLogLog的空间复杂度都是常数。在Redis里面，每个 HyperlogLog键只需要花费12KB内存，就可以计算接近2^64个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。
但是，因为HyperLogLog只会根据输入元素来计算基数，而不会储存输入元素本身，所以HyperLogLog不能像集合那样，返回输入的各个元素.
UV：Unique Visitor，独立访客，一般理解为客户端IP，需要考虑到去重
统计某个网站的UV、统计某个文章的UV，用户搜索网站关键词的数量，统计用户每天搜索不同词条个数等
基数：是一种数据集，去重后的真实个数
基数统计：用于统计一个集合中不重复的元素个数，就是对集合去重复后剩余元素的计算
基本命令：
PFADD key element [element ..] 添加指定元素到HyperLogLog中。
PFCOUNT key [key...] 返回给定HyperLogLog的基数估算值。
PFMERGE destkey sourqekey_[sourcekey ...] 将多个HyperLogLog合并为一个HyperLogLog
127.0.0.1:6379> pfadd hello1 1 3 5 7 9 (integer) 1 127.0.0.1:6379> pfadd hello2 1 2 4 4 4 5 9 10 (integer) 1 127.0.0.1:6379> PFCOUNT hello1 (integer) 5 127.0.0.1:6379> PFCOUNT hello2 (integer) 6 127.0.0.1:6379> pfmerge hires hello1 hello2 OK 127.0.0.1:6379> PFCOUNT hires (integer) 8
Streams（流）：用于存储一系列的键值对。它类似于日志文件，可以用来记录事件或消息。Redis Stream主要用于消息队列(MQ，Message Queue), Redis本身是有一个Redis 发布订阅(pub/sub)来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis宕机等，消息就会被丢弃。简单来说发布订阅(pub/sub)可以分发消息，但无法记录历史消息。而Redis Stream提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失
队列的思想，先进先出
127.0.0.1:6379> lrange list1 0 -1 5 4 3 2 1 6 hello java 127.0.0.1:6379> rpop list1 1 java 127.0.0.1:6379> rpop list1 1 hello 127.0.0.1:6379> rpop list1 2 6 1 127.0.0.1:6379> rpop list1 3 2 3 4 127.0.0.1:6379> rpop list1 4 5
Pub/Sub
steam的底层结构
Message Content
消息内容
Consumer group
消费组，通过XGROUP CREATE命令创建，同一个消费组可以有多个消费者
Last_delivered_id
游标，每个消费组会有个游标last_delivered_id，任意一个消费者读取了消息都会使游标last_delivered_id往前移动。
Consumer
消费者，消费组中的消费者
Pending_ids
消费者会有一个状态变量，用于记录被当前消费已读取但未ack的消息Id，如果客户端没有ack，这个变量里面的消息ID会越来越多，一旦某个消息被ack它就开始减少。这个pending_ids变量在Redis官方被称之为PEL (Pending Entries List)，记录了当前已经被客户端读取的消息，但是还没有ack (Acknowledgecharacter:确认字符），它用来确保客户端至少消费了消息-T次，而不会在网络传输的中途丢失了没处理
队列相关指令
XADD 添加消息到队列末尾
消息ID必须要比上个ID大 默认用星号表示自动生成规矩
127.0.0.1:6379> xadd mystream * id 11 cname z3 1691664748604-0 127.0.0.1:6379> xadd mystream * k1 v1 k2 v2 k3 v3 1691664772100-0
信息条目指的是序列号，在相同的毫秒下序列号从O开始递增，序列号是64位长度，理论上在同一毫秒内生成的数据量无法到达这个级别，因此不用担心序列号会不够用。millisecondsTime指的是Redis节点服务器的本地时间，如果存在当前的毫秒时间戳比以前已经存在的数据的时间戳小的话〈本地时间钟后跳〉，那么系统将会采用以前相同的毫秒创建新的ID，也即redis 在增加信息条目时会检查当前id 与上一条目的id，自动纠正错误的情况，一定要保证后面的 id 比前面大，一个流中信息条目的ID必须是单调增的，这是流的基础。
客户端显示传入规则:
Redis对于ID有强制要求，格式必须是时间戳-自增Id这样的方式，且后续ID不能小于前一个ID
Stream的消息内容，也就是图中的Message Content它的结构类似Hash结构，以key-value的形式存在。
XTRIM 限制Stream的长度，如果已经超长会进行截取。
MAXLEN，允许的最大长度，对流进行修剪限制长度
127.0.0.1:6379> xadd mystream * k1 v1 k2 v2 1691713579063-0 127.0.0.1:6379> xadd mystream * k3 v3 k4 v4 1691713592176-0 127.0.0.1:6379> xrange mystream - + 1691664748604-0 id 11 cname z3 1691713579063-0 k1 v1 k2 v2 1691713592176-0 k3 v3 k4 v4 127.0.0.1:6379> xtrim mystream maxlen 2 1 127.0.0.1:6379> xrange mystream - + 1691713579063-0 k1 v1 k2 v2 1691713592176-0 k3 v3 k4 v4截取的是时间最新的两条信息，也就是时间戳大的
MINID 允许的最小id，从某个id值开始比该id值小的将会被抛弃
127.0.0.1:6379> xrange mystream - + 1691713579063-0 k1 v1 k2 v2 1691713592176-0 k3 v3 k4 v4 127.0.0.1:6379> xtrim mystream minid 1691713579063-0 0 127.0.0.1:6379> xtrim mystream minid 1691713592176-0 1 127.0.0.1:6379> xrange mystream - + 1691713592176-0 k3 v3 k4 v4 127.0.0.1:6379>
XDEL 删除消息
XLEN 获取Stream中的消息长度
127.0.0.1:6379> xrange mystream - + 1691664748604-0 id 11 cname z3 1691664772100-0 k1 v1 k2 v2 k3 v3 127.0.0.1:6379> XDEL MYSTREAM 1691664772100-0 0 127.0.0.1:6379> XDEL mystream 1691664772100-0 1 127.0.0.1:6379> xrange mystream - + 1691664748604-0 id 11 cname z3 127.0.0.1:6379> xlen mysteam 0 127.0.0.1:6379> xlen mystream 1
XRANGE 获取消息列表(可以指定范围)，忽略删除的消息
XREVRANGE 和XRANGE相比区别在于反向获取，ID从大到小
127.0.0.1:6379> xrange mystream - + 1691664748604-0 id 11 cname z3 1691664772100-0 k1 v1 k2 v2 k3 v3 127.0.0.1:6379> xrange mystream - + count 1 1691664748604-0 id 11 cname z3 127.0.0.1:6379> xrevrange mystream + - 1691664772100-0 k1 v1 k2 v2 k3 v3 1691664748604-0 id 11 cname z3XREAD 获取消息(阻塞/非阻塞)，返回大于指定ID的消息
非阻塞
$代表特殊ID，表示以当前Stream已经存储的最大的ID作为最后一个ID，当前Stream中不存在大于当前最大ID的消息，因此此时返回nil.
0-0代表从最小的ID开始获取Stream中的消息，当不指定count，将会返回Stream中的所有消息，注意也可以使用0（0/000也都是可以的......)
127.0.0.1:6379> xread count 2 streams mystream $ 127.0.0.1:6379> xread count 2 streams mystream 0 mystream 1691713592176-0 k3 v3 k4 v4 1691715367061-0 k6 v6 k7 v7
阻塞
开两个客户端连接，一个生产者一个消费者等待消费
127.0.0.1:6379> 消费者 ERR unknown command '消费者', with args beginning with: 127.0.0.1:6379> xread count 1 block 0 streams mystream $ mystream 1691716180552-0 k11 v11[root@localhost ~]# redis-cli -a 123456 Warning: Using a password with '-a' or '-u' option on the command line interface may not be safe. 127.0.0.1:6379> xadd mystream * k11 v11 "1691716180552-0"
小结
Stream的基础方法，使用xadd存入消息和xread循环阻塞读取消息的方式可以实现简易版的消息队列
交互流程：
消费组相关指令
XGROUP CREATE 创建消费者组
XGROUP CREATE mystream groupA $ $表示从Stream尾部开始消费 XGROUP CREATE mystream groupB 0 0表示从Stream头部开始消费创建消费者组的时候必须指定ID,ID为О表示从头开始消费，为$表示只消费新的消息，队尾新来
XREADGROUP GROUP
“>”，表示从第一条尚未被消费的消息开始读取
stream中的消息—旦被消费组里的一个消费者读取了，就不能再被该消费组内的其他消费者读取了，即同一个消费组里的消费者不能消费同一条消息。刚才的XREADGROUP命令再执行—次，此时读到的就是空值.游标已经从头走到尾了。
但是，其他组还能继续读取，其他组的游标还没读取呢。
127.0.0.1:6379> xgroup create mystream groupA $ OK 127.0.0.1:6379> xgroup create mystream groupB 0 OK 127.0.0.1:6379> xgroup create mystream groupC 0 OK 127.0.0.1:6379> xreadgroup group groupA consumer1 streams mystream > (nil) 127.0.0.1:6379> xreadgroup group groupB consumer1 streams mystream > 1) 1) "mystream" 2) 1) 1) "1691713592176-0" 2) 1) "k3" 2) "v3" 3) "k4" 4) "v4" 2) 1) "1691715367061-0" 2) 1) "k6" 2) "v6" 3) "k7" 4) "v7" 3) 1) "1691715380635-0" 2) 1) "k9" 2) "v9" 3) "k8" 4) "v8" 4) 1) "1691715483732-0" 2) 1) "k10" 2) "v10" 3) "k10" 4) "v10" 5) 1) "1691715499815-0" 2) 1) "k12" 2) "v12" 3) "k11" 4) "v11" 6) 1) "1691716180552-0" 2) 1) "k11" 2) "v11" 127.0.0.1:6379> xreadgroup group groupB consumer2 streams mystream > (nil)
消费组groupA内的消费者consumer1从mystream消息队列中读取所有消息
但是，不同消费组的消费者可以消费同一条消息
消费组的目的，让组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的
127.0.0.1:6379> xgroup create mystream groupD 0 OK 127.0.0.1:6379> xreadgroup group groupD consumer1 count 1 streams mystream > 1) 1) "mystream" 2) 1) 1) "1691713592176-0" 2) 1) "k3" 2) "v3" 3) "k4" 4) "v4" 127.0.0.1:6379> xreadgroup group groupD consumer2 count 1 streams mystream > 1) 1) "mystream" 2) 1) 1) "1691715367061-0" 2) 1) "k6" 2) "v6" 3) "k7" 4) "v7" 127.0.0.1:6379> xreadgroup group groupD consumer3 count 3 streams mystream > 1) 1) "mystream" 2) 1) 1) "1691715380635-0" 2) 1) "k9" 2) "v9" 3) "k8" 4) "v8" 2) 1) "1691715483732-0" 2) 1) "k10" 2) "v10" 3) "k10" 4) "v10" 3) 1) "1691715499815-0" 2) 1) "k12" 2) "v12" 3) "k11" 4) "v11" 127.0.0.1:6379>
一个问题
基于Stream 实现的消息队列，如何保证消费者在发生故障或宕机再次重启后，仍然可以读取未处理完的消息?
Streams会自动使用内部队列〈也称为 PENDING List）留存消费组里每个消费者读取的消息保底措施，直到消费者使用XACK命令通知Streams"消息已经处理完成"。
消费确认增加了消息的可靠性，一般在业务处理完成之后，需要执行XACK命令确认消息已经被消费完成
XPENDING
查询每个消费组内所有消费者「已读取、但尚未确认」的消息
127.0.0.1:6379> xpending mystream groupD 1) (integer) 5 2) "1691713592176-0" 3) "1691715499815-0" 4) 1) 1) "consumer1" 2) "1" 2) 1) "consumer2" 2) "1" 3) 1) "consumer3" 2) "3"
查看某个消费者具体读取了哪些数据
127.0.0.1:6379> XPENDING mystream groupD - + 10 consumer3 1) 1) "1691715380635-0" 2) "consumer3" 3) (integer) 1079592 4) (integer) 1 2) 1) "1691715483732-0" 2) "consumer3" 3) (integer) 1079592 4) (integer) 1 3) 1) "1691715499815-0" 2) "consumer3" 3) (integer) 1079592 4) (integer) 1groupD组中的consumer3读取了这3条数据 已读未确认
XACK
127.0.0.1:6379> XACK mystream groupD 1691715380635-0 (integer) 1 127.0.0.1:6379> XPENDING mystream groupD - + 10 consumer3 1) 1) "1691715483732-0" 2) "consumer3" 3) (integer) 1223509 4) (integer) 1 2) 1) "1691715499815-0" 2) "consumer3" 3) (integer) 1223509 4) (integer) 1
XINFO 用于打印stream\Consumer\Group的详细信息
127.0.0.1:6379> xinfo stream mystream 1) "length" 2) (integer) 6 3) "radix-tree-keys" 4) (integer) 1 5) "radix-tree-nodes" 6) (integer) 2 7) "last-generated-id" 8) "1691716180552-0" 9) "max-deleted-entry-id" 10) "1691664772100-0" 11) "entries-added" 12) (integer) 9 13) "recorded-first-entry-id" 14) "1691713592176-0" 15) "groups" 16) (integer) 4 17) "first-entry" 18) 1) "1691713592176-0" 2) 1) "k3" 2) "v3" 3) "k4" 4) "v4" 19) "last-entry" 20) 1) "1691716180552-0" 2) 1) "k11" 2) "v11"
Geospatial（地理空间）：用于存储地理位置信息，例如经度和纬度，然后可以进行各种地理空间查询。
主要分为三步
将三维的地球变为二维的坐标，再将二维的坐标转换为一维的点块，最后将一维的点块转换为二进制再通过base32编码。
基本命令
GEOAPD 多个经度(longitude)、纬度(latitude)、位置名称(member)添加到指定的 key中
坐标位置名称是中文会乱码，先退出然后再登录时加上 --raw即可
GEOPOS 从键里面返回所有给定位置元素的位置（经度和纬度)
GEODIST返回两个给定位置之间的距离。
GEORADIUS以给定的经纬度为中心，返回与中心的距离不超过给定最大距离的所有位置元素。以半径为中心，查找附近的XXX
GEORADIUSBYMEMBER跟GEORADIUS类似
GEOHASH返回一个或多个位置元素的Geohash表示 geohash算法生成的base32编码值
Bitfields（位字段，位域）：使用字符串作为位数组，但是与Bitmaps不同，Bitfields可以操作多个位的字段。了解即可。
BITFIELD命令的作用在于它能够将很多小的整数储存到一个长度较大的位图中，又或者将一个非常庞大的键分割为多个较小的键来进行储存，从而非常高效地使用内存，使得 Redis 能够得到更多不同的应用——特别是在实时分析领域:BITFIELD 能够以指定的方式对计算溢出进行控制的能力，使得它可以被应用于这一领域。
能干的就是位域修改和溢出控制
将一个Redis字符串看作是一个由二进制位组成的数组，并能对变长位宽和任意没有字节对齐的指定整型位域进行寻址和修改
基本语法
GET <type><offset>一返回指定的位域
SET<type> <offset><value>-设置指定位域的值并返回它的原值
INCRBY〈type><offset><increment>一自增或自减(如果increment为负数)指定位域的值并返回它的新值
还有一个命令通过设置溢出行为来改变调用INCRBY指令的后序操作：OVERFLOW [WRAP|SAT|FAIL]
WRAN:使用回绕(wrap around）方法处理有符号整数和无符号整数的溢出情况
SAT:使用饱和计算(saturation arithmetic)方法处理溢出，下溢计算的结果为最小的整数值，而上溢计算的结果为最大的整数值
FAIL:命令将拒绝执行那些会导致上溢或者下溢情况出现的计算,并向用户返回空值表示计算未被执行
当需要一个整型时，有符号整型需在位数前加i，无符号在位数前加u。例如，u8是一个8位的无符号整型，i16是一个16位的有符号整型
BITFIELD key [GET type offset][SET type offset value] [INCRBY type offset increment] [OVERFLOW WRAPISATIFAIL]
前八位存h，值是104，字节码是01101000
127.0.0.1:6379> get fieldkey
"hello"
127.0.0.1:6379> BITFIELD fieldkey get i8 0
1) (integer) 104
127.0.0.1:6379> BITFIELD fieldkey get i8 8
1) (integer) 101
127.0.0.1:6379> BITFIELD fieldkey get i8 16
1) (integer) 108hello 等价于0110100001100101011011000110110001101111
127.0.0.1:6379> bitfield fieldkey incrby u4 2 1 #从第3个位开始,对接下来的4位无符号数+1
1) (integer) 11
127.0.0.1:6379> bitfield fieldkey incrby u4 2 1
1) (integer) 12
127.0.0.1:6379> bitfield fieldkey incrby u4 2 1
1) (integer) 13
127.0.0.1:6379> bitfield fieldkey incrby u4 2 1
1) (integer) 14
127.0.0.1:6379> bitfield fieldkey incrby u4 2 1
1) (integer) 15
127.0.0.1:6379> bitfield fieldkey incrby u4 2 1 #默认overflow为wrap.即循环溢出
1) (integer) 0
127.0.0.1:6379> get fieldkey
"@ello"
d
Redis持久化
Redis是内存数据库，宕机后数据会消失，Redis重启后快速恢复数据，要提供持久化机制。
Redis的两种持久化方式：RDB（Redis DataBase）和AOF（Append Only File）
1、RDB（Redis DataBase）
1.1、是什么
一句话：在指定的时间间隔，执行数据集的时间点快照
实现类似照片记录效果的方式，就是把某一时刻的数据和状态以文件的形式写到磁盘上，也就是快照。这样一来即使故障宕机，快照文件也不会丢失，数据的可靠性也就得到了保证。
这个快照文件就称为RDB文件(dump.rdb)，其中，RDB就是Redis DataBase的缩写。
1.2 、能干啥
在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是行话讲的Snapshot内存快照,它恢复时再将硬盘快照文件直接读回到内存里。
Redis的数据都在内存中，保存备份时它执行的是全量快照,也就是说，把内存中的所有数据都记录到磁盘中，一锅端。
Rdb保存的是dump.rdb文件。
1.3、 配置文件
1.4、自动触发
Redis7版本，按照redis.conf里配置的save <seconds> <changes>
本次案例5秒2次修改，redis配置文件中434行修改配置文件
修改dump文件保存路径，配置文件505行修改自定义路径dir /myredis/dumpfiles
修改dump文件名称，配置文件481行修改名字加上6379端口号，在以后多机中好识别。修改了配置文件就重新启动redis服务，先SHUTDOWN停止服务，然后exit退出然后再重新启动。
查看我们修改的配置文件有没有起作用
使用config get可以获取配置文件中的内容
触发备份，先设置两个k，然后再查看/myredis/dumpfiles下有没有备份文件
现在它的大小是107，这次再set一个k3然后等5秒不进行任何操作，它还是107大，说明没有进行再次备份。当我再设置一个k4它就会触发备份，此时就变成121大了，5秒2次修改
如何恢复
将备份文件(dump.rdb）移动到redis安装目录并启动服务即可，备份成功后故意用flushdb清空redis，看看是否可以恢复数据。
我们将dump6379.rdb换个名字成bak，然后清空k，flushdb，然后再看备份目录发现又生成了一个dump6379.rdb，redis要保证数据的一致性就会把数据的终结命令也备份下来。
停止服务后再重新进就没有之前的key了
所以，执行flushall/flushdb命令也会产生dump.rdb文件，但里面是空的，无意义。
注意：当执行shutdown时它也会记录备份，保证redis服务停机之前有一个最新的备份。
给原来有数据的备份换回来
然后再打开服务查看keys就有了
注意: 不可以把备份文件dump.rdb和生产redis服务器放在同一台机器，必须分开各自存储，以防生产机物理损坏后备份文件也挂了。
1.5 、手动触发
Redis提供了两个命令来生成RDB文件，分别是save和bgsave
save
在主程序中执行会阻塞当前redis服务器，直到持久化工作完成执行save命令期间,Redis不能处理其他命令，线上禁止使用。所以，生产环境不能使用这个命令。少量数据体现不出来，当有大量数据时，保存一次备份可能需要很长时间，这段时间的redis就被阻塞，什么都干不了了。
BGSAVE(默认) 生产环境中只允许使用BGSAVE不能使用save
Redis会在后台异步进行快照操作，不阻塞快照同时还可以响应客户端请求,该触发方式会fork一个子进程由子进程复制持久化过程。
Redis会使用bgsave对当前内存中的所有数据做快照，这个操作是子进程在后台完成的，这就允许主进程同时可以修改数据。
Fork
在Linux程序中，fork ()会产生一个和父进程完全相同的子进程，但子进程在此后多会exec系统调用，出于效率考虑，尽量避免膨胀。cop一份
新添加一个k5，备份文件没有变化
手动执行，BGSAVE
LASTSAVE
可以通过lastsave命令获取最后一次成功执行快照的时间，给了一个时间戳
可以使用date -d @1691743564获取能看懂的时间
1.6、RDB优缺点
优点
RDB是Redis数据的一个非常紧凑的单文件时间点表示。RDB文件非常适合备份。例如，您可能希望在最近的24小时内每小时归档一次RDB文件，并在30天内每天保存一个RDB快昭。这使您可以在发生灾难时轻松恢复不同版本的数据集。
RDB非常适合灾难恢复，它是一个可以传输到远程数据中心或Amazon S3(可能已加密）的压缩文件。
RDB最大限度地提高了Redis 的性能，因为Redis 父进程为了持久化而需要做的唯一工作就是派生一个将完成所有其余工作的子进程。父进程永远不会执行磁盘I/О或类似操作。
AOF相比，RDB允许使用大数据集更快地重启。
在副本上，RDB支持重启和故障转移后的部分重新同步。
小结
适合大规模的数据恢复
按照业务定时备份
对数据完整性和一致性要求不高
RDB文件在内存中的加载速度要比AOF快得多
缺点
如果您需要在Redis停止工作时。(例如断电后）将数据丢失的可能性降到最低，那么RDB并不好。您可以配置生成RDB的不词保存点（例如，在对数据集至少5分钟和100次写入之后，您可以有多个保存点)。但是，您通常会每五分钟或更长时间创建一次RDB快照，因此，如果Redis 由于任何原因在没有正确关闭的情况下停止工作，您应该准备好丢失最新分钟的数据。
RDB需要经常 fork)以便使用子进程在磁盘上持久化。如果数据集很大，fork()可能会很耗时，并且如果数据集很大并且CPU性能不是很好，可能会导致Redis停止为客户端服务几毫秒甚至一秒钟。AOF也需要fork()但频率较低，您可以调整要重写日志的频率，而不需要对持久性进行任何权衡。
小结
在一定间隔时间做一次备份，所以如果redis意外down掉的话，就会丢失从当前至最近一次快照期间的数据，快照之间的数据会丢失
内存数据的全量同步，如果数据量太大会导致I/0严重影响服务器性能
RDB依赖于主进程的fork，在更大的数据集中，这可能会导致服务请求的瞬间延迟。fork的时候内存中的数据被克隆了一份，大致2倍的膨胀性，需要考虑
意外down机模拟，仍然是5秒2次修改的前提
正常的写入k1 k2 然后会生成一个备份
然后再写k3，这时候k3是存在的，没有写k4所以达不到备份条件，备份文件不会变化。
然后这时候找到redis的服务进程kill -9它，再次重新启动redis就会发现k3丢失了。
1.7、如何检查并修复dump.rdb文件
dump.rdb文件在写入或者是迁移的过程中是有可能损坏的，一个数据的损坏可能会导致全部的数据都读不出来了，所以需要修复它。
在/usr/local/bin中有redis-check-rdb
使用它可以检查dump.rdb文件有问题就修复，以下是正常的。
1.8、哪些情况下会触发RDB快照？
配置文件中默认的快照配置
手动save/bgsave命令
执行flushall/flushdb命令也会产生dump.rdb文件，但里面是空的
执行shutdown且没有设置开启AOF持久化
主从复制时，主节点自动触发
1.9、如何禁用快照
动态所有停止RDB保存规则的方法: redis-cli config set save ""
快照禁用
1.2.0、RDB优化配置项详解
配置文件SNAPSHOTTING模块
save <seconds> <changes>
dbfilename
dir
stop-writes-on-bgsave-error
默认是yes，如果配置成no，表示你不在乎数据不一致或者有其他的手段发现和控制这种不一致，那么在快照写入失败时，也能确保redis继续接受新的写请求
rdbcompression
默认为yes，对于存储到磁盘中的快照，可以设置是否进行压缩存储。如果是的话,redis会采用LZF算法进行压缩。如果你不想消耗CPU来进行压缩的话，可以设置为关闭此功能
rdbchecksum
默认是yes，在存储快照后，还可以让redis使用CRC64算法来进行数据校验，但是这样做会增加大约10%的性能消耗，如果希望获取到最大的性能提升，可以关闭此功能
rdb-del-sync-files
rdb-del-sync-files:在没有持久性的情况下删除复制中使用的RDB文件启用。默认情况下no，此选项是禁用的。
小结
RDB是一个非常紧凑的文件。
RDB在保存RDB文件时父进程唯一需要做的就是fork出一个子进程,接下来的工作全部由子进程来做,父进程不需要再做其他IO操作,所以RDB持久化方式可以最大化redis的性能.
与AOF相比,在恢复大的数据集的时候,RDB方式会更快一些.
数据丢失风险大
RDB需要经常fork子进程来保存数据集到硬盘上,当数据集比较大的时候,fork的过程是非常耗时的,可能会导致Redis在一些亳秒级不能相应客户端请求
2.0、AOF(Append Only File)
1.2.1、是什么
以日志的形式来记录每个写操作，将Redis执行过的所有写指令记录下来(读操作不记录)，只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作（那这样再执行一次就会很慢咯）。
默认情况下，redis是没有开启AOF(append only file)的。
开启AOF功能需要设置配置:appendonly yes
1.2.2、能干嘛
当你执行一个修改数据的命令（例如SET、DEL、INCR等）时，Redis会将这个命令追加到AOF文件的末尾。当Redis重启时，它会重新执行AOF文件中的所有命令，以恢复数据。
1.2.3、AOF持久化工作流程
Client作为命令的来源，会有多个源头以及源源不断的请求命令。
在这些命令到达Redis Server以后并不是直接写入AOF文件，会将其这些命令先放入AOF缓存中进行保存。这里的AOF缓冲区实际上是内存中的一片区域，存在的目的是当这些命令达到一定量以后再写入磁盘，避免频繁的磁盘IO操作。
AOF缓冲会根据AOF缓冲区同步文件的三种写回策略将命令写入磁盘上的AOF文件。
随着写入AOF内容的增加为避免文件膨胀，会根据规则进行命令的合并(又称AOF重写)，从而起到AOF文件压缩的目的。
1.2.4、三种写回策略
Always 永远
同步写回，每个写命令执行完立刻同步地将日志写回磁盘
everysec 每隔一秒（默认的）
每秒写回，每个写命令执行完，只是先把日志写到AOF文件的内存缓冲区,每隔1秒把缓冲区中的内容写入磁盘。
no 拒绝，不再写入
操作系统控制的写回，每个写命令执行完，只是先把日志写到AOF文件的内存缓冲区，由操作系统决定何时将缓冲区内容写回磁盘
1.2.5、AOF配置/启动/修复/恢复
配置文件说明
配置文件中打开aof
设置aof文件保存的位置和名称
位置：dir /myredis/
名字：/myredis/appenddirnam
文件名就使用默认的
把原来的那个rdb文件路径改成 dir /myredis即可
MP-AOF实现
顾名思义，MP-AOF就是将原来的单个AOF文件拆分成多个AOF文件。在MP-AOF中，我们将AOF分为三种类型。
本别为：
BASE:表示基础AOF，它一般由子进程通过重写产生，该文件最多只有一个。
INCR:表示增量AOF，它一般会在AOFRW开始执行时被创建，该文件可能存在多个。
HISTORY:表示历史AOF，它由BASE和INCR AOF变化而来，每次AOFRW成功完成时，本次AOFRW之前对应的BASE和INCR AOF都将变为HISTORY,HISTORY类型的AOF会被Redis自动删除。
为了管理这些AOF文件，我们引入了一个manifest(清单）文件来跟踪、管理这些AOF。同时，为了便于AOF备份和拷贝，我们将所有的AOF文件和manifest文件放入一个单独的文件目录中，目录名由appenddirname配置
所以生成的appendonlydir/下会有这三个文件
base基本文件
incr增量文件
manifest清单文件
测试：停止redis服务再开启
正常恢复
为了证明是aof的恢复策略，将原来的的dumpfiles和新生成的dump6379.rdb 文件删除，然后先将appendonlydir/目录备份一下成appendonlydir.bak。然后执行flushdb命令清空k。再停止重启redis服务查看keys就会是空的，因为flushdb也是写命令。
停止服务重新连接再查看keys就是空
把原来有数据的换回来，再重新查看就有了
进入appendonlydir目录ll查看这三个文件大小，发现base和manifest都是88，incr是283
现在再set k3 重新查看就会发现只有incr变大了，其他两个没有变化。
异常恢复
故意乱写正常的AOF文件,模拟网络闪断文件写error，重启Redis之后就会进行AOF文件的载入，发现启动都不行，异常修复命令:redis-check-aof --fix进行修复
模拟场景：每一秒钟写入一次,内容才写了一小半，没有写完整。突然，redis挂了，导致,aof文件错误
我们知道，每一次写操作都被记录在了appendonly.aof.1.incr.aof文件中所以打开编辑随便瞎写保存
然后shutdown重新加载，发现aof错误，连服务器都无法启动
之前有一个修复文件的，redis-check-aof修复工具，现在使用就可以修复错误的aof文件
进行修复：redis-check-aof appendonly.aof.1.incr.aof
看一下它压根就没有修复，因为没有加 --fix，重新再来
现在才算是修复成功，那些不符合它的语法规则的内容全部清空
然后我们再重新启动redis，现在就好了以前的keys都还在
1.2.6、AOF优点
使用AOF Redis更加持久∶您可以有不同的fsync策略:根本不fsync、每秒 fsync、每次查询时fsync。使用每秒fsync 的默认策略，写入性能仍然很棒。fsync是使用后台线程执行的，当没有fsync正在进行时，主线程将努力执行写入，因此您只能丢失一秒钟的写入。
AOF日志是一个仅附加日志，因此不会出现寻道问题，也不会在断电时出现损坏问题。即使由于某种原因（磁盘已满或其他原因）日志以写一半的命令结尾，redis-check-aof 工具也能够轻松修复它。
当AOF变得太大时，Redis 能够在后台自动重写AOF。重写是完全安全的，因为当Redis继续附加到旧文件时，会使用创建当前数据集所需的最少操作集生成一个全新的文件，一旦第二个文件准备就绪，Redis就会切换两者并开始附加到新的那一个。
AOF以易于理解和解析的格式依次包含所有操作的日志。您甚至可以轻松导出AOF 文件。例如，即使您不小心使用该FLUSHALL命令刷新了所有内容，只要在此期间没有执行日志重写，您仍然可以通过停止服务器、删除最新命令并重新启动Redis来保存您的数据集.直接到appendonly.aof.1.incr.aof找到flushall删除它再重启。
一句话：更好的保护数据不丢失、性能高、可做紧急恢复
1.2.7、AOF缺点
AOF文件通常比相同数据集的等效RDB文件大。
根据确切的 fsync策略，AOF可能比RDB慢。一般来说，将fsync设置为每秒性能仍然非常高，并且在禁用fsync的情况下，即使在高负载下它也应该与RDB一样快。即使在巨大的写入负载的情况下，RDB仍然能够提供关于最大延迟的更多保证。
1.2.8、AOF重写机制
由于AOF持久化是Redis不断将写命令记录到AOF 文件中，随着Redis不断的进行，AOF的文件会越来越大,文件越大，占用服务器内存越大以及AOF恢复要求时间越长。
为了解决这个问题，Redis新增了重写机制，当AOF文件的大小超过所设定的峰值时，Redis就会自动启动AOF文件的内容压缩.只保留可以恢复数据的最小指令集或者可以手动使用命令 bgrewriteaof 来重新。
一句话：启动AOF文件的内容压缩，只保留可以恢复数据的最小指令集。
触发机制
官网默认的配置
注意，同时满足，且的关系才会触发 1根据上次重写后的aof大小，判断当前aof大小是不是增长了1倍
2重写时满足的文件大小
auto-aof-rewrite-percentage配置项设置了一个阈值，当当前AOF文件大小比上次重写后的AOF文件大小大这个百分比时，Redis会自动触发AOF重写。例如，auto-aof-rewrite-percentage 100表示当当前AOF文件大小是上次重写后的AOF文件大小的两倍时，Redis会自动触发AOF重写。
需要注意的是，为了避免在AOF文件很小的时候频繁触发重写，Redis还有一个auto-aof-rewrite-min-size配置项，只有当AOF文件大小大于这个值时，auto-aof-rewrite-percentage才会生效。默认情况下，auto-aof-rewrite-min-size的值是64MB。
自动触发
满足配置文件中的选项后，Redis会记录上次重写时的AOF大小， 默认配置是当AOF文件大小是上次rewrite后大小的一倍且文件大于64M时
启动AOF文件的内容压缩，只保留可以恢复数据的最小指令集。
一个例子：
第一次你写 set k1 v1
然后改成 set k1 v2
最后改成 set k1 v3
如果不重写，那么这3条语句都在aof文件中(如果这样的命令有100条或者1000条呢)，内容占空间不说启动的时候都要执行一遍，共计3条命令;但是，我们实际效果只需要set k1 v3这一条，所以，开启重写后，只需要保存set k1 v3就可以了只需要保留最后一次修改值，相当于给aof文件瘦身减肥，性能更好。
AOF重写不仅降低了文件的占用空间，同时更小的AOF也可以更快地被Redis加载。
测试步骤：
开启aof当然必须的
重写峰值修改为1k 配置文件第1480行修改
关闭混合，设置为no
删除之前的全部aof和rdb，清除干扰项
配置完成后，重新连接，查看生成的配置文件，最初始的0
然后set k1 1
查看大小，变成51
然后不断地写k1，查看一下appendonly.aof.1.incr.aof文件发现它每一次设置k1都给记录下来了，这样就很多很大，占用内存。继续一直写k1直到1024到达阈值。
到了1024，发现appendonly.aof.1.base.aof和appendonly.aof.1.incr.aof都变成了2版本的，appendonly.aof.2.incr.aof我写多了超太多重新记录了......
我们看一下appendonly.aof.2.base.aof，发现里面就是k1的最终到达1k时的结果，显然比没有压缩前少多了。
手动触发
客户端向服务器发送bgrewriteaof命令即可
这个时候配置文件又变成3了
小结
AOF文件重写并不是对原文件进行重新整理，而是直接读取服务器现有的键值对，然后用一条命令去代替之前记录这个键值对的多条命令，生成一个新的文件后去替换原来的AOF文件。 AOF文件重写触发机制:通过redis.conf配置文件中的 auto-aof-rewrite-percentage:默认值为100，以及auto-aof-rewritemin-size: 64mb配置，也就是说默认Redis会记录上次重写时的AOF大小，默认配置是当AOF文件大小是上次rewrite后大小的一倍且文件大于64M时触发。
重写原理
1: 在重写开始前，redis会创建一个“重写子进程”，这个子进程会读取现有的AOF文件，并将其包含的指令进行分析压缩并写入到一个格时文件中。 2∶与此同时，主进程会将新接收到的写指令一边累积到内存缓冲区中，一边继续写入到原有的AOF文件中，这样做是保证原有的AOF文件的可用性，避免在重写过程中出现意外。 3:当“重写子进程”完成重写工作后，它会给父进程发一个信号，父进程收到信号后就会将内存中缓存的写指令追加到新AOF文件中 4:当追加结束后，redis就会用新AOF文件来代替旧AOF文件，之后再有新的写指令，就都会追加到新的AOF文件中 5:重写aof文件的操作，并没有读取旧的aof文件，而是将整个内存中的数据库内容用命令的方式重写了一个新的aof文件，这点和快照有点类似
1.2.9、AOF配置文件APPEND ONLY MODE模块
总结
3、RDB+AOF混合模式
可否共存？如果共存听谁的？
从配置文件中可以看到开了AOF那就是优先使用AOF的
数据恢复顺序和加载流程（重点）
在同时开启rdb 和aof持久化时，重启时只会加载aof文件，不会加载rdb 文件
怎么选？用哪个？
RDB持久化方式能够在指定的时间间隔能对你的数据进行快照存储
AOF持久化方式记录每次对服务器写的操作,当服务器重启的时候会重新执行这些命令来恢复原始的数据,AOF命令以redis协议追加保存每次写的操作到文件末尾.
同时开启两种持久化方式
在这种情况下，当redis重启的时候会优先载入AOF文件来恢复原始的数据，因为在通常情况下AOF文件保存的数据集要比RD免文件保存的数据集要完整.
RDB的数据不实时，同时使用两者时服务器重启也只会找AOF文件。那要不要只使用AOF呢?
作者建议不要，因为RDB更适合用于备份数据库(AOF在不断变化不好备份)，留着rdb作为一个万一的手段。
混合方式
结合了RDB和AOF的优点，既能快速加载又能避免丢失过多的数据。
打开方法：
设置aof-use-rdb-preamble的值为yes yes表示开启，设置为no表示禁用
两者的配合策略：RDB镜像做全量持久化，AOF做增量持久化。
先使用RDB进行快照存储，然后使用AOF持久化记录所有的写操作，当重写策略满足或手动触发重写的时候，将最新的数据存储为新的RDB记录。。这样的话，重启服务的时候会从RDB和AOF两部分恢复数据，既保证了数据完整性，又提高了恢复数据的性能。简单来说:混合持久化方式产生的文件一部分是RDB格式，一部分是AOF格式。----》AOF包括了RDB头部+AOF混写
纯缓存模式
同时关闭RDB+AOF
怎么关？----->save " " 禁用rdb 禁用rdb持久化模式下，我们仍然可以使用命令save、bgsave生成rdb文件
appendonly no
禁用aof 禁用aof持久化模式下，我们仍然可以使用命令bgrewriteaof生成aof文件
Redis事务
Redis 事务就是多条命令合成一块形成一个组，成为一个不可分割，具备原子性，不被干扰打断，统一执行的一套批处理命令。
在Redis中，事务是通过MULTI、EXEC、DISCARD和WATCH这几个命令来实现的。当你执行MULTI命令后，后续的所有命令都会被放入一个队列，然后当你执行EXEC命令时，队列中的所有命令会被一次性、顺序地执行。这就保证了事务的原子性，即事务中的所有命令要么全部执行，要么全部不执行。
需要注意的是，虽然Redis的事务保证了原子性，但是它并不支持回滚。也就是说，如果事务中的某个命令执行失败，Redis不会回滚已经执行的命令。相反，Redis会继续执行事务中的其他命令，然后在EXEC命令的返回值中指示出哪个命令失败了1。
可以一次执行多个命令，本质是一组命令的集合。一个事务中的
所有命令都会序列化，按顺序地串行化执行而不会被其它命令插入，不许加塞
关系型数据库 | Redis |
|---|---|
bengin开启事务 | multi开启事务 |
执行一个个的SQL | 执行一组命令 |
commot/rollback | exec |
能够：一个队列中，一次性、顺序性、排他性的执行一系列命令
Redis事务VS数据库事务
单独的隔离操作 | Redis的事务仅仅是保证事务里的操作会被连续独占的执行，redis命令执行是单线程架构，在执行完事务内所有指令前是不可能再去同时执行其他客户端的请求的 |
|---|---|
没有隔离级别的概念 | 因为事务提交前任何指令都不会被实际执行，也就不存在"事务内的查询要看到事务里的更新，在事务外查询不能看到"这种问题了 |
不保证原子性 | Redis的事务不保证原子性，也就是不保证所有指令同时成功或同时失败，只有决定是否开始执行全部指令的能力，没有执行到一半进行回滚的能力 |
排它性 | Redis会保证一个事务内的命令依次执行，而不会被其它命令插入 |
使用
Redis事务常用命令
DISCARD 取消事务，放弃执行事务块内的所有命令。
EXEC 执行所有事务块内的命令。
MULTI 标记—个事务块的开始。
UNWATCH 取消WATCH命令对所有key的监视。
WATCH key [key ...] 监视一个(或多个) key ,如果在事务执行之前这个(或这些)key被其他命令所改动，那么事务将被打断。
正常执行
MULTI 事务开始
EXEC 执行事务块中的命令
放弃事务
MULTI
DISCARD
全体连坐（编译时就异常）
一条命令出错全部打回，都不执行
冤头债主（运行时异常）
哪个命令错了把那个命令揪出来，其他的正常执行
Redis不提供事务回滚的功能，开发者必须在事务执行出错后，自行恢复数据库状态。
前期语法都没错，编译通过;执行exec后报错:对的就执行完成，错的没办法执行。
watch监控
Redis使用Watch来提供乐观锁定，类似于CAS(Check-and-Set)
悲观锁(Pessimistic Lock)，顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改, 所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会block直到它拿到锁。
乐观锁(Optimistic Lock)，顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改 所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据。 乐观锁策略：提交版本必须大于记录当前版本才能执行更新
watch
初始化k1和balance两个key,先监控再开启multi，保证两key变动在同一个事务内，正常情况。
有加塞篡改的情况下，这个balance在事务中还没有执行时，启动第二个连接把balance修改成300，然后再正常的去执行事务，发现返回了空事务没有执行
所以，使用基于乐观锁的watch监控键key，当在事务执行（exec）之前有人修改了被监控的key，那么这个事务就会都失败，命令组都不会执行了。会返回一个nil提示失败了
UNWATCH
放弃监控
一旦执行了exec之前加的监控锁都会被取消掉了 当客户端连接丢失的时候(比如退出链接)，所有东西都会被取消监视
小结
开启:以MULTI开始一个事务
入队:将多个命令入队到事务中，接到这些命令并不会立即执行,而是放到等待执行的事务队列里面
执行:由EXEC命令触发事务
Redis管道
如何优化频繁命令往返造成的性能瓶颈?
-----批处理打包一次性发给服务器，mset命令可以同时执行多条命令
Redis是一种基于客户端-服务端模型以及请求/响应协议的TCP服务。一个请求会遵循以下步骤:
1 客户端向服务端发送命令分四步(发送命令-命令排队-命令执行-返回结果)，并监听Socket返回，通常以阻塞模式等待服务端响应。
2 服务端处理命令，并将结果返回给客户端。
上述两步称为:Round Trip Time(简称RTT,数据包往返于两端的时间)，问题笔记最下方
如果同时需要执行大量的命令，那么就要等待上一条命令应答后再执行，这中间不仅仅多了RTT(Round Time Trip)，而且还频繁调用系统IO，发送网络请求，同时需要redis调用多次read()和write()系统方法，系统方法会将数据从用户态转移到内核态，这样就会对进程上下文有比较大的影响了，性能不太好，o(_T)o
定义
管道(pipelinge)可以一次性发送多条命令给服务端，服务端依次处理完完毕后，通过一条响应一次性将结果返回，通过减少客户端与redis的通信次数来实现降低往返延时时间。pipeline实现的原理是队列，先进先出特性就保证数据的顺序性。
简单讲：Pipeline是为了解决RTT往返回时，仅仅是将命令打包一次性发送，对整个Redis的执行不造成其它任何影响，也就是，批处理命令变种优化措施，类似Redis的原生批命令(mget和mset)
。
案例
在/myredis目录下新建一个cmd.txt文件，里面写好要执行的命令
管道小结
Pipeline与原生批量命令对比
原生批量命令是原子性(例如:mset, mget)，pipeline是非原子性； 原生批量命令一次只能执行一种命令（没办法执行不同类型的命令），pipeline支持批量执行不同命令；
原生批命令是服务端实现，而pipeline需要服务端与客户端共同完成；
Pipeline与事务对比
事务具有原子性,管道不具有原子性； 管道一次性将多条命令发送到服务器，事务是一条一条的发，事务只有在接收到exec命令后才会执行，管道不会；
执行事务时会阻塞其他命令的执行，而执行管道中的命令时不会；
使用Pipeline注意事项
pipeline缓冲的指令只是会依次执行，不保证原子性，如果执行中指令发生异常，将会继续执行后续的指令；
使用pipeline组装的命令个数不能太多，不然数据量过大客户端阻塞的时间可能过久，同时服务端此时也被迫回复一个队列答复，占用很多内存；
Redis发布订阅（了解）
定义
是一种消息通信模式:发送者(PUBLISH)发送消息，订阅者(SUBSCRIBE)接收消息，可以实现进程间的消息传递。
Redis可以实现消息中间件MQ的功能，通过发布订阅实现消息的引导和分流。不推荐使用该功能，专业的事情交给专业的中间件处理，redis做好分布式缓存功能就行。
Redis客户端可以订阅任意数量的频道，类似我们微信关注多个然众号
发布/订阅其实是一个轻量的队列，只不过数据不会被持久化，一般用来处理实时性较高的异步消息
常用命令
PSUBSCRIBE pattern [pattern ...订阅—个或多个符答给定模式的领道。
PUBSUB subcommand [argument [argument ...J]查看订阅与发布系统状态。
PUBLISH channel message将信息发送到指定的频道。
PUNSUBSCRIBE[pattern[pattern退订所有给定模式的频道。
SUBSCRIBE channel [channel ...订阅给定的一个或多个频道的信息。
推荐先执行订阅后再发布，订阅成功之前发布的消息是收不到的
订阅的客户端每次可以收到一个3个参数的消息
消息的种类、始发频道的名称、实际的消息内容
UNSUBSCRIBE [channel [channel ...]指退订给定的频道。
A和B是订阅者，C是发布者，订阅者先订阅，发布者再发布
缺点
发布的消息在Redis系统中不能持久化，因此，必须先执行订阅，再等待消息发布。如果先发布了消息，那么该消息由于没有订阅者,消息将被直接丢弃；
消息只管发送对于发布者而言消息是即发即失的，不管接收，也没有ACK机制，无法保证消息的消费成功；
以上的缺点导致Redis的Pub/Sub模式就像个小玩具，在生产环境中几乎无用武之地，为此 Redis5.0版本新增了Stream数据结构，不但支持多播，还支持数据持久化，相比Pub/Sub更加的强大；
Redis主从复制（replica）
master以写为主，Slave以读为主
当master数据变化的时候，自动将新的数据异步同步到其它slave数据库
能干什么
读（从）写（主）分离
容灾恢复
数据备份
水平扩容支撑高并发
怎么做
配从(库)不配主(库)
权限细节很重要
master如果配置了requirepass参数，需要密码登陆
那么slave就要配置masterauth来设置校验密码，否则的话master会拒绝slave的访问请求
基本操作命令
info replication
可以查看复制节点的主从关系和配置信息
replicaof主库IP主库端口
一般写入进redis.conf配置文件内
slaveof主库IP主库端口
每次与master断开之后，都需要重新连接，除非你配置进redis.conf文件 在运行期间修改slave节点的信息，如果该数据库已经是某个主数据库的从数据库，那么会停止和原主数据库的同步关系转而和新的主数据库同步，重新拜码头
slaveof no one
使当前数据库停止与其他数据库的同步，转成主数据库，自立为王
1、演示
一主二从，三台虚拟机都安装redis，配置文件
192.168.111.19 master
192.168.111.23 slave1
192.168.111.24 slave2
master配置文件配置
1.开启daemonize yes
2.注释掉bind 127.0.0.1
3.protected-mode no
4.指定端口
5.指定当前工作目录，dir
6.pid文件名字,pidfile
7.log文件名字,logfile
8.requirepass
9.dump.rdb名字
10.aof文件,appendfilename（可用可不用）
11.从机访问主机的通行密码masterauth，必须（从机需要配置主机不需要）
从机1的6380的配置
从机2的6381的配置和从机1的一样
先启动master然后再启动两个slave
启动两个slave时一定要指定端口号
查看master的日志,可以看到两个从机连接成功
vim /myredis/6379.log
master使用命令查看连接状态
info replication
slave也查看一下连接状态
查看keys可以看到主机的键都备份过来了
从机不能执行写命令
从机切入点问题
slave是从头开始复制还是从切入点开始复制?master启动，写到k3 slave1跟着master同时启动，跟着写到k3 slave2写到k3后才启动，那之前的是否也可以复制?
初始三台机器全部都空的，然后把slave2断开连接，掉队了，然后在master写入k1k2k3,写完再把slave2连接起来，看看它有没有k1k2k3
首次一锅端，后续跟随，master写，slave跟
主机shutdown后，从机会上位吗？
答案是不会，从机不动，原地待命，从机数据可以正常使用;等待主机重启动归来。不会谋权篡位的，等待主公归来。
把master挂掉，再看两个slave的状态，只是与主机断开连接了而已
主机shutdown后，重启后主从关系还在吗?从机还能否顺利复制?
上面的master挂掉了，这里我们连上看看keys还在不在，再设置k4k5看从机们能不能跟得上
某台从机down后，master继续，从机重启后它能跟上大部队吗?
和上面的问题一样是肯定的，随时同步。
命令操作手动指定
将那两个slave配置文件中的主机配置注释掉，让它自己也是master，等于说现在三个master
现在使用命令去指定slave跟的主机是谁
SLAVEOF 192.168.111.19 6379新加key
连接上主机查看状态，刚连上的时候等他一会儿需要同步
list也在
注意：SLAVEOF指定master只不过是个临时的，单次生效而已，一旦重启slave就没用了，所以确定了主从关系后一定要写在配置文件中才长期有效
薪火相传
上一个slave可以是下一个slave的master，slave同样可以接收其他slaves的连接和同步请求，那么该slave作为了链条中下一个的master,可以有效减轻主master的写压力
中途变更转向:会清除之前的数据，重新建立拷贝最新的
slaveof 新主库IP新主库端口
现在恢复到之前的配置文件，两个slave还都认192.168.111.19做master，然后使用SLAVEOF命令让192.168.111.24的机器连192.168.111.23。
127.0.0.1:6381> slaveof 192.168.111.23 6380
OK
再查看master的状态，发现就只有slave1连接了
在master中set k5 v5在slave1和slave2上都能同步
虽然此时slave1是slave2的master，但是它还是没用写的权限。
如果某个slave想变成master，那就直接 SLAVEOF no one，关闭slave就成了 master
2、复制原理和工作流程
slave启动，同步初请
slave启动成功连接到master后会发送一个sync命令。
slave首次全新连接master,一次完全同步（全量复制)将被自动执行，slave自身原来有的数据就会被全覆盖掉。
首次连接，全量复制
master节点收到sync命令后会开始在后台保存快照(即RDB持久化，主从复制时会触发RDB),同时收集所有接收郅的用于修改数据集命令缓存起来，master节点执行RDB持久化完后，master将rdb快照文件和所有缓存的命令发送到所有slave,以完成一次完全同步
而slave服务在接收到数据库文件数据后，将其存盘并加载到内存中，从而完成复制初始化
心跳持续，保持通信
repl-ping-replica-period 10
master发出PING包的周期，默认是10秒
进入平稳,增量复制
Master继续将新的所有收集到的修改命令自动依次传给slave,完成同步
从机下线，重连续传
master会检查backlog里面的offset，master和slave都会保存一个复制的offset还有一个masterId,offset是保存在backlog中的。Master只会把已经复制的offset后面的数据复制给Slave，类似断点续传
复制的缺点
复制延时,信号衰减
由于所有的写操作都是先在Master上操作，然后同步更新到Slave上，所以从Master同步到Slave机器有一定的延迟，当系统很繁忙的时候，延迟问题会更加严重，Slave机器数量的增加也会使这个问题更加严重。
master也会挂掉
默认情况下，不会在slave节点中自动重选一个master
难不成每一次都要人工干预？------------>无人值守安装变成刚需（哨兵）从剩下的从节点中选出来一个当master
Redis哨兵（sentinel）
是什么
吹哨人巡查监控后台master主机是否故障，如果故障了根据投票数自动将某一个从库转换为新主库，继续对外服务。哨兵起步3个。
作用
监控redis运行状态,r包括master和slave
当master down机，能自动将slave切换成新master
能做的
主从监控
监控主从redis库运行是否正常
消息通知
哨兵可以将故障转移的结果发送给客户端
故障转移
如果Master异常，则会进行主从切换，将其中一个slave作为新Master
配置中心
客户端通过连接哨兵来获得当前Redis服务的主节点地址
1、哨兵实例
Redis Sentinel架构，前提说明
3个哨兵自动监控和维护集群,不存放数据，只是吹哨人
1主2从 ，用于数据读取和存放
3个哨兵直接在master的机器上做，启动哨兵服务就是读取哨兵的配置文件，复制三份使用不同的端口。
[root@localhost myredis]# cp -r /opt/redis-7.0.0/sentinel.conf .配置文件中两个重要的选项
sentinel monitor <master-name> <ip> <redis-port> <quorum>
设置要监控的master服务器
quorum表示最少有几个哨兵认可客观下线,同意故障迁移的法定票数。
我们知道，网络是不可靠的，有时候一个sentinel会因为网络堵塞而误以为一个master redis已经死掉了，在sentinel集群环境下需要多个sentinel互相沟通来确认某个master是否真的死了，quorum这个参数是进行客观下线的一个依据，意思是至少有quorum个sentinel认为这个master有故障，才会对这个master进行下线以及故障转移。因为有的时候，某个sentinel节点可能因为自身网络原因，导致无法连接master，而此时master并没有出现故障，所以，这就需要多个sentinel都一致认为该master有问题，才可以进行下一步操作，这就保证了公平性和高可用。
sentinel auth-pass <master-name> <password>
master设置了密码,连接master服务的密码
sentinel down-after-milliseconds <master-name> <milliseconds>:
指定多少毫秒之后，主节点没有应答哨兵，此时哨兵主观上认为主节点下线
sentinel parallel-syncs <master-name> <nums>: 表示允许并行同步的slave个数，当Master挂了后，哨兵会选出新的Master，此时，剩余的slave会向新的master发起同步数据
sentinel failover-timeout <master-name> <milliseconds>: 故障转移的超时时间，进行故障转移时，如果超过设置的毫秒，表示故障转移失败
sentinel notification-script <master-name> <script-path> : 配置当某一事件发生时所需要执行的脚本
sentinel client-reconfig-script <master-name> <script-path>:
客户端重新配置主节点参数脚本
1.1、哨兵配置文件配置
sentinel26379.conf
bind 0.0.0.0 daemonize yes protected-mode no port 26379 logfile "/myredis/sentinel26379.log" pidfile /var/run/redis-sentinel26379.pid dir /myredis sentinel monitor mymaster 192.168.111.19 6379 2 sentinel auth-pass mymaster 123456直接使用就好
sentinel26380.conf
sentinel26381.conf
现在先把主从复制的master和两个slave启动起来，测试一下能否正常同步
现在启动我们的那三个哨兵
redis-sentinel sentinel26379.conf --sentinel
redis-sentinel sentinei26380.conf --sentinel
redis-sentinel sentinel26381.conf --sentinel
三个哨兵已经启动
我们查看一下哨兵的日志文件，它所监控的master的信息以及master下面的两个slave的信息都在这里，还有另外两个哨兵也在这里。
注意：日志中有一个Sentinel new configuration saved on disk这句话表示哨兵会自动给配置文件添加内容，里面包含了我所监控的master信息和master下面的从机有哪些，以及另外两个哨兵是谁
现在让master挂掉
关心的问题
两台从机数据是否OK
两台从机数据在master挂掉的一点时间内，服务会断开，数据读不出来。那是因为哨兵正在进行判断master没了，然后法定票数选举，重新发送心跳包建立连接。过一小会再读取两台从机的数据就可以读出来，之前断开的连接都不需要你自己手动连接的就可以恢复。
是否会从剩下的2台机器上选出新的master
master让给了6380这个从节点了，但我的另一个6381并没有连接上新master
之前down机的master机器重启回来， 谁将会是新老大?会不会双master冲突?
即使master回来了，master的位置也不是它了，换成6380了，它6379只能跟着6380混了。
vim sentinel26379.log 查看日志
6379的身份变成slave了
6380下面跟着6379和6381
现在查看一下6379的配置文件，到最底部，发现最下面的内容是自动生成的，这里配置了新的maaster的ip和端口号
然后去看一下6380的配置文件，发现以前我们配置的以6379为master的那一行选项被自动删除了
replicaof 192.168.111.19 6379 这个被自动删除了，自然它自己就变成master了
下面又自动新加的内容
结论
文件的内容，在运行期间会被sentinel动态进行更改
Master-Slave切换后，master_redis.conf、slave_redis.conf和sentinel.conf的内容都会发生改变,即master_ redis.conf中会多一行slaveof的配置，sentinel.conf的监控目标会随之调换
生产都是不同机房不同服务器，很少出现3个哨兵全挂掉的情况
可以同时监控多个master，一行一个
2、哨兵运行流程和选举原理（重点，背下来吧）
当一个主从配置中的master失效之后，sentinel可以选举出一个新的master，用于自动接替原master的工作，主从配置中的其他redis服务器自动指向新的master同步数据。一般建议sentinel采取奇数台，防止某一台sentinel无法连接到master导致误切换
SDown主观下线(Subjectively Down)
SDOWN(主观不可用)是单个sentinel自己主观上检测到的关于master的状态，从sentinel的角度来看，如果发送了PING心跳后，在一定时间内没有收到合法的回复，那么这个Sentinel会主观的(单方面的)认为这个master不可以用了。
sentinel配置文件中的down-after-milliseconds设置了判断主观下线的时间长度。
sentinel down-after-milliseconds <masterName> <timeout> 默认30秒表示master被当前sentinel实例认定为失效的间隔时间，这个配置其实就是进行主观下线的一个依据，master在多长时间内（默认30秒）一直没有给Sentine返回有效信息，则认定该master主观下线。也就是说如果多久没联系上redis-servevr，认为这个redis-server进入到失效（SDOWN）状态。
ODown客观下线(Objectively Down)
ODOWN需要一定数量的sentinel，多个哨兵达成一致意见才能认为一个master客观上已经宕掉。
quorum这个参数是进行客观下线的一个依据，法定人数/法定票数。
意思是至少有quorum个sentinel认为这个master有故障才会对这个master进行下线以及故障转移。因为有的时候，某个sentinel节点可能因为自身网络原因导致无法连接master，而此时master并没有出现故障，所以这就需要多个sentinel都一致认为该master有问题，才可以进行下一步操作，这就保证了公平性和高可用。
选举出领导者哨兵(哨兵中选出兵王)
当主节点被判断客观下线以后，各个哨兵节点会进行协商，先选举出一个领导者哨兵节点(兵王)并由该领导者节点，也即被选举出的兵王进行failover(故障迁移)
那么，兵王这个领导者又是怎么选出来的呢？
使用的是Raft算法
监视该主节点的所有哨兵都有可能被选为领导者，选举使用的算法是Raft算法;
Raft算法的基本思路是先到先得:
即在一轮选举中，哨兵A向B发送成为领导者的申请，如果B没有同意过其他哨兵，则会同意A成为领导者。
由兵王开始推动故障切换流程并选出一个新master
有三个步骤
新主登基
选出新master的规则，剩余slave节点健康前提下
redis.conf文件中，优先级slave-priority或者replica-priority最高的从节点(数字越小优先级越高)
复制偏移位置offset最大的从节点
最小Run ID的从节点 字典顺序，ASCII码
群臣俯首
被选出来的新master先执行slaveof no one命令成为新的主节点，并通过slaveof命令让其他节点成为其从节点。
Sentinel leader会对选举出的新master执行slaveof no one操作，将其提升为master节点；
Sentinel leader向其它slave发送命令，让剩余的slave成为新的master节点的slave；
旧主拜服
将之前已下线的老master设置为新选出的新master的从节点，当老master重新上线后，它会成为新master的从节点。
Sentinel leader会让原来的master降级为slave并恢复正常工作。
上述的failover操作均由sentinel自己独自完成，完全无需人工干预。
3、哨兵使用注意事项
哨兵节点的数量应为多个，哨兵本身应该集群，保证高可用；
哨兵节点的数量应该是奇数；
各个哨兵节点的配置应一致；
如果哨兵节点部署在Docker等容器里面，尤其要注意端口的正确映射；
哨兵集群+主从复制，并不能保证数据零丢失；
Redis集群（cluster）
由于数据量过大，单个Master复制集难以承担，因此需要对多个复制集进行集群，形成水平扩展每个复制集只负责存储整个数据集的一部分，这就是Redis的集群，其作用是提供在多个Redis节点间共享数据的程序集。
一句话总结：Redis集群是一个提供在多个Redis节点间共享数据的程序集，Redis集群可以支持多个Master。
Redis集群能干嘛？
Redis集群支持多个Master，每个Master又可以挂载多个Slave
读写分离
支持数据的高可用
支持海量数据的读写存储操作
由于Cluster自带Sentinel的故障转移机制，内置了高可用的支持，无需再去使用哨兵功能
客户端与Redis的节点连接，不再需要连接集群中所有的节点，只需要任意连接集群中的一个可用节点即可
槽位slot负责分配到各个物理服务节点，由对应的集群来负责维护节点、插槽和数据之间的关系
1、集群算法-分片-槽位slot
1.1、槽位
集群的密钥空间被分成16384个槽,有效地设置了16384个主节点的集群大小上限(但是，建议的最大节点大小约为1000个节点)。 集群中的每个主节点处理16384个哈希槽的一个子集。当没有集群重新配置正在进行时(即哈希槽从一个节点移动到另一个节点)，集群是稳定的。当集群稳定时，单个哈希槽将由单个节点提供服务(但是，服务节点可以有一个或多个副本，在网络分裂或故障的情况下替换它，并且可以用于扩展读取陈旧数据是可接受的操作)。
Redis集群没有使用一致性hast,而是引入了哈希槽的概念。
Redis 集群有16384个哈希槽,每个key通过CRC16校验后对16384取模来决定放置哪个槽.集群的每个节点负责一部分hash槽。
举个例子,比如当前集群有3个节点那么:
1.2、分片
什么是分片？
使用Redis集群时我们会将存储的数据分散在多台redis机器上，这称为分片。简而言之，集群中的每一个redis实例都被认为是整个数据的一个分片。
如何找到给定key的分片？
为了找到给定key的分片，我们对key进行CRC16(key)算法处理并通过对总分片数量取模。然后，使用确定性哈希函数，这意味着给定的key将多次始终映射到同一个分片，我们可以推断将来读取特定key的位置。
槽位和分片的优势
最大优势，方便扩缩容和数据分派查找
这种结构很客易添加或者删除节点.比如攻果我想新添加个节点D,我需要从节点A,B,.C中得部分槽到D上.如果我想移除节点A需要将A中的槽移到B和C节点上,然后将没有任何槽的A节点从集群中移除即呵.由于从一个节点将哈希槽移动到另一个节点并不会停止服务,所以无论添加删除或者改变某个节点的哈希槽的数量都不会造成集群不可用的状态.
1.3、slot槽位映射的3种解决方案
哈希取余分区(小厂)
假设：2亿条记录就是2亿个k,v，我们单机不行必须要分布式多机，假设有3台机器构成一个集群，用户每次读写操作都是根据公式：hash(key) %N个机器台数，计算出哈希值，用来决定数据映射到哪一个节点上。
优点：
简单粗暴，直接有效，只需要预估好数据规划好节点，例如3台、8台、10台，就能保证一段时间的数据支撑。使用Hash算法让固定的一部分请求落到同一台服务器上，这样每台服务器固定处理一部分请求（并维护这些请求的信息），起到负载均衡+分而治之的作用。
缺点：
原来规划好的节点，进行扩容或者缩容就比较麻烦了，不管扩缩，每次数据变动导致节点有变动，映射关系需要重新进行计算，在服务器个数固定不变时没有问题，如果需要弹性扩容或故障停机的情况下，原来的取模公式就会发生变化:Hash(key)/3会变成Hash(key)/?[此时地址经过取余运算的结果将发生很大变化，根据公式获取的服务器也会变得不可控。 某个redis机器宕机了，由于台数数量变化，会导致hash取余全部数据重新洗牌。
一致性哈希算法分区(中厂)
一致性哈希算法在1997年由麻省理工学院中提出的，设计目标是为了解决，分布式缓存数据变动和映射问题，某个机器宕机了，分母数量改变了，自然取余数不OK了。
目的是当服务器个数发生变动时，尽量减少影响客户端到服务器的映射关系
3大步骤
算法构建一致性哈希环
一致性哈希算法必然有个hash函数并按照算法产生hash值，这个算法的所有可能哈希值会构成一个全量集，这个集合可以成为一个hash空间[0,2^32-1]，这个是一个线性空间，但是在算法中，我们通过适当的逻辑控制将它首尾相连(0 = 2^32),这样让它逻辑上形成了一个环形空间。
它也是按照使用取模的方法，前面笔记介绍的节点取模法是对节点（服务器）的数量进行取模。而一致性Hash算法是对2^32取模，简单来说，一致性Hash算法将整个哈希值空间组织成一个虚拟的圆环，如假设某哈希函数H的值空间为0-232-1(即哈希值是一个32位无符号整形），整个哈希环如下图:整个空间按顺时针方向组织，圆环的正.上方的点代表0，0点右侧的第一个点代表1，以此类推，2、3、4、……直到232-1，也就是说0点左侧的第一个点代表232-1，0和232-1在零点中方向重合，我们把这个由2^32个点组成的圆环称为Hash环。
服务器IP节点映射
将集群中各个IP节点映射到环上的某一个位置。
将各个服务器使用Hash进行一个哈希，具体可以选择服务器的IP或主机名作为关键字进行哈希，这样每台机器就能确定其在哈希环上的位置。假如4个节点NodeA、B、C、D，经过IP地址的哈希函数计算(hash(ip))，使用IP地址哈希后在环空间的位置如下:
key落到服务器的落键规则
当我们需要存储一个kv键值对时，首先计算key的hash值，hash(key)，将这个key使用相同的函数Hash计算出哈希值并确定此数据在环上的位置，从此位置沿环顺时针“行走”，第一台遇到的服务器就是其应该定位到的服务器，并将该键值对存储在该节点上。 如我们有Object A、 Object B、object C、 object D四个数据对象，经过哈希计算后，在环空间上的位置如下:根据一致性Hash算法，数据A会被定为到Node A上，B被定为到Node B上，C被定为到Node C上，D被定为到Node D上。
优点
一致性哈希算法的容错性
假设Node c宕机，可以看到此时对象A、B、D不会受到影响。一般的，在一致性Hash算法中，如果一台服务器不可用，则受影响的数据仅仅是此服务器到其环空间中前一台服务器（即沿着逆时针方向行走遇到的第一台服务器)之间数据，其它不会受到影响。简单说，就是C挂了，受到影响的只是B、C之间的数据且这些数据会转移到D进行存储。
扩展性
数据量增加了，需要增加一台节点NodeX，X的位置在A和B之间，那收到影响的也就是A到X之间的数据，重新把A到X的数据录入到X上即可，不会导致hash取余全部数据重新洗牌。
缺点：一致性哈希算法的数据倾斜问题
一致性Hash算法在服务节点太少时，容易因为节点分布不均匀而造成数据倾斜（被缓存的对象大部分集中缓存在某一台服务器上）问题,例如系统中只有两台服务器:
小结
为了在节点数目发生改变时尽可能少的迁移数据
将所有的存储节点排列在首尾相接的Hash环上，每个key在计算Hash后会顺时针找到临近的存储节点存放。而当有节点加入或退出时仅影响该节点在Hash环上顺时针相邻的后续节点。
加入和删除节点只影响哈希环中顺时针方向的相邻的节点，对其他节点无影响。
数据的分布和节点的位置有关，因为这些节点不是均匀的分布在哈希环上的，所以数据在进行存储时达不到均匀分布的效果。
哈希槽分区(大厂)
HASH _SLOT = CRC16(key) mod 16384
哈希槽实质就是一个数组，数组[0,2^14-1]形成hash slot空间。
解决均匀分配的问题，在数据和节点之间又加入了一层，把这层称为哈希槽(slot)，用于管理数据和节点之间的关系，现在就相当于节点上放的是槽，槽里放的是数据。
槽解决的是粒度问题，相当于把粒度变大了，这样便于数据移动。哈希解决的是映射问题，使用key的哈希值来计算所在的槽，便于数据分配
一个集群只能有16384个槽，编号0-16383(0-2^14-1)。这些槽会分配给集群中的所有主节点，分配策略没有要求。 集群会记录节点和槽的对应关系，解决了节点和槽的关系后，接下来就需要对key求哈希值，然后对16384取模，余数是几key就落入对应的槽里。HASH_SLOT = CRC16(key) mod 16384。以槽为单位移动数据，因为槽的数目是固定的，处理起来比较容易，这样数据移动问题就解决了。
哈希槽的计算
Redis集群中内置了16384个哈希槽，redis 会根据节点数量大致均等的将哈希槽映射到不同的节点。当需要在Redis集群中放置一个key-valul时，redis先对key使用crc16算法算出一个结果然后用结果对16384求余数[ CRC16(key) % 16384]，这样每个key都会对应一个编号在0-16383之间的哈希槽，也就是映射到某个节点上。
为什么Redis集群的最大槽数是16384个？
CRC16算法产生的hash值有16bit，该算法可以产生2^16=65536个值。换句话说值是分布在0～65535之间，有更大的65536不用为什么只用16384就够?
答案连接：https://github.com/redis/redis/issues/2576
原因是：
正常的心跳数据包携带节点的完整配置，可以以幂等方式将其替换为旧配置以更新旧配置。这意味着它们包含原始形式的节点的插槽配置，该节点使用 2k 空间和 16k 插槽，但将使用 8k 空间使用 65k 插槽。
同时，由于其他设计权衡，Redis 集群不太可能扩展到超过 1000 个节点。
因此，16k 在正确的范围内，以确保每个主站有足够的插槽，最多 1000 个母节点，但数量足够小，可以轻松地将插槽配置作为原始位图传播。请注意，在小型集群中，位图很难压缩，因为当 N 较小时，位图将设置插槽/N 位，这是设置的位的很大百分比。
如果槽位为65536，发送心跳信息的消息头达8k，发送的心跳包过于庞大。
在消息头中最占空间的是myslots[CLUSTER_SLOTS/8]。当槽位为65536时，这块的大小是:65536➗8➗1024=8kb 在消息头中最占空间的是myslots[CLUSTER_SLOTS/8]。当槽位为16384时，这块的大小是:16384➗8➗1024=2kb 因为每秒钟，redis节点需要发送一定数量的ping消息作为心跳包，如果槽位为65536，这个ping消息的消息头太大了，浪费带宽。
redis的集群主节点数量基本不可能超过1000个。
集群节点越多，心跳包的消息体内携带的数据越多。如果节点过1000个，也会导致网络拥堵。因此redis作者不建议redis cluster节点数量超过1000个。那么，对于节点数在1000以内的redis cluster集群，16384个槽位够用了。没有必要拓展到65536个。
槽位越小，节点少的情况下，压缩比高，容易传输。
Redis主节点的配置信息中它所负责的哈希槽是通过一张bitmap的形式来保存的，在传输过程中会对bitmap进行压缩，但是如果bitmap的填充率slots/N很高的话(N表示节点数)，bitmap的压缩率就很低。如果节点数很少，而哈希槽数量很多的话，bitmap的压缩率就很低。
Redis集群不保证强一致性，这意味着在特定的条件下，Redis集群可能会丢掉一些被系统收到的写入请求命令。
2、集群环境案例
2.1、三主三从配置
3台真实虚拟机，各自新建 mkdir -p /myredis/cluster
新建6个独立的redis实例服务
按照这个改好端口号就行
bind 0.0.0.0
daemonize yes
protected-mode no
port 6379
logfile "/myRedis/cluster/cluster6379.log"
pidfile /myRedis/cluster6379.pid
dir /myRedis/cluster
dbfilename dump6379.rdb
appendonly yes
appendfilename "appendonly6379.aof"
requirepass 123456
masterauth 123456
cluster-enabled yes
cluster-config-file nodes-6379.conf
cluster-node-timeout 5000
启动这些实例
redis-server /myRedis/cluster/cluster6381.conf
.........
通过redis-cli命令为6台机器构建集群关系
redis-cli -a 123456 --cluster create --cluster-replicas 1 192.168.111.19:6379 192.168.111.19:6382 192.168.111.23:6380 192.168.111.23:6383 192.168.111.24:6381 192.168.111.23:6384这个连接的时候可能会因为防火墙导致无法连接到其他主机而报错
集群搭建成功会产生nodes的配置文件
连接进入6379作为切入点，查看并检验集群状态
CLUSTER NODES 查看节点关系拓扑图
CLUSTER INFO 获取Redis集群的信息
最终真实的主从关系分配图
2.2、3主3从集群读写
在6379中新写两个key，k1和k2
发现k1经过HASH _SLOT = CRC16(key) mod 16384算法算出它的槽位是12706，但是这个槽位是归给6381管的，你得移步到6381中写入。但是真实情况写，不可能我去一个个的找它在哪，我想实现的效果是我在一台服务器上随便设置的key可以自动的找到对应的槽位自动写入对应的服务器上。
那么就需要我们在连接redis服务时加一个-c 表示开启路由转发功能，当我在一个机器上写入不属于这个机器管辖的槽位的key时，它会自动重定向到这个槽位所属的服务器上写入。
redis-cli -a 123456 -p 6379 -c 连接集群的机器就要加上-c
查看某一个key应该属于对应的槽位值
CLUSTER KEYSLOT keyName
2.3、主从容错切换
把6379给挂掉，看它的从机6383能不能上位顶替它
答案是肯定的，6383接替了6379的master位置成为新的master，我们再重新将6379连接上线，发现6379成了6383的slave了
集群不保证数据一致性100%OK，一定会有数据丢失情况，Redis集群不保证强一致性，这意味着在特定的条件下，Redis集群可能会丢掉一些被系统收到的写入请求命令
手动故障转移or节点从属调整该如何处理？
现在的情况是6379和6383对换位置了，但跟原来分配的不一样，想要恢复6379的master身份怎么办？
CLUSTER FAILOVER 直接调换主从关系使用上面的命令后6379就回到master，6383回到slave
2.4、主从扩容
新建6385、6386两个服务实例配置文件+新建后启动
bind 0.0.0.0
daemonize yes
protected-mode no
port 6385
logfile "/myRedis/cluster/cluster6385.log"
pidfile /myRedis/cluster6385.pid
dir /myRedis/cluster
dbfilename dump6385.rdb
appendonly yes
appendfilename "appendonly6385.aof"
requirepass 123456
masterauth 123456
cluster-enabled yes
cluster-config-file nodes-6385.conf
cluster-node-timeout 5000启动后这两个节点还都是master，还没有被集群接纳。
将新增的6385节点(空槽号)作为master节点加入原集群
redis-cli -a 密码--cluster add-node 自己实际IP地址:6385 自己实际IP地址:6379
6385就是将要作为master新增节点 6379就是原来集群节点里面的领路人，相当于6385拜拜6379的码头从而找到组织加入集群
redis-cli -a 123456 --cluster add-node 192.168.111.24:6385 192.168.111.19:6379
检查集群情况
redis-cli -a 123456 --cluster check 192.168.111.19:6379
重新分派槽号(reshard )
redis-cli -a 123456 --cluster reshard 192.168.111.19:6379它问你要移动多少，就填4096，因为16384➗4=4096，4个人平均分。
然后将新master6385的id粘贴进去，之前的三个主机每人要为它匀点槽位直到四个各4096个槽位才分配完成
再次检查集群情况
redis-cli -a 123456 --cluster check 192.168.111.19:6379
发现6385的槽位是分散的三个一共凑出来的4096，这是前三个master每人拿出1364个槽位给6385成立新master
为主节点6385分配从节点6386
redis-cli -a 123456 --cluster add-node 192.168.111.24:6386 192.168.111.24:6385 --cluster-slave --cluster-master-id 203b88b869cefdc2d3efb8a1962173a548eca14b（6385的id）
再再次检查集群情况
2.5、主从缩容
Redis集群缩容，删除6385和6386，恢复3主3从
1先清除从节点6386
找到6386的id然后移除节点
redis-cli -a 123456 --cluster del-node 192.168.111.24:6386 216d84a6ac6bc86097a8733ba336c17eb76a88ad
检查一下，只剩下7台机器
2清出来的槽号重新分配给6379
将6385的槽号清空，重新分配，本例将清出来的槽号都给6379
找到6379和6385的id，让6385的槽位全部给6379
检查一下
3再删除6385
3、集群常用操作命令和CRC16算法分析
不在同一个slot槽位下的多键操作支持不好，通识占位符登场
不在同一个slot槽位下的键值无法使用mset、mget等多键操作
可以通过{来定义同一个组的概念，使key中(内相同内容的键值对放到一个slot槽位去，对照下图类似1k2k3都映射为x，自然槽位一样。
192.168.111.19:6379> mset k1{v} v1 k2{v} v2 k3{v} v3
-> Redirected to slot [7761] located at 192.168.111.23:6380
OK
192.168.111.23:6380> mget k1{v} k2{v} k3{v}
1) "v1"
2) "v2"
3) "v3"
192.168.111.23:6380> 
Redis集群有16384个哈希槽，每个key通过CRC16校验后对16384取模来决定放置哪个槽。集群的每个节点负责一部分hash槽
常用命令
集群是否完整才能对外提供服务
cluster-require-full-coverage
默认YES，现在集群架构是3主3从的redis cluster由3个master平分16384个slot，每个master的小集群负责1/3的slot，对应一部分数据。cluster-require-ful-coverage:默认值yes，即需要集群完整性，方可对外提供服务通常情况，如果这3个小集群中，任何一个(1主1从）挂了，你这个集群对外可提供的数据只有2/3了，整个集群是不完整的, redis默认在这种情况下，是不会对外提供服务的。
如果你的诉求是，集群不完整的话也需要对外提供服务，需要将该参数设置为no，这样的话你挂了的那个小集群是不行了，但是其他的小集群仍然可以对外提供服务。
CLUSTER COUNTKEYSINSLOT槽位数字编号
1，代表该槽位被占用
0，代表该槽位没被占用
CLUSTER KEYSLOT键名称
该键应该存在哪个槽位上
SpringBoot集成Redis
1.1、本地Java连接Redis常见问题
bind配置请注释掉
保护模式设置为no
Linux系统的防火墙设置，直接关了它
redis服务器的IP地址和密码是否正确
忘记写访问redis的服务端口号和auth密码
1.2、集成Jedis（老系统中比较多使用，了解）
Jedis Client是Redis官网推荐的一个面向java客户端，库文件实现宇对各类API进行封装调用.
建Module
redis7_study
改POM
添加Jedis 依赖
<dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>4.4.3</version> </dependency>写YML
改个端口
server.port=7777业务类
连接测试，向服务器写入key和读取key，基本的5个数据类型测试
测试类
package com.zm.test; import redis.clients.jedis.Jedis; import java.util.HashMap; import java.util.List; import java.util.Set; public class JedisDemo { public static void main(String[] args) { //1 通过指定ip和端口号获取连接 Jedis jedis = new Jedis("192.168.111.19", 6379); //2 指定访问服务器的密码 jedis.auth("123456"); //3 获取了jedis客户端，可以像jdbc一样连接redis服务器 System.out.println(jedis.ping()); //测试.string jedis.set("k1","hello-redis"); jedis.set("k2","hello-jedis"); jedis.set("k3","hello-v3"); Set<String> keys = jedis.keys("*"); System.out.println("keys = " + keys); System.out.println("这里是k1---》"+jedis.get("k1")); //设置k1的ttl jedis.expire("k1",20L); //使用mget获取key值 List<String> mget = jedis.mget("k1", "k2", "k3"); for (String s : mget) { System.out.println("mget =-----> " + s); } System.out.println("k1的ttl---->"+jedis.ttl("k1")); //list jedis.lpush("list","11","acd","fasdf","24dsd"); List<String> list = jedis.lrange("list", 0, -1); for (String s : list) { System.out.println("list = ---->" + s); } //mset和mget jedis.mset("k1","v11","k2","v22","k3","v33"); List<String> mget1 = jedis.mget("k1","k2","k3"); System.out.println("mget1------>"); mget1.forEach(System.out::println); //哈希hashes jedis.hset("user1","name","zs"); System.out.println("user1------>"+jedis.hget("user1", "name")); HashMap<String, String> map = new HashMap<String,String>(); map.put("name","lisi"); map.put("age","18"); map.put("sex","man"); jedis.hset("user2",map); System.out.println("user2------>"+jedis.hgetAll("user2")); //集合Sets // - SADD key member [member ...] 添加元素自动去重 // - SMEMBERS key 遍历集合中的所有元素 // - SISMEMBER key member 判断元素是否在集合中 jedis.sadd("jihe","1","1","3","2","4","5","2","3","9","7"); System.out.println("集合---->"+jedis.smembers("jihe")); System.out.println("是否在集合中---->"+jedis.sismember("jihe", "3")); } }结果：
到服务器上查看也是一样的结果
1.3、集成Iettuce（了解）
Jedis和Lettuce的区别
jedis和Lettuce都是Redis的客户端，它们都可以连接Redis服务器，但是在SpringBoot2.0之后默认都是使用的Lettuce这个客户端连接Redis服务器。因为当使用Jedis客户端连接Redis服务器的时候，每个线程都要拿自己创建的Jedis实例去连接Redis客户端，当有很多个线程的时候，不仅开销大需要反复的创建关闭一个Jedis连接，而且也是线程不安全的，一个线程通过Jedis实例更改Redis服务器中的数据之后会影响另一个线程;
但是如果使用Lettuce这个客户端连接Redis服务器的时候，就不会出现上面的情况，Lettuce底层使用的是Netty ,当有多个线程都需要连接Redis服务器的时候，可以保证只创建一个Lettuce连接，使所有的线程共享这一个Lettuce连接，这样可以减少创建关闭一个Lettuce连接时候的开销;而且这种方式也是线程安全的，不会出现一个线程通过Lettuce更改Redis服务器中的数据之后而影响另一个线程的情况;
添加依赖
<!-- https://mvnrepository.com/artifact/io.lettuce/lettuce-core -->
<dependency>
<groupId>io.lettuce</groupId>
<artifactId>lettuce-core</artifactId>
<version>6.2.1.RELEASE</version>
</dependency>写测试类
package com.zm.test;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import java.util.List;
public class LettuceDemo {
public static void main(String[] args) {
//1 使用构建器链式编程来build我们的RedisURI
RedisURI uri = RedisURI.builder()
.redis("192.168.111.19")
.withPort(6379)
.withAuthentication("default","123456")
.build();
//2 创建连接客户端
RedisClient client = RedisClient.create(uri);
StatefulRedisConnection<String, String> connect = client.connect();
//3 通过connect创建操作对象command
RedisCommands<String,String> commands = connect.sync();
//-------------------应用逻辑部分------------------
List keys = commands.keys("*");
System.out.println("**************keys-->"+keys);
commands.set("k4","hello-生菜");
System.out.println("****************k4"+commands.get("k4"));
//4 关闭释放各种资源
connect.close();
client.shutdown();
}
}
1.4、集成RedisTemplate-推荐使用（重点）
连接单机
先导依赖
<!--Redis-Template-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
<!--swagger2-->
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>2.9.2</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>2.9.2</version>
</dependency>application.properfiles
server.port=7777
#swagger2
spring.swagger2.enabled=true
spring.mvc.pathmatch.matching-strategy=ant_path_matcher
#redis单机
spring.redis.database=0
spring.redis.host=192.168.111.19
spring.redis.port=6379
spring.redis.password=123456
spring.redis.lettuce.pool.max-active=8
spring.redis.lettuce.pool.max-wait=-1ms
spring.redis.lettuce.pool.max-idle=8
spring.redis.lettuce.pool.min-idle=0配置类
RedisConfig（暂时先不写）
Swagger2Config配置
@Configuration
@EnableSwagger2
public class Swagger2Config {
@Value("${spring.swagger2.enabled}")
private Boolean enabled;
@Bean
public Docket api() {
return new Docket(DocumentationType.SWAGGER_2)
.apiInfo(apiInfo())
.enable(enabled)
.select()
.apis(RequestHandlerSelectors.basePackage("com.zm"))
.paths(PathSelectors.any())
.build();
}
private ApiInfo apiInfo() {
return new ApiInfoBuilder()
.title("springboot利用swagger2构建api接口文档")
.description("springboot+redis整合练习")
.version("1.0.0")
.contact(new Contact("ZM", "https://www.cnblogs.com/zm-111/", "3339332352@qq.com"))
.build();
}
}OrderService操作redis数据
@Service
public class OrderService {
//定义字符常量ord:，key的前缀
public static final String ORDER_KEY = "ord:";
//@Resource注解告诉Spring框架自动找到一个类型为RedisTemplate的bean，并将其注入到redisTemplate字段。
@Resource
private RedisTemplate redisTemplate;
public void addOrder(){
//+1生成1-100，不加0-99
int keyID = ThreadLocalRandom.current().nextInt(100)+1;
String serialNO = UUID.randomUUID().toString();
String key = ORDER_KEY + keyID;
String value = "京东订单" + serialNO;
//opsForValue()：用于获取一个操作字符串数据类型的对象。
redisTemplate.opsForValue().set(key,value);
System.out.println("key--->"+key);
System.out.println("value--->"+value);
}
public String getOrderByID(Integer keyID){
return (String) redisTemplate.opsForValue().get(ORDER_KEY+keyID);
}
}RedisController
@RestController
@Api(tags = "订单接口")
public class RedisController {
@Resource
private OrderService orderService;
@ApiOperation("新增订单")
@RequestMapping(value = "/order/add",method = RequestMethod.POST)
public void addOrder(){
orderService.addOrder();
}
@ApiOperation("通过keyID查询订单")
@RequestMapping(value = "/order/{keyID}",method = RequestMethod.GET)
public void getOrderByID(@PathVariable Integer keyID){
orderService.getOrderByID(keyID);
}
}先启动SpringBoot，访问Swagger UI
测试
执行addOrder
查看控制台
查看redis服务器中，key存进去了没有
get一下试试，有序列化
我们生成的key是ord:30，但是在redis里存储的却是一长串，get到的也是
使用我们写的测试
这样得出一个序列化问题
第一种解决方案
OrderService类中使用StringRedisTemplate换掉RedisTemplate
@Resource private StringRedisTemplate redisTemplate;RedisController中返回
@ApiOperation("通过keyID查询订单") @RequestMapping(value = "/order/{keyID}",method = RequestMethod.GET) public String getOrderByID(@PathVariable Integer keyID){ return orderService.getOrderByID(keyID); }启动SpringBoot再次生成一个
那么直接在丝袜哥上测试返回这个key的value
查看控制台
再到redis服务器上查看，key是正常的，但是value却还是老样子
要解决这个问题最简单粗暴得方法就是退出redis连接，然后再次连接时加上支持中文得选项
--raw即可
第二种解决方案
第一种解决方法太局限于String类型了，我们还有其他的数据类型它也不能处理。
所以还使用RedisTemplate
源码中对于RedisTemplate使用的都是默认的序列方式，JDK序列化方式
既然它默认使用的是JDK的序列化方式，那我们指定我们需要的序列化方式别让它使用默认的就行了。
这样我们就需要一个RedisConfig配置类，提前指定好需要的序列化方式好使用
@Configuration public class RedisConfig { //redis序列化的工具配置类 /* * this.redisTemplate.opsForvalue();//提供了操作string类型的所有方法 * this.redisTemplate.opsForList();/提供了操作list类型的所有方法 * this.redisTemplate.opsForset();1 /提供了操作set的所有方法 * this.redisTemplate.opsForHash();//提供了操作hash表的所有方法 * this.redisTemplate.opsForzSet();//提供了操作zset的所有方法*/ @Bean //LettuceConnectionFactory是RedisConnectionFactory接口的实例，封装了连接的获取方法，可以视作连接池。它用于建立Redis连接。 public RedisTemplate<String,Object> redisTemplate(LettuceConnectionFactory lettuceConnectionFactory){ RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>(); redisTemplate.setConnectionFactory(lettuceConnectionFactory); //设置key序列化方式为string redisTemplate.setKeySerializer(new StringRedisSerializer()); //设置value的序列化方式为json，使用GenericJackson2jsonRedisSerializer替换默认序列化 redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer()); return redisTemplate; } }现在启动服务再测试一下看是否和原来使用StringRedisTemplate一样都不会再乱码
查看后台
去redis服务器查看
连接集群
启动redis集群6台实例
配置文件application.properfile改写成集群连接
#redis集群
#获取失败最大重定向次数
spring.redis.cluster.max-redirects= 3
spring.redis.lettuce.pool.max-active=8
spring.redis.lettuce.pool.max-wait=-1ms
spring.redis.lettuce.pool.max-idle=8
spring.redis.lettuce.pool.min-idle=0
spring.redis.cluster.nodes=192.168.111.25:6379,192.168.111.25:6382,192.168.111.23:6380,192.168.111.23:6383,192.168.111.24:6381,192.168.111.24:6384连接测试
到redis集群中查看，一切正常没有问题
问题：
人为模拟，master-6379机器意外宕机，手动shutdown，6383上位成功
先对redis集群命令方式，手动验证各种读写命令。
随便写一下看集群读写是否正常
Redis Cluster集群能自动感知并自动完成主备切换，对应的slave6384会被选举为新的master
微服务客户端再次读写访问
我们在丝袜哥进行测试，发现它会一直转圈，但最后也成功了
控制台
集群中查看，也很正常
看着都正常，但是它转圈圈就转了好长时间，在生产中在这里卡一会儿在客户端就卡死，所以，SpringBoot客户端没有动态感知到RedisCluster的最新集群信息造成卡顿
那么怎么解决这个SpringBoot不能动态感知最新集群信息的问题？
1排除lettuce采用jedis(不推荐)
2重写连接工厂实例（极度不推荐）
3刷新节点集群拓扑动态感应（就是这个官方推荐）
Lettuce 处理Moved和Ask永久重定向，由于命令重定向，必须刷新节点拓扑视图。而自适应拓扑刷新(Adaptiveupdates）与定时拓扑刷新(Periodic updates）默认关闭
调用 RedisClusterClient.reloadPartitions
后台基于时间间隔的周期刷新
后台基于持续的断开和移动、重定向的自适应更新
修改配置文件application.properfile
添加以下内容
#支持集群拓扑动态感应刷新,自适应拓扑刷新是否使用所有可用的更新，默认false关闭 spring.redis.lettuce.cluster.refresh.adaptive=true #定时刷新2秒 spring.redis.lettuce.cluster.refresh.period=2000重启测试，这次转圈变快了
没错成功