Redis--高性能分布式缓存

缓存原理&设计

缓存基本思想

在互联网技术中，缓存是系统快速响应的关键技术之一以空间换时间的一种技术(艺术)

缓存的使用场景
- DB缓存，减轻DB服务器压力
  一般情况下数据存在数据库中，应用程序直接操作数据库。
  当访问量上万，数据库压力增大，可以采取的方案有:
  读写分离，分库分表
  当访问量达到10万、百万，需要引入缓存。将已经访问过的内容或数据存储起来，当再次访问时先找缓存，缓存命中返回数据。不命中再找数据库，并回填缓存。
- 提高系统响应数据库的数据是存在文件里，也就是硬盘。与内存做交换(swap) 在大量瞬间访问时(高并发)MySQL单机会因为频繁IO而造成无法响应。MySQL的InnoDB是有行锁将数据缓存在Redis中，也就是存在了内存中。
  内存天然支持高并发访问。可以瞬间处理大量请求。
  qps到达11万/S读请求 8万写/S
- 做Session分离
  传统的session是由tomcat自己进行维护和管理。集群或分布式环境，不同的tomcat管理各自的session。只能在各个tomcat之间，通过网络和Io进行session的复制，极大的影响了系统的性能。
  1、各个Tomcat间复制session，性能损耗
  2、不能保证各个Tomcat的Session数据同步
  将登录成功后的Session信息，存放在Redis中，这样多个服务器(Tomcat)可以共享Session信息。 Redis的作用是数据的临时存储
- 做分布式锁(Redis) 一般讲锁是多线程的锁，是在一个进程中的多个进程(JVM)在并发时也会产生问题，也要控制时序性可以采用分布式锁。使用Redis实现 sexNX, zookeeper也能做
- 做乐观锁(Redis)
  - 同步锁和数据库中的行锁、表锁都是悲观锁悲观锁的性能是比较低的，响应性比较差
  - 高性能、高响应(秒杀)采用乐观锁 (CAS)
  - Redis可以实现乐观锁 watch + incr

使用缓存的优势

提升用户体验用户体验(User Experience):用户在使用产品过程中建立起来的一种纯主观感受。缓存的使用可以提升系统的响应能力，大大提升了用户体验。
减轻服务器压力客户端缓存、网络端缓存减轻应用服务器压力。
服务端缓存减轻数据库服务器的压力。
提升系统性能系统性能指标:响应时间、延迟时间、吞吐量、并发用户数和资源利用率等。
缓存技术可以:
缩短系统的响应时间
减少网络传输时间和应用延迟时间
提高系统的吞吐量
增加系统的并发用户数
提高了数据库资源的利用率

使用缓存的代价

额外的硬件支出缓存是一种软件系统中以空间换时间的技术
需要额外的磁盘空间和内存空间来存储数据
搭建缓存服务器集群需要额外的服务器
采用云服务器的缓存服务就不用额外的服务器了
高并发缓存失效在高并发场景下会出现缓存失效(缓存穿透、缓存雪崩、缓存击穿)
造成瞬间数据库访问量增大，甚至崩溃
缓存与数据库数据同步缓存与数据库无法做到数据的时时同步
Redis无法做到主从时时数据同步
缓存并发竞争多个redis的客户端同时对一个key进行set值得时候由于执行顺序引起的并发问题

缓存的读写模式

Cache Aside Pattern(常用)
- 读请求 Cache Aside Pattern(旁路缓存)，是最经典的缓存+数据库读写模式。读的时候，先读缓存，缓存没有的话，就读数据库，然后取出数据后放入缓存，同时返回响应。
- 写请求更新的时候，先更新数据库，然后再删除缓存。
  
  为什么是删除缓存，而不是更新缓存呢?
  1、缓存的值是一个结构:hash、list，更新数据需要遍历, 先遍历(耗时)后修改
  2、懒加载，使用的时候才更新缓存使用的时候才从DB中加载也可以采用异步的方式填充缓存
  开启一个线程定时将DB的数据刷到缓存中
- 高并发脏读
  - 先更新数据库，再更新缓存 update与commit之间，更新缓存，commit失败, 则DB与缓存数据不一致
  - 先删除缓存，再更新数据库 update与commit之间，有新的读，缓存空，读DB数据到缓存数据是旧的数据
    commit后 DB为新数据, 则DB与缓存数据不一致
  - 先更新数据库，再删除缓存(推荐) update与commit之间，有新的读，缓存空，读DB数据到缓存数据是旧的数据 commit后 DB为新数据
    则DB与缓存数据不一致
    采用延时双删策略
Read/Write Through Pattern 应用程序只操作缓存，缓存操作数据库。
Read-Through(穿透读模式/直读模式):应用程序读缓存，缓存没有，由缓存回源到数据库，并写入
缓存。(guavacache)
Write-Through(穿透写模式/直写模式):应用程序写缓存，缓存写数据库。

该种模式需要提供数据库的handler，开发较为复杂。
Write Behind Caching Pattern 应用程序只更新缓存。
缓存通过异步的方式将数据批量或合并后更新到DB中
不能时时同步，甚至会丢数据

Redis基础

Redis介绍

什么是Redis
- redies(Remote Dictionary Server)远程字典服务器，是用C语言开发的一个开源的高性能键值对( key-value )内存数据库。
- 它提供了五种数据类型来存储值:字符串类型、散列类型、列表类型、集合类型、有序集合类型
- 它是一种 NoSQL 数据存储。
Redis应用场景
- 缓存使用，减轻DB压力
- DB使用，用于临时存储数据(字典表，购买记录)
- 解决分布式场景下Session分离问题(登录信息)
- 任务队列(秒杀、抢红包等等) 乐观锁
- 应用排行榜 zset
- 签到 bitmap
- 分布式锁
- 冷热数据交换

Redis单机版安装和使用

官网地址:http://redis.io/
中文官网地址:http://www.redis.cn/
下载地址:http://download.redis.io/releases/

Redis安装

c++编译器

yum  install -y gcc-c++  
yum install -y wget  

下载并解压缩 Redis 源码压缩包

wget http://download.redis.io/releases/redis-5.0.5.tar.gz  
tar -zxf redis-5.0.5.tar.gz  

编译redis源码进入redis-5.0.5目录，执行编译命令
```
cd redis-5.0.5/src  
make  
```

安装 Redis 需要通过 PREFIX 指定安装路径

mkdir  /usr/redis -p  
make   install PREFIX=/usr/redis  

Redis启动

前端启动
启动命令: redis-server ，直接运行 bin/redis-server 将以前端模式启动
```
./redis-server  
```

后端启动(守护进程启动)

拷贝 redis-5.0.5/redis.conf
```
cp  redis.conf /usr/redis/bin/  
```

修改 redis.conf

vim redis.conf

		    
# 默认绑定的是回环地址，默认不能被其他机器访问   
# bind 127.0.0.1  
		    
# 是否开启保护模式，由yes该为no protected-mode no  
		    
# 将`daemonize`由`no`改为`yes` daemonize yes

启动服务
```
./redis-server redis.conf  
```
停止服务
```
./redis-cli shutdown  
```

命令说明
- redis-server 启动redis服务
- redis-cli 进入redis命令客户端
  - redis-cli -h host -p port 默认端口6379
- redis-benchmark 性能测试的工具
- redis-check-aof aof 文件进行检查的工具
- redis-check-dump rdb 文件进行检查的工具
- redis-sentinel 启动哨兵监控服务

Redis数据类型

Redis是一个Key-Value的存储系统，使用ANSI C语言编写。 key的类型是字符串。
value的数据类型有:
常用的:string字符串类型、list列表类型、set集合类型、sortedset(zset)有序集合类型、hash类型。
不常见的:bitmap位图类型、geo地理位置类型。
Redis5.0新增一种:stream类型

注意:Redis中命令是忽略大小写，(set SET)，key是不忽略大小写的 (NAME name)

Redis的Key的设计
1. 用:分割
2. 把表名转换为key前缀, 比如: user: 3. 第二段放置主键值
3. 第三段放置列名
user:9:username 表示user表里边id为8，column name = username
常用类型
- string字符串类型
  Redis的String能表达3种值的类型:字符串、整数、浮点数 100.01 是个六位的串
  - 应用场景 1、key和命令是字符串
    2、普通的赋值
    3、incr用于乐观锁 incr:递增数字，可用于实现乐观锁 watch(事务)
    4、setnx用于分布式锁当value不存在时采用赋值，可用于实现分布式锁
- list列表类型
  list列表类型可以存储有序、可重复的元素
  获取头部或尾部附近的记录是极快的
  list的元素个数最多为2^32-1个(40亿)
  - 应用场景 1、作为栈或队列使用
    列表有序可以作为栈和队列使用
    2、可用于各种列表，比如用户列表、商品列表、评论列表等。
- set集合类型
  Set:无序、唯一元素
  集合中最大的成员数为 2^32 - 1
  - 应用场景适用于不能重复的且不需要顺序的数据结构
    比如:关注的用户，还可以通过spop进行随机抽奖
- sortedset有序集合类型
  SortedSet(ZSet) 有序集合: 元素本身是无序不重复的
  每个元素关联一个分数(score)
  可按分数排序，分数可重复
  - 应用场景由于可以按照分值排序，所以适用于各种排行榜。比如:点击排行榜、销量排行榜、关注排行榜等。
- hash类型
  Redis hash 是一个 string 类型的 field 和 value 的映射表，它提供了字段和字段值的映射。
  每个 hash 可以存储 2^32 - 1 键值对(40多亿)。
  - 应用场景对象的存储，表数据的映射

不常用类型

bitmap位图类型
bitmap是进行位操作的通过一个bit位来表示某个元素对应的值或者状态,其中的key就是对应元素本身。
bitmap本身会极大的节省储存空间。
- 应用场景 1、用户每月签到，用户id为key ，日期作为偏移量 1表示签到
  2、统计活跃用户, 日期为key，用户id为偏移量 1表示活跃
  3、查询用户在线状态，日期为key，用户id为偏移量 1表示在线
geo地理位置类型 geo是Redis用来处理位置信息的。在Redis3.2中正式使用。主要是利用了Z阶曲线、Base32编码和geohash算法
- Z阶曲线
  在x轴和y轴上将十进制数转化为二进制数，采用x轴和y轴对应的二进制数依次交叉后得到一个六位数编码。把数字从小到大依次连起来的曲线称为Z阶曲线，Z阶曲线是把多维转换成一维的一种方法。
- Base32编码
  Base32这种数据编码机制，主要用来把二进制数据编码成可见的字符串，其编码规则是:任意给定一个二进制数据，以5个位(bit)为一组进行切分(base64以6个位(bit)为一组)，对切分而成的每个组进行编码得到1个可见字符。Base32编码表字符集中的字符总数为32个(0-9、b-z去掉a、i、l、o)，这也是 Base32名字的由来。
- geohash算法
  Gustavo在2008年2月上线了geohash.org网站。Geohash是一种地理位置信息编码方法。经过 geohash映射后，地球上任意位置的经纬度坐标可以表示成一个较短的字符串。可以方便的存储在数据库中，附在邮件上，以及方便的使用在其他服务中
  
  Redis中经纬度使用52位的整数进行编码，放进zset中，zset的value元素是key，score是GeoHash的 52位整数值。在使用Redis进行Geo查询时，其内部对应的操作其实只是zset(skiplist)的操作。通过zset 的score进行排序就可以得到坐标附近的其它元素，通过将score还原成坐标值就可以得到元素的原始坐标。
- 应用场景 1、记录地理位置
  2、计算距离
  3、查找”附近的人”
- 常用指令
```
127.0.0.1:6379> geoadd user:addr 116.3140.05 zhangf 116.38 39.88 zhaoyun 116.47 40.00 diaochan #添加用户地址zhangf、zhaoyun、 diaochan的经纬度  
(integer)3  
127.0.0.1:6379> geohash user:addr zhangf diaochan #获得zhangf和diaochan的geohash码  
1) "wx4eydyk5m0"  
2) "wx4gd3fbgs0"  
127.0.0.1:6379> geopos user:addr zhaoyun #获得zhaoyun的经纬度  
1)1)“116.38000041246414185"  
  2) "39.88000114172373145"  
127.0.0.1:6379> geodist user:addr zhangf diaochan #计算zhangf到diaochan的距离,单位是m  
"14718.6972"  
127.0.0.1:6379> geodist user:addr zhangf diaochan km #计算zhangf到diaochan的距离，单位是km  
"14.7187"  
127.0.0.1:6379> geodist user:addr zhangf zhaoyun km  
"19.8276"  
127.0.0.1:6379> georadiusbymember user:addr zhangf 20 km withcoord withdist count 3 asc  
#获得距离zhangf20km以内的按由近到远的顺序排出前三名的成员名称、距离及经纬度  
#withcoord:获得经纬度 withdist:获得距离 withhash:获得geohash码  
1)1) "zhangf"  
  2)“0.0000"  
  3)1)"116.31000012159347534"  
    2)"40.04999982043828055  
2)1) "diaochan"  
  2)“14.7187"  
  3)1)“116.46999925374984741"  
    2)“39.99999991084916218"  
3)1) "zhaoyun"  
  2)"19.8276"  
  3)1)“116.38000041246414185"  
    2)“39.88000114172373145"  
```

stream数据流类型
stream是Redis5.0后新增的数据结构，用于可持久化的消息队列。几乎满足了消息队列具备的全部内容，包括:

消息ID的序列化生成
消息遍历
消息的阻塞和非阻塞读取
消息的分组消费
未完成消息的处理

消息队列监控
每个Stream都有唯一的名称，它就是Redis的key，首次使用 xadd 指令追加消息时自动创建。

消息队列的使用

127.0.0.1:6379> xadd topic:001 * name zhangefei age 23  
"1613470234204-0"  
127.0.0.1:6379> xadd topic:001 * name zhaoyun age 24 name diaochan age 16  
"1613470291836-0"  
127.0.0.1:6379> xrange topic:001 - +  
1) 1) "1613470234204-0"  
   2) 1) "name"  
      2) "zhangefei"  
      3) "age"  
      4) "23"  
2) 1) "1613470291836-0"  
   2) 1) "name"  
      2) "zhaoyun"  
      3) "age"  
      4) "24"  
      5) "name"  
      6) "diaochan"  
      7) "age"  
      8) "16"  
127.0.0.1:6379> xrange topic:001 + -  
(empty list or set)  
127.0.0.1:6379> xread STREAMS topic:001 1613470291836-0  
(nil)  
127.0.0.1:6379> xread COUNT 1 streams topic:001 0  
1) 1) "topic:001"  
   2) 1) 1) "1613470234204-0"  
         2) 1) "name"  
            2) "zhangefei"  
            3) "age"  
            4) "23"  
127.0.0.1:6379> xread  streams topic:001 0  
1) 1) "topic:001"  
   2) 1) 1) "1613470234204-0"  
         2) 1) "name"  
            2) "zhangefei"  
            3) "age"  
            4) "23"  
      2) 1) "1613470291836-0"  
         2) 1) "name"  
            2) "zhaoyun"  
            3) "age"  
            4) "24"  
            5) "name"  
            6) "diaochan"  
            7) "age"  
            8) "16"  
# 创建的group1              
127.0.0.1:6379> xgroup create topic:001 group1 0  
OK  
# 创建cus1加入到group1 消费 没有被消费过的消息 消费第一条  
127.0.0.1:6379> xreadgroup group group1 cus1 count 1 streams topic:001 >  
1) 1) "topic:001"  
   2) 1) 1) "1613470234204-0"  
         2) 1) "name"  
            2) "zhangefei"  
            3) "age"  
            4) "23"  
# 继续消费 第二条  
127.0.0.1:6379> xreadgroup group group1 cus1 count 1 streams topic:001 >  
1) 1) "topic:001"  
   2) 1) 1) "1613470291836-0"  
         2) 1) "name"  
            2) "zhaoyun"  
            3) "age"  
            4) "24"  
            5) "name"  
            6) "diaochan"  
            7) "age"  
            8) "16"  
# 没有可消费  
127.0.0.1:6379> xreadgroup group group1 cus1 count 1 streams topic:001 >  
(nil)  

Redis常用命令

官方命令大全网址:http://www.redis.cn/commands.html

keys 返回满足给定pattern 的所有key

redis 127.0.0.1:6379> keys list*  
1) "list"  
2) "list5"  
3) "list6"  
4) "list7"  
5) "list8"  
    
keys *  

del
```
# 删除key  
DEL key  
```
exists 确认一个key 是否存在

expire Redis在实际使用过程中更多的用作缓存，然而缓存的数据一般都是需要设置生存时间的，即:到期后数据销毁。

EXPIRE key seconds  
设置key的生存时间(单位:秒)key在多少秒后会自动删除  
	    
TTL key  
查看key生于的生存时间  
	    
PERSIST key  
清除生存时间  
	    
PEXPIRE key milliseconds生存时间设置单位为:毫秒  

rename

# 重命名key  
rename  oldkey  newkey  

type 显示指定key的数据类型

Redis的Java客户端—Jedis

新建maven项目后导入Jedis包

<dependency>  
    <groupId>redis.clients</groupId>  
    <artifactId>jedis</artifactId>  
    <version>2.9.0</version>  
</dependency>  

示例代码 api基本cli一模一样

@Test  
public void testConn(){  
	//与Redis建立连接 IP+port  
	Jedis redis = new 	Jedis("192.168.127.128", 6379);   
	//在Redis中写字符串 key value 	  
	redis.set("jedis:name:1","jd-zhangfei");   
	//获得Redis中字符串的值 	  
System.out.println(redis.get("jedis:name:1"));   
	//在Redis中写list 	  
	redis.lpush("jedis:list:1","1","2","3","4","5");   
	//获得list的长度 	  
	System.out.println(redis.llen("jedis:list:1"));  
}  

缓存过期和淘汰策略

Redis性能高:
官方数据
读:110000次/s
写:81000次/s

长期使用，key会不断增加，Redis作为缓存使用，物理内存也会满
内存与硬盘交换(swap) 虚拟内存，频繁IO 性能急剧下降
maxmemory
- 不设置的场景
  - Redis的key是固定的，不会增加
  - Redis作为DB使用，保证数据的完整性，不能淘汰，可以做集群，横向扩展
  - 缓存淘汰策略:禁止驱逐 (默认)
- 设置的场景 Redis是作为缓存使用，不断增加Key
  maxmemory : 默认为0 不限制
  问题:达到物理内存后性能急剧下架，甚至崩溃
  内存与硬盘交换(swap) 虚拟内存，频繁IO 性能急剧下降
  设置多少?
  与业务有关
  1个Redis实例，保证系统运行 1 G ，剩下的就都可以设置Redis 物理内存的3/4
  
  slaver : 留出一定的内存
- 设置在redis.conf中
```
maxmemory   1024mb  
```
  命令: 获得maxmemory数
```
CONFIG GET maxmemory  
```
  设置maxmemory后，当趋近maxmemory时，通过缓存淘汰策略，从内存中删除对象
  
  不设置maxmemory 无最大内存限制 maxmemory-policy noeviction (禁止驱逐) 不淘汰
  设置maxmemory maxmemory-policy 要配置
expire数据结构在Redis中可以使用expire命令设置一个键的存活时间(ttl: time to live)，过了这段时间，该键就会自动被删除。
- expire原理
```
typedef struct redisDb {  
    dict *dict;  -- key Value  
    dict *expires; -- key ttl  
    dict *blocking_keys;  
    dict *ready_keys;  
    dict *watched_keys;  
    int id;  
} redisDb;  
```
  上面的代码是Redis 中关于数据库的结构体定义，这个结构体定义中除了 id 以外都是指向字典的指针，其中我们只看 dict 和 expires。
  dict 用来维护一个 Redis 数据库中包含的所有 Key-Value 键值对，expires则用于维护一个 Redis 数据库中设置了失效时间的键(即key与失效时间的映射)。
  当我们使用 expire命令设置一个key的失效时间时，Redis 首先到 dict 这个字典表中查找要设置的key 是否存在，如果存在就将这个key和失效时间添加到 expires 这个字典表。
  当我们使用 setex命令向系统插入数据时，Redis 首先将 Key 和 Value 添加到 dict 这个字典表中，然后将 Key 和失效时间添加到 expires 这个字典表中。
  简单地总结来说就是，设置了失效时间的key和具体的失效时间全部都维护在 expires 这个字典表中。
删除策略 Redis的数据删除有定时删除、惰性删除和主动删除三种方式。 Redis目前采用惰性删除+主动删除的方式。
- 定时删除在设置键的过期时间的同时，创建一个定时器，让定时器在键的过期时间来临时，立即执行对键的删除
  操作。
  需要创建定时器，而且消耗CPU，一般不推荐使用。
- 惰性删除在key被访问时如果发现它已经失效，那么就删除它。调用expireIfNeeded函数，该函数的意义是:读取数据之前先检查一下它有没有失效，如果失效了就删除它。
```
int expireIfNeeded(redisDb *db, robj *key) {  
//获取主键的失效时间 get当前时间-创建时间>ttl  
long long when = getExpire(db,key); //假如失效时间为负数，说明该主键未设置失效时间(失效时间默认为-1)，直接返回0 if (when < 0) return 0; //假如Redis服务器正在从RDB文件中加载数据，暂时不进行失效主键的删除，直接返回0 if (server.loading) return 0;  
...  
//如果以上条件都不满足，就将主键的失效时间与当前时间进行对比，如果发现指定的主键 //还未失效就直接返回0  
if (mstime() <= when) return 0; //如果发现主键确实已经失效了，那么首先更新关于失效主键的统计个数，然后将该主键失 //效的信息进行广播，最后将该主键从数据库中删除  
    server.stat_expiredkeys++;  
    propagateExpire(db,key);  
    return dbDelete(db,key);  
}  
```
- 主动删除在redis.conf文件中可以配置主动删除策略,默认是no-enviction(不删除)
```
maxmemory-policy allkeys-lru  
```
  - LRU LRU (Least recently used) 最近最少使用，算法根据数据的历史访问记录来进行淘汰数据，其核心思想是“如果数据最近被访问过，那么将来被访问的几率也更高”。
    
    最常见的实现是使用一个链表保存缓存数据，详细算法实现如下:
    1. 新数据插入到链表头部;
    2. 每当缓存命中(即缓存数据被访问)，则将数据移到链表头部;
    3. 当链表满的时候，将链表尾部的数据丢弃。
    4. 在Java中可以使用LinkHashMap(哈希链表)去实现LRU
    - Redis的LRU 数据淘汰机制 struct redisObject 中可以发现，每一个 redis 对象都会设置相应的 lru。可以想象的是，每一次访问数据的时候，会更新 redisObject.lru。
      typedef struct redisObject { unsigned type:4; unsigned encoding:4; unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or * LFU data (least significant 8 bits frequency * and most significant 16 bits access time). */ int refcount; void *ptr; } robj;
      LRU 数据淘汰机制是这样的:在数据集中随机挑选几个键值对，取出其中 lru 最大的键值对淘汰。
      不可能遍历key 用当前时间-最近访问越大说明访问间隔时间越长
      - volatile-lru 从已设置过期时间的数据集(server.db[i].expires)中挑选最近最少使用的数据淘汰
      - allkeys-lru 从数据集(server.db[i].dict)中挑选最近最少使用的数据淘汰
  - LFU LFU (Least frequently used) 最不经常使用，如果一个数据在最近一段时间内使用次数很少，那么在将来一段时间内被使用的可能性也很小。
    - volatile-lfu 从已设置过期时间的数据集(server.db[i].expires)中挑选最近使用频率最低的数据淘汰
    - allkeys-lfu 从数据集(server.db[i].dict)中挑选最近不常使用的数据淘汰
  - random
    - volatile-random 从已设置过期时间的数据集(server.db[i].expires)中任意选择数据淘汰
    - allkeys-random 从数据集(server.db[i].dict)中任意选择数据淘汰
  - ttl
    - volatile-ttl 从已设置过期时间的数据集(server.db[i].expires)中挑选将要过期的数据淘汰
      redis 数据集数据结构中保存了键值对过期时间的表，即 redisDb.expires。
      TTL 数据淘汰机制:从过期时间的表中随机挑选几个键值对，取出其中 ttl 最小的键值对淘汰。
  - noenviction 禁止驱逐数据，不删除 (默认)
  - 缓存淘汰策略的选择
    - allkeys-lru : 在不确定时一般采用策略。冷热数据交换
    - volatile-lru : 比allkeys-lru性能差
      存 : 过期时间
    - allkeys-random : 希望请求符合平均分布(每个元素以相同的概率被访问)
    - 自己控制:volatile-ttl 缓存穿透

Redis扩展特性

发布与订阅

Redis提供了发布订阅功能，可以用于消息的传输
Redis的发布订阅机制包括三个部分，publisher，subscriber和Channel

发布者和订阅者都是Redis客户端，Channel则为Redis服务器端。
发布者将消息发送到某个的频道，订阅了这个频道的订阅者就能接收到这条消息。

api
- SUBSCRIBE channel [channel …]
- PUBLISH channel message
- UNSUBSCRIBE [channel [channel …]]
- PSUBSCRIBE pattern [pattern …]
- PUNSUBSCRIBE [pattern [pattern …]]
发布订阅的机制订阅某个频道或模式:
客户端(client):
属性为pubsub_channels，该属性表明了该客户端订阅的所有频道
属性为pubsub_patterns，该属性表示该客户端订阅的所有模式

服务器端(RedisServer):
属性为pubsub_channels，该服务器端中的所有频道以及订阅了这个频道的客户端
属性为pubsub_patterns，该服务器端中的所有模式和订阅了这些模式的客户端
```
typedef struct redisClient {  
    ...  
dict *pubsub_channels; //该client订阅的channels，以channel为key用dict的方式组织 list *pubsub_patterns; //该client订阅的pattern，以list的方式组织  
...  
} redisClient;  
struct redisServer {  
    ...  
dict *pubsub_channels; 为key，订  
//redis server进程中维护的channel dict，它以channel 阅channel的client list为value  
//redis server进程中维护的pattern list  
    list *pubsub_patterns;  
    int notify_keyspace_events;  
    ...  
};  
```
当客户端向某个频道发送消息时，Redis首先在redisServer中的pubsub_channels中找出键为该频道的结点，遍历该结点的值，即遍历订阅了该频道的所有客户端，将消息发送给这些客户端。

然后，遍历结构体redisServer中的pubsub_patterns，找出包含该频道的模式的结点，将消息发送给订阅了该模式的客户端。
使用场景
- 哨兵模式在Redis哨兵模式中，哨兵通过发布与订阅的方式与Redis主服务器和Redis从服务器进行通信
- Redisson框架使用 Redisson是一个分布式锁框架，在Redisson分布式锁释放的时候，是使用发布与订阅的方式通知的

事务

所谓事务(Transaction) ，是指作为单个逻辑工作单元执行的一系列操作

ACID
- Atomicity(原子性): Redis:一个队列中的命令执行或不执行
- Consistency(一致性): Redis: 集群中不能保证时时的一致性，只能是最终一致性
- Isolation(隔离性): Redis: 命令是顺序执行的，在一个事务中，有可能被执行其他客户端的命令的
- Durability(持久性): Redis有持久化但不保证数据的完整性

Redis事务

Redis的事务是通过multi、exec、discard和watch这四个命令来完成的。
Redis的单个命令都是原子性的，所以这里需要确保事务性的对象是命令集合。
Redis将命令集合序列化并确保处于同一事务的命令集合连续且不被打断的执行

Redis不支持回滚操作

事务命令
- multi 用于标记事务块的开始,Redis会将后续的命令逐个放入队列中，然后使用exec原子化地执行这个命令队列
- exec 执行命令队列
- discard 清除命令队列
- watch 监视key
- unwatch 清除监视key

示例代码

127.0.0.1:6379> multi  
OK  
127.0.0.1:6379> set s1 222  
QUEUED  
127.0.0.1:6379> hset set1 name zhangfei  
QUEUED  
127.0.0.1:6379> exec  
1) OK  
	    
2) (integer) 1  
127.0.0.1:6379> multi  
OK  
127.0.0.1:6379> set s2 333  
QUEUED  
127.0.0.1:6379> hset set2 age 23  
QUEUED  
127.0.0.1:6379> discard  
OK  
127.0.0.1:6379> exec  
(error) ERR EXEC without MULTI  
	    
# watch 之后可以在事务外改这个key值，让multi-exec失效  
127.0.0.1:6379> watch s1  
OK  
127.0.0.1:6379> multi  
OK  
127.0.0.1:6379> set s1 555 QUEUED  
# 此时在没有exec之前，通过另一个命令窗口对监控的s1字段进行修改  
127.0.0.1:6379> exec   
(nil)  
127.0.0.1:6379> get s1  
222  
127.0.0.1:6379> unwatch  
OK  

事务的执行

typedef struct redisClient{  
  // flags  
int flags //状态  
// 事务状态   
multiState mstate;   
// .....  
}redisClient;  
	    
// 事务状态  
typedef struct multiState{  
// 事务队列,FIFO顺序  
// 是一个数组,先入队的命令在前,后入队在后   
multiCmd *commands;  
// 已入队命令数  
int count;  
}multiState;  
	    
// 事务队列  
typedef struct multiCmd{  
// 参数  
robj **argv;  
// 参数数量  
int argc;  
// 命令指针  
struct redisCommand *cmd;  
}multiCmd;  

事务开始在RedisClient中，有属性flags，用来表示是否在事务中 flags=REDIS_MULTI
命令入队 RedisClient将命令存放在事务队列中 (EXEC,DISCARD,WATCH,MULTI除外)
事务队列 multiCmd *commands 用于存放命令
执行事务 RedisClient向服务器端发送exec命令，RedisServer会遍历事务队列,执行队列中的命令,最后将执行的结果一次性返回给客户端。

如果某条命令在入队过程中发生错误，redisClient将flags置为REDIS_DIRTY_EXEC，EXEC命令将会失败返回。

Watch的执行
```
typedef struct redisDb{  
    // .....  
// 正在被WATCH命令监视的键 dict *watched_keys;  
// .....  
}redisDb;  
```
- 使用WATCH命令监视数据库键 redisDb有一个watched_keys字典,key是某个被监视的数据的key,值是一个链表.记录了所有监视这个数据的客户端。
- 监视机制的触发当修改数据后，监视这个数据的客户端的flags置为REDIS_DIRTY_CAS
- 事务执行 RedisClient向服务器端发送exec命令，服务器判断RedisClient的flags，如果为REDIS_DIRTY_CAS，则清空事务队列。

Redis的弱事务性

Redis语法错误整个事务的命令在队列里都清除

127.0.0.1:6379> multi  
OK  
127.0.0.1:6379> sets m1 44  
(error) ERR unknown command `sets`, with args beginning with: `m1`, `44`,  
127.0.0.1:6379> set m2 55  
QUEUED  
127.0.0.1:6379> exec  
(error) EXECABORT Transaction discarded because of previous errors.  
127.0.0.1:6379> get m1  
"22"  

flags=REDIS_DIRTY_EXEC

Redis运行错误在队列里正确的命令可以执行 (弱事务性) 弱事务性 :
1、在队列里正确的命令可以执行 (非原子操作)
2、不支持回滚
```
127.0.0.1:6379> multi  
OK  
127.0.0.1:6379> set m1 55  
QUEUED  
127.0.0.1:6379> lpush m1 1 2 3 #不能是语法错误 QUEUED  
127.0.0.1:6379> exec  
1) OK  
2) (error) WRONGTYPE Operation against a key holding the wrong kind of value  
127.0.0.1:6379> get m1  
"55"  
```
Redis不支持事务回滚(为什么呢)
1、大多数事务失败是因为语法错误或者类型错误，这两种错误，在开发阶段都是可以预见的
2、Redis为了性能方面就忽略了事务回滚。 (回滚记录历史版本)

pipeline

使用pipeline，可以在客户端使用。减少和服务器的数据传输次数，一次性提交批量任务

示例代码

@Test  
    public void testPipeline() {  
        Jedis redis = new Jedis("centos7-3", 6379);  
    
        // 使用pipeline，减少和服务器的数据传输次数，一次性提交批量任务  
        // 对数据操作的api同jedis，最后需要调用sync，提交请求  
        Pipeline pipelined = redis.pipelined();  
    
        pipelined.set("jedis:1:name", "jd-April");  
    
        pipelined.lpush("jedis:list:4", "5", "4", "3", "3");  
    
        System.out.println(pipelined.get("jedis:1:name"));  
        System.out.println(pipelined.llen("jedis:list:4"));  
    
//         提交批处理到server  
//        pipelined.sync();  
        //         提交批处理到server，并返回结果  
        List<Object> objects = pipelined.syncAndReturnAll();  
        System.out.println(objects);  
        /*  
        Response string  
        Response long  
        [OK, 4, jd-April, 4]  
         */  
    }  

Lua脚本

由于redis本身提供的事务太弱，如果对事务有较高的需求，可以使用lua脚本

lua安装

curl -R -O http://www.lua.org/ftp/lua-5.4.2.tar.gz  
tar zxf lua-5.4.2.tar.gz  
    
yum -y install readline-devel ncurses-devel  
    
#在src目录下  
make linux  
或make install  

Lua环境协作组件从Redis2.6.0版本开始，通过内置的lua编译/解释器，可以使用EVAL命令对lua脚本进行求值。

脚本的命令是原子的，RedisServer在执行脚本命令中，不允许插入新的命令
脚本的命令可以复制，RedisServer在获得脚本后不执行，生成标识返回，Client根据标识就可以随时执行
EVAL命令
- EVAL script numkeys key [key …] arg [arg …]
  - script参数:是一段Lua脚本程序，它会被运行在Redis服务器上下文中，这段脚本不必(也不应该) 定义为一个Lua函数。
  - numkeys参数:用于指定键名参数的个数。
  - key [key …]参数: 从EVAL的第三个参数开始算起，使用了numkeys个键(key)，表示在脚本中所用到的那些Redis键(key)，这些键名参数可以在Lua中通过全局变量KEYS数组，用1为基址的形式访问( KEYS[1] ， KEYS[2] ，以此类推)。
  - arg [arg …]参数:可以在Lua中通过全局变量ARGV数组访问，访问的形式和KEYS变量类似( ARGV[1] 、 ARGV[2] ，诸如此类)。
```
eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 first second  
```
- lua脚本中调用Redis命令
  - redis.call():
    - 返回值就是redis命令执行的返回值
    - 如果出错，则返回错误信息，不继续执行
  - redis.pcall():
    - 返回值就是redis命令执行的返回值
    - 如果出错，则记录错误信息，继续执行
  - 注意事项
    - 在脚本中，使用return语句将返回值返回给客户端，如果没有return，则返回nil
```
eval "return redis.call('set',KEYS[1],ARGV[1])" 1 n1 zhaoyun  
```
EVALSHA EVAL 命令要求你在每次执行脚本的时候都发送一次脚本主体(script body)。
Redis 有一个内部的缓存机制，因此它不会每次都重新编译脚本，不过在很多场合，付出无谓的带宽来
传送脚本主体并不是最佳选择。
为了减少带宽的消耗， Redis 实现了 EVALSHA 命令，它的作用和 EVAL 一样，都用于对脚本求值，但
它接受的第一个参数不是脚本，而是脚本的 SHA1 校验和(sum)
- SCRIPT命令
  - SCRIPT FLUSH 清除所有脚本缓存
  - SCRIPT EXISTS 根据给定的脚本校验和，检查指定的脚本是否存在于脚本缓存
  - SCRIPT LOAD 将一个脚本装入脚本缓存，返回SHA1摘要，但并不立即运行它
    192.168.24.131:6380> script load "return redis.call('set',KEYS[1],ARGV[1])" "c686f316aaf1eb01d5a4de1b0b63cd233010e63d" 192.168.24.131:6380> evalsha c686f316aaf1eb01d5a4de1b0b63cd233010e63d 1 n2 zhangfei OK 192.168.24.131:6380> get n2
  - SCRIPT KILL 杀死当前正在运行的脚本

脚本管理命令实现使用redis-cli直接执行lua脚本

vim test.lua  
    
# 文件内容  
return redis.call('set',KEYS[1],ARGV[1])  
    
    
# 执行脚本  
./redis-cli -h 127.0.0.1 -p 6379 --eval test.lua name:6 , caocao #，两边有空格  

vim list.lua  
    
# 文件内容  
local key=KEYS[1]  
local list=redis.call("lrange",key,0,-1);  
return list;  
    
    
./redis-cli --eval list.lua list  

脚本复制 Redis 传播 Lua 脚本，在使用主从模式和开启AOF持久化的前提下: 当执行lua脚本时，Redis 服务器有两种模式:脚本传播模式和命令传播模式。
- 脚本传播模式主机和从机先后执行相同的脚本。在这一模式下执行的脚本不能有时间、内部状态、随机函数(spop)等。执行相同的脚本以及参数必须产生相同的效果。在Redis5，也是处于同一个事务中。
  
  脚本传播模式是 Redis 复制脚本时默认使用的模式 Redis会将被执行的脚本及其参数复制到 AOF 文件以及从服务器里面。
- 命令传播模式处于命令传播模式的主服务器会将执行脚本产生的所有写命令用事务包裹起来，然后将事务复制到 AOF 文件以及从服务器里面。
  因为命令传播模式复制的是写命令而不是脚本本身，所以即使脚本本身包含时间、内部状态、随机函数
  等，主服务器给所有从服务器复制的写命令仍然是相同的。
  为了开启命令传播模式，用户在使用脚本执行任何写操作之前，需要先在脚本里面调用以下函数:
```
redis.replicate_commands()  
```
  redis.replicate_commands() 只对调用该函数的脚本有效:在使用命令传播模式执行完当前脚本之后，服务器将自动切换回默认的脚本传播模式。
  - 示例如果我们在主服务器执行以下命令
    eval "redis.replicate_commands();redis.call('set',KEYS[1],ARGV[1]);redis.call('set',KEYS[2],ARGV[2])" 2 n1 n2 zhaoyun11 zhaoyun22
    那么主服务器将向从服务器复制以下命令:
    MULTI *3 $3 set $2 n1 $9 zhaoyun11 *3 $3 set $2 n2 $9 zhaoyun22 *1 $4 EXEC
管道(pipeline),事务和脚本(lua)三者的区别
- 三者都可以批量执行命令
- 管道无原子性，命令都是独立的，属于无状态的操作
- 事务和脚本是有原子性的，其区别在于脚本可借助Lua语言可在服务器端存储的便利性定制和简化操作
- 脚本的原子性要强于事务，脚本执行期间，另外的客户端其它任何脚本或者命令都无法执行，脚本的执行时间应该尽量短，不能太耗时的脚本

监视器monitor

Redis客户端通过执行MONITOR命令可以将自己变为一个监视器，实时地接受并打印出服务器当前处理的命令请求的相关信息。
此时，当其他客户端向服务器发送一条命令请求时，服务器除了会处理这条命令请求之外，还会将这条
命令请求的信息发送给所有监视器。

Redis客户端1

127.0.0.1:6379> monitor  
OK  
1613565408.095609 [0 127.0.0.1:37476] "set" "mm1" "mmmmm3"  

Redis客户端2
```
127.0.0.1:6379> set mm1 mmmmm3  
OK  
```

慢查询日志

慢查询设置在redis.conf中可以配置和慢查询日志相关的选项:

#执行时间超过多少微秒的命令请求会被记录到日志上 0 :全记录 <0 不记录 slowlog-log-slower-than 10000  
    
#slowlog-max-len 存储慢查询日志条数  
slowlog-max-len 128  

Redis使用列表存储慢查询日志，采用队列方式(FIFO)

config set的方式可以临时设置，redis重启后就无效
config set slowlog-log-slower-than 微秒

config set slowlog-max-len 条数

示例代码

127.0.0.1:6379> config set slowlog-log-slower-than 0  
OK  
127.0.0.1:6379>  config set slowlog-max-len 2  
OK  
127.0.0.1:6379>  set name:001 zhaoyun  
OK  
127.0.0.1:6379>  set name:002 zhangfei  
OK  
127.0.0.1:6379> get name:002  
"zhangfei"  
127.0.0.1:6379> SLOWLOG get  
1) 1) (integer) 4       # 日志的唯一标识符(uid)  
   2) (integer) 1613565819    #命令执行时的UINIX时间戳  
   3) (integer) 7			#命令执行时长(微秒)  
   4) 1) "get"				  
      2) "name:002"  
   5) "127.0.0.1:37476"  
   6) ""  
2) 1) (integer) 3  
   2) (integer) 1613565813  
   3) (integer) 4  
   4) 1) "set"  
      2) "name:002"  
      3) "zhangfei"  
   5) "127.0.0.1:37476"  
   6) ""  

slowlog get num 获取慢查询日志
slowlog len 获取慢查询日志的数量
slowlog reset 清空慢查询日志

慢查询记录的保存在redisServer中保存和慢查询日志相关的信息

struct redisServer {  
    // ...  
// 下一条慢查询日志的 ID  
long long slowlog_entry_id;  
// 保存了所有慢查询日志的链表 FIFO list *slowlog;  
// 服务器配置 slowlog-log-slower-than 选项的值 long long slowlog_log_slower_than;  
// 服务器配置 slowlog-max-len 选项的值 unsigned long slowlog_max_len;  
// ...  
};  

lowlog 链表保存了服务器中的所有慢查询日志，链表中的每个节点都保存了一个 slowlogEntry 结构，每个 slowlogEntry 结构代表一条慢查询日志。

typedef struct slowlogentry {  
//唯一标识符  
long long id;  
//命令执行时的时间，格式为UNIX时间戳  
time_t time;  
//执行命令消耗的时间，以微秒为单位  
long long duration;  
//命令与命令参数  
robj **argv;  
//命令与命令参数的数量  
int argc;  
} slowlogEntry;  

慢查询定位&处理使用slowlog get 可以获得执行较慢的redis命令，针对该命令可以进行优化:

1、尽量使用短的key，对于value有些也可精简，能使用int就int。
2、避免使用keys *、hgetall等全量操作。
3、减少大key的存取，打散为小key 100K以上
4、将rdb改为aof模式
rdb fork 子进程数据量过大主进程阻塞 redis性能大幅下降
关闭持久化， (适合于数据量较小，有固定数据源)
5、想要一次添加多条数据的时候可以使用管道
6、尽可能地使用哈希存储
7、尽量限制下redis使用的内存大小，这样可以避免redis使用swap分区或者出现OOM错误内存与硬盘的swap

Redis持久化机制

为什么要持久化

Redis是内存数据库，宕机后数据会消失。
Redis重启后快速恢复数据，要提供持久化机制
Redis持久化是为了快速的恢复数据而不是为了存储数据
Redis有两种持久化方式:RDB和AOF

注意:Redis持久化不保证数据的完整性。
当Redis用作DB时，DB数据要完整，所以一定要有一个完整的数据源(文件、mysql) 在系统启动时，从这个完整的数据源中将数据load到Redis中
数据量较小，不易改变，比如:字典库(xml、Table)

info 查看配置信息

# Persistence  
loading:0  
rdb_changes_since_last_save:1  
rdb_bgsave_in_progress:0  
rdb_last_save_time:1589363051  
rdb_last_bgsave_status:ok  
rdb_last_bgsave_time_sec:-1  
rdb_current_bgsave_time_sec:-1  
rdb_last_cow_size:0  
aof_enabled:1  
aof_rewrite_in_progress:0  
aof_rewrite_scheduled:0  
aof_last_rewrite_time_sec:-1  
aof_current_rewrite_time_sec:-1  
aof_last_bgrewrite_status:ok  
aof_last_write_status:ok  
aof_last_cow_size:0  
aof_current_size:58  
aof_base_size:0  
aof_pending_rewrite:0  
aof_buffer_length:0  
aof_rewrite_buffer_length:0  
aof_pending_bio_fsync:0  
aof_delayed_fsync:0  

RDB

RDB(Redis DataBase)，是redis默认的存储方式，RDB方式是通过快照( snapshotting )完成的。
这一刻的数据不关注过程

触发快照的方式

符合自定义配置的快照规则
执行save或者bgsave命令
执行flushall命令
执行主从复制操作 (第一次)

配置参数定期执行在redis.conf中配置:save 多少秒内数据变了多少

save "" # 不使用RDB存储 不能主从  
save 900 1 # 表示15分钟(900秒钟)内至少1个键被更改则进行快照。 save 300 10 # 表示5分钟(300秒)内至少10个键被更改则进行快照。 save 60 10000 # 表示1分钟内至少10000个键被更改则进行快照。  

漏斗设计提高性能

命令显式触发在客户端输入bgsave命令。

127.0.0.1:6379> bgsave  
Background saving started  

RDB执行流程
1. Redis父进程首先判断:当前是否在执行save，或bgsave/bgrewriteaof(aof文件重写命令)的子进程，如果在执行则bgsave命令直接返回。
2. 父进程执行fork(调用OS函数复制主进程)操作创建子进程，这个复制过程中父进程是阻塞的， Redis不能执行来自客户端的任何命令。
3. 父进程fork后，bgsave命令返回”Background saving started”信息并不再阻塞父进程，并可以响应其他命令。
4. 子进程创建RDB文件，根据父进程内存快照生成临时快照文件，完成后对原有文件进行原子替换。 (RDB始终完整)
5. 子进程发送信号给父进程表示完成，父进程更新统计信息。
6. 父进程fork子进程后，继续工作。
RDB文件结构
1、头部5字节固定为“REDIS”字符串 2、4字节“RDB”版本号(不是Redis版本号)，当前为9，填充后为0009
3、辅助字段，以key-value的形式
4、存储数据库号码
5、字典大小
6、过期key
7、主要数据，以key-value的形式存储
8、结束标志
9、校验和，就是看文件是否损坏，或者是否被修改。
- 辅助字段
RDB的优缺点优点
RDB是二进制压缩文件，占用空间小，便于传输(传给slaver)
主进程fork子进程，可以最大化Redis性能，主进程不能太大，Redis的数据量不能太大，复制过程中主进程阻塞

缺点
不保证数据完整性，会丢失最后一次快照以后更改的所有数据

AOF

AOF(append only file)是Redis的另一种持久化方式。Redis默认情况下是不开启的。开启AOF持久化后
Redis 将所有对数据库进行过写入的命令(及其参数)(RESP)记录到 AOF 文件，以此达到记录数据库状态的目的，
这样当Redis重启后只要按顺序回放这些命令就会恢复到原始状态了。 AOF会记录过程，RDB只管结果

AOF持久化实现配置 redis.conf

# 可以通过修改redis.conf配置文件中的appendonly参数开启   
appendonly yes  
    
# AOF文件的保存位置和RDB文件的位置相同，都是通过dir参数设置的。 dir ./  
    
# 默认的文件名是appendonly.aof，可以通过appendfilename参数修改 appendfilename appendonly.aof  

AOF原理 AOF文件中存储的是redis的命令，同步命令到 AOF 文件的整个过程可以分为三个阶段:
命令传播:Redis 将执行完的命令、命令的参数、命令的参数个数等信息发送到 AOF 程序中。缓存追加:AOF 程序根据接收到的命令数据，将命令转换为网络通讯协议的格式，然后将协议内容追加到服务器的 AOF 缓存中。文件写入和保存:AOF 缓存中的内容被写入到 AOF 文件末尾，如果设定的 AOF 保存条件被满足的话， fsync 函数或者 fdatasync 函数会被调用，将写入的内容真正地保存到磁盘中。
- 命令传播当一个 Redis 客户端需要执行命令时，它通过网络连接，将协议文本发送给 Redis 服务器。服务器在接到客户端的请求之后，它会根据协议文本的内容，选择适当的命令函数，并将各个参数从字符串文本转换为 Redis 字符串对象( StringObject )。每当命令函数成功执行之后，命令参数都会被传播到 AOF 程序。
- 缓存追加当命令被传播到 AOF 程序之后，程序会根据命令以及命令的参数，将命令从字符串对象转换回原来的协议文本。协议文本生成之后，它会被追加到 redis.h/redisServer 结构的 aof_buf 末尾。
  
  redisServer 结构维持着 Redis 服务器的状态， aof_buf 域则保存着所有等待写入到 AOF 文件的协议文本(RESP)。
- 文件写入和保存每当服务器常规任务函数被执行、或者事件处理器被执行时， aof.c/flushAppendOnlyFile 函数都会被调用，这个函数执行以下两个工作:
  
  WRITE:根据条件，将 aof_buf 中的缓存写入到 AOF 文件。
  SAVE:根据条件，调用 fsync 或 fdatasync 函数，将 AOF 文件保存到磁盘中。
- AOF 保存模式 Redis 目前支持三种 AOF 保存模式，它们分别是:
  AOF_FSYNC_NO :不保存。 AOF_FSYNC_EVERYSEC :每一秒钟保存一次。(默认) AOF_FSYNC_ALWAYS :每执行一个命令保存一次。(不推荐)
  - 不保存在这种模式下，每次调用 flushAppendOnlyFile 函数， WRITE 都会被执行，但 SAVE 会被略过。在这种模式下， SAVE 只会在以下任意一种情况中被执行:
    Redis 被关闭 AOF 功能被关闭系统的写缓存被刷新(可能是缓存已经被写满，或者定期保存操作被执行) 这三种情况下的 SAVE 操作都会引起 Redis 主进程阻塞。
  - 每一秒钟保存一次(推荐) 在这种模式中， SAVE 原则上每隔一秒钟就会执行一次，因为 SAVE 操作是由后台子线程(fork)调用的，所以它不会引起服务器主进程阻塞。
  - 每执行一个命令保存一次在这种模式下，每次执行完一个命令之后， WRITE 和 SAVE 都会被执行。
    另外，因为 SAVE 是由 Redis 主进程执行的，所以在 SAVE 执行期间，主进程会被阻塞，不能接受命令请求。
  - AOF 保存模式对性能和安全性的影响
AOF重写 AOF记录数据的变化过程，越来越大，需要重写“瘦身”
Redis可以在 AOF体积变得过大时，自动地在后台(Fork子进程)对 AOF进行重写。重写后的新 AOF文件包含了恢复当前数据集所需的最小命令集合。所谓的“重写”其实是一个有歧义的词语，实际上， AOF 重写并不需要对原有的 AOF 文件进行任何写入和读取，它针对的是数据库中键的当前值。

Redis 不希望 AOF 重写造成服务器无法处理请求，所以 Redis 决定将 AOF 重写程序放到(后台)子进
程里执行，这样处理的最大好处是:
1、子进程进行 AOF 重写期间，主进程可以继续处理命令请求。
2、子进程带有主进程的数据副本，使用子进程而不是线程，可以在避免锁的情况下，保证数据的安全性。
- AOF重写缓存
  使用子进程也有一个问题需要解决: 因为子进程在进行 AOF 重写期间，主进程还需要继续处理命令，而新的命令可能对现有的数据进行修改，这会让当前数据库的数据和重写后的 AOF 文件中的数据不一致。
  为了解决这个问题， Redis 增加了一个 AOF 重写缓存，这个缓存在 fork 出子进程之后开始启用， Redis 主进程在接到新的写命令之后，除了会将这个写命令的协议内容追加到现有的 AOF 文件之外，还会追加到这个缓存中。
- 重写过程分析
  Redis 在创建新 AOF 文件的过程中，会继续将命令追加到现有的 AOF 文件里面，即使重写过程中发生停机，现有的 AOF 文件也不会丢失。而一旦新 AOF 文件创建完毕，Redis 就会从旧 AOF 文件切换到新 AOF 文件，并开始对新 AOF 文件进行追加操作。
  
  当子进程在执行 AOF 重写时，主进程需要执行以下三个工作:
  处理命令请求。将写命令追加到现有的 AOF 文件中。将写命令追加到 AOF 重写缓存中。这样一来可以保证:
  现有的 AOF 功能会继续执行，即使在 AOF 重写期间发生停机，也不会有任何数据丢失。所有对数据库进行修改的命令都会被记录到 AOF 重写缓存中。当子进程完成 AOF 重写之后，它会向父进程发送一个完成信号，父进程在接到完成信号之后，会调用一个信号处理函数，并完成以下工作:
  将 AOF 重写缓存中的内容全部写入到新 AOF 文件中。对新的 AOF 文件进行改名，覆盖原有的 AOF 文件。
  
  Redis数据库里的+AOF重写过程中的命令——->新的AOF文件—->覆盖老的
  当步骤 1 执行完毕之后，现有 AOF 文件、新 AOF 文件和数据库三者的状态就完全一致了。当步骤 2 执行完毕之后，程序就完成了新旧两个 AOF 文件的交替。
  这个信号处理函数执行完毕之后，主进程就可以继续像往常一样接受命令请求了。在整个 AOF 后台重写过程中，只有最后的写入缓存和改名操作会造成主进程阻塞，在其他时候， AOF 后台重写都不会对主进程造成阻塞，这将 AOF 重写对性能造成的影响降到了最低。
- 触发方式
  - 配置触发在redis.conf中配置
    # 表示当前aof文件大小超过上一次aof文件大小的百分之多少的时候会进行重写。如果之前没有重写过，以启动时aof文件大小为准 auto-aof-rewrite-percentage 100 # 限制允许重写最小aof文件大小，也就是文件大小小于64mb的时候，不需要进行优化 auto-aof-rewrite-min-size 64mb
  - 执行bgrewriteaof命令
    127.0.0.1:6379> bgrewriteaof Background append only file rewriting started

混合持久化

RDB和AOF各有优缺点，Redis 4.0 开始支持 rdb 和 aof 的混合持久化。如果把混合持久化打开，aof rewrite 的时候就直接把 rdb 的内容写到 aof 文件开头。

RDB的头+AOF的身体—->appendonly.aof

开启混合持久化
在 redis.conf 中
```
aof-use-rdb-preamble yes  
```
我们可以看到该AOF文件是rdb文件的头和aof格式的内容，在加载时，首先会识别AOF文件是否以 REDIS字符串开头，如果是就按RDB格式加载，加载完RDB后继续按AOF格式加载剩余部分。

AOF文件的载入与数据还原

因为AOF文件里面包含了重建数据库状态所需的所有写命令，所以服务器只要读入并重新执行一遍AOF 文件里面保存的写命令，就可以还原服务器关闭之前的数据库状态 Redis读取AOF文件并还原数据库状态的详细步骤如下:
1、创建一个不带网络连接的伪客户端(fake client):因为Redis的命令只能在客户端上下文中执行，而载入AOF文件时所使用的命令直接来源于AOF文件而不是网络连接，所以服务器使用了一个没有网络连接的伪客户端来执行AOF文件保存的写命令，伪客户端执行命令的效果和带网络连接的客户端执行命令的效果完全一样
2、从AOF文件中分析并读取出一条写命令
3、使用伪客户端执行被读出的写命令 4、一直执行步骤2和步骤3，直到AOF文件中的所有写命令都被处理完毕为止当完成以上步骤之后，AOF文件所保存的数据库状态就会被完整地还原出来

RDB与AOF对比

1、RDB存某个时刻的数据快照，采用二进制压缩存储，AOF存操作命令，采用文本存储(混合)
2、RDB性能高、AOF性能较低
3、RDB在配置触发状态会丢失最后一次快照以后更改的所有数据，AOF设置为每秒保存一次，则最多丢2秒的数据
4、Redis以主服务器模式运行，RDB不会保存过期键值对数据，Redis以从服务器模式运行，RDB会保存过期键值对，当主服务器向从服务器同步时，再清空过期键值对。
AOF写入文件时，对过期的key会追加一条del命令，当执行AOF重写时，会忽略过期key和del命令。

高可用方案

“高可用性”(High Availability)通常来描述一个系统经过专门的设计，从而减少停工时间，而保持其服务的高度可用性。
单机的Redis是无法保证高可用性的，当Redis服务器宕机后，即使在有持久化的机制下也无法保证不丢失数据。
所以我们采用Redis多机和集群的方式来保证Redis的高可用性。单进程+单线程 + 多机 (集群)

主从复制

Redis支持主从复制功能，可以通过执行slaveof(Redis5以后改成replicaof)或者在配置文件中设置 slaveof(Redis5以后改成replicaof)来开启复制功能。

主对外从对内，主可写从不可写
主挂了，从不可为主
一主一从
一主多从
传递复制
作用
- 读写分离
  - 一主多从，主从同步
  - 主负责写，从负责读
  - 提升Redis的性能和吞吐量
  - 主从的数据一致性问题
- 数据容灾
  - 从机是主机的备份
  - 主机宕机，从机可读不可写
  - 默认情况下主机宕机后，从机不可为主机
  - 利用哨兵可以实现主从切换，做到高可用

主从配置

主Redis配置无需特殊配置

从Redis配置修改从服务器上的 redis.conf 文件:

# replicaof <masterip> <masterport>  
# 表示当前【从服务器】对应的【主服务器】的IP是192.168.10.135，端口是6379  
replicaof 192.168.10.135 6379  

原理与实现
- 复制流程
  - 保存主节点信息当客户端向从服务器发送slaveof(replicaof) 主机地址(127.0.0.1) 端口(6379)时:从服务器将主机 ip(127.0.0.1)和端口(6379)保存到redisServer的masterhost和masterport中。
    Struct redisServer{ char *masterhost;//主服务器ip int masterport;//主服务器端口 };
    从服务器将向发送SLAVEOF命令的客户端返回OK，表示复制指令已经被接收，而实际上复制工作是在 OK返回之后进行。
  - 建立socket连接
    slaver与master建立socket连接
    slaver关联文件事件处理器该处理器接收RDB文件(全量复制)、接收Master传播来的写命令(增量复制)
    
    主服务器accept从服务器Socket连接后，创建相应的客户端状态。相当于从服务器是主服务器的Client 端。
  - 发送ping命令
    Slaver向Master发送ping命令
    1、检测socket的读写状态
    2、检测Master能否正常处理
    
    Master的响应:
    1、发送“pong” , 说明正常
    2、返回错误，说明Master不正常
    3、timeout，说明网络超时
  - 权限验证
    主从正常连接后，进行权限验证
    主未设置密码(requirepass=“”) ，从也不用设置密码(masterauth=“”)
    主设置密码(requirepass!=””),从需要设置密码(masterauth=主的requirepass的值)
    或者从通过auth命令向主发送密码
  - 发送端口信息
    在身份验证步骤之后，从服务器将执行命令REPLCONF listening-port ，向主服务器发送从服务器的监听端口号。
  - 同步数据分为全量同步和增量同步
  - 命令传播当同步数据完成后，主从服务器就会进入命令传播阶段，主服务器只要将自己执行的写命令发送给从服
    务器，而从服务器只要一直执行并接收主服务器发来的写命令。
同步数据集 Redis 2.8以前使用SYNC命令同步复制
Redis 2.8之后采用PSYNC命令替代
- 旧版本 同步操作:
  1. 通过从服务器发送到SYNC命令给主服务器
  2. 主服务器生成RDB文件并发送给从服务器，同时发送保存所有写命令给从服务器
  3. 从服务器清空之前数据并执行解释RDB文件
  4. 保持数据一致(还需要命令传播过程才能保持一致)
  命令传播操作:
  同步操作完成后，主服务器执行写命令，该命令发送给从服务器并执行，使主从保存一致。
  缺陷
  没有全量同步和增量同步的概念，从服务器在同步时，会清空所有数据。
  主从服务器断线后重复制，主服务器会重新生成RDB文件和重新记录缓冲区的所有命令，并全量同步到从服务器上。
- 新版
  在Redis 2.8之后使用PSYNC命令，具备完整重同步和部分重同步模式。
  - Redis 的主从同步，分为全量同步和增量同步。
  - 只有从机第一次连接上主机是全量同步。
  - 断线重连有可能触发全量同步也有可能是增量同步( master 判断 runid 是否一致)。
  - 除此之外的情况都是增量同步。
    - 全量同步 Redis的全量同步过程主要分三个阶段:
      - 同步快照阶段:Master创建并发送快照RDB给Slave，Slave载入并解析快照。Master同时将此阶段所产生的新的写命令存储到缓冲区。
      - 同步写缓冲阶段:Master向Slave同步存储在缓冲区的写操作命令。
      - 同步增量阶段:Master向Slave同步写操作命令。
    - 增量同步
      - Redis增量同步主要指Slave完成初始化后开始正常工作时， Master 发生的写操作同步到 Slave 的过程。
      - 通常情况下， Master 每执行一个写命令就会向 Slave 发送相同的写命令，然后 Slave 接收并执行。
心跳检测在命令传播阶段，从服务器默认会以每秒一次的频率向主服务器发送命令:
```
replconf ack <replication_offset>  
#ack :应答   
#replication_offset:从服务器当前的复制偏移量  
```
主要作用有三个:
- 检测主从的连接状态检测主从服务器的网络连接状态
  通过向主服务器发送INFO replication命令，可以列出从服务器列表，可以看出从最后一次向主发送命令距离现在过了多少秒。lag的值应该在0或1之间跳动，如果超过1则说明主从之间的连接有故障。
- 辅助实现min-slaves Redis可以通过配置防止主服务器在不安全的情况下执行写命令
  min-slaves-to-write 3 (min-replicas-to-write 3 )
  min-slaves-max-lag 10 (min-replicas-max-lag 10)
  上面的配置表示:从服务器的数量少于3个，或者三个从服务器的延迟(lag)值都大于或等于10 秒时，主服务器将拒绝执行写命令。这里的延迟值就是上面INFOreplication命令的lag值。
- 检测命令丢失如果因为网络故障，主服务器传播给从服务器的写命令在半路丢失，那么当从服务器向主服务器发送REPLCONF ACK命令时，主服务器将发觉从服务器当前的复制偏移量少于自己的复制偏移量，然后主服务器就会根据从服务器提交的复制偏移量，在复制积压缓冲区里面找到从服务器缺少的数据，并将这些数据重新发送给从服务器。(补发) 网络不断
  增量同步:网断了，再次连接时

哨兵模式

哨兵(sentinel)是Redis的高可用性(High Availability)的解决方案: 由一个或多个sentinel实例组成sentinel集群可以监视一个或多个主服务器和多个从服器。
在主从模式开启的前提下开启哨兵模式。
当主服务器进入下线状态时，sentinel可以将该主服务器下的某一从服务器升级为主服务器继续提供服务，从而保证redis的高可用性。

部署方案

搭建配置

主从配置

master

mkdir -p /var/redis-ms/redis-master  
		    
# 复制一份编译好的redis到新的地址  
cp -r /usr/redis/ /var/redis-ms/redis-master  
		    
# 从源码文件夹复制一份模版redis.conf  
cp /opt/lagou/software/redis-5.0.5/redis.conf  /var/redis-ms/redis-master/bin/  
		    
		    
# 修改配置文件  
vim redis.conf  
		    
# 将`daemonize`由`no`改为`yes` daemonize yes  
		    
# 默认绑定的是回环地址，默认不能被其他机器访问   
# bind 127.0.0.1  
		    
# 是否开启保护模式，由yes该为no, 这样其他机器才能访问   
protected-mode no  

slave1

# 复制master文件夹  
cp -r /var/redis-ms/redis-master/ /var/redis-ms/redis-slaver1  
		    
# 修改配置文件redis.conf  
vim redis.conf  
		    
# 本地启动端口  
port 6380  
# 配置主机ip和端口  
replicaof 127.0.0.1 6369  

slave2

# 复制master文件夹  
cp -r /var/redis-ms/redis-master/ /var/redis-ms/redis-slaver2  
		    
# 修改配置文件redis.conf  
vim redis.conf  
		    
# 本地启动端口  
port 6381  
# 配置主机ip和端口  
replicaof 127.0.0.1 6369  

哨兵配置

sentinel1

# 复制master文件夹  
cp -r /var/redis-ms/redis-master/ /var/redis-ms/redis-sentinel1  
		    
# 从源码文件夹复制一份模版  
cp /opt/lagou/software/redis-5.0.5/sentinel.conf  /var/redis-ms/redis-master/bin/  
		    
# 哨兵sentinel实例运行的端口 默认26379   
port 26379  
		    
daemonize yes  
		    
# 哨兵sentinel监控的redis主节点的 ip port  
# master-name 可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。  
# quorum 当这些quorum个数sentinel哨兵认为master主节点失联 那么这时 客观上认为主节点失联了   
# sentinel monitor <master-name> <ip> <redis-port> <quorum>  
sentinel monitor mymaster 127.0.0.1 6379 2  
		    
# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认30秒，改成3秒   
# sentinel down-after-milliseconds <master-name> <milliseconds>  
		    
sentinel down-after-milliseconds mymaster 3000  

sentinel2

# 复制sentinel1文件夹  
cp -r /var/redis-ms/redis-sentinel1/ /var/redis-ms/redis-sentinel2  
		    
# 修改配置文件中的端口  
vim sentinel.conf  
		    
port 26380  

sentinel3

# 复制sentinel1文件夹  
cp -r /var/redis-ms/redis-sentinel1/ /var/redis-ms/redis-sentinel3  
		    
# 修改配置文件中的端口  
vim sentinel.conf  
		    
port 26381  

启动流程 #启动redis-master和redis-slaver
- 在redis-master目录下 ./redis-server redis.conf
- 在redis-slaver1目录下./redis-server redis.conf
- 在redis-slaver2目录下 ./redis-server redis.conf
#启动redis-sentinel
- 在redis-sentine11目录下./redis-sentinel sentinel.conf
- 在redis-sentine12目录下./redis-sentinel sentinel.conf
- 在redis-sentine13目录下./redis-sentinel sentinel.conf

执行流程
- 启动并初始化Sentinel
  Sentinel是一个特殊的Redis服务器
  不会进行持久化
  Sentinel实例启动后, 每个Sentinel会创建2个连向主服务器的网络连接
  - 命令连接:用于向主服务器发送命令，并接收响应;
  - 订阅连接:用于订阅主服务器的—sentinel—:hello频道。
- 获取主服务器信息 Sentinel默认每10s一次，向被监控的主服务器发送info命令，获取主服务器和其下属从服务器的信息。
- 获取从服务器信息
  当Sentinel发现主服务器有新的从服务器出现时，Sentinel还会向从服务器建立命令连接和订阅连接。在命令连接建立之后，Sentinel还是默认10s一次，向从服务器发送info命令，并记录从服务器的信息。
- 向主服务器和从服务器发送消息(以订阅的方式) 默认情况下，Sentinel每2s一次，向所有被监视的主服务器和从服务器所订阅的—sentinel—:hello频道
  上发送消息，消息中会携带Sentinel自身的信息和主服务器的信息。
```
PUBLISH _sentinel_:hello "< s_ip > < s_port >< s_runid >< s_epoch > < m_name > <  
m_ip >< m_port ><m_epoch>"  
	    
int sentinelSendHello(sentinelRedisInstance *ri)   
```
- 接收来自主服务器和从服务器的频道信息当Sentinel与主服务器或者从服务器建立起订阅连接之后，Sentinel就会通过订阅连接，向服务器发送以下命令:
```
subscribe —sentinel—:hello  
```
  Sentinel彼此之间只创建命令连接，而不创建订阅连接，因为Sentinel通过订阅主服务器或从服务器，就可以感知到新的Sentinel的加入，而一旦新Sentinel加入后，相互感知的Sentinel通过命令连接来通信就可以了。
- 检测主观下线状态 Sentinel每秒一次向所有与它建立了命令连接的实例(主服务器、从服务器和其他Sentinel)发送PING命令
  实例在down-after-milliseconds毫秒内返回无效回复(除了+PONG、-LOADING、-MASTERDOWN外)
  实例在down-after-milliseconds毫秒内无回复(超时)
  Sentinel就会认为该实例主观下线(SDown)
- 检查客观下线状态当一个Sentinel将一个主服务器判断为主观下线后 Sentinel会向同时监控这个主服务器的所有其他Sentinel发送查询命令
  主机的
```
SENTINEL is-master-down-by-addr <ip> <port> <current_epoch> <runid>  
```
  其他Sentinel回复
```
<down_state>< leader_runid >< leader_epoch >  
```
  判断它们是否也认为主服务器下线。如果达到Sentinel配置中的quorum数量的Sentinel实例都判断主服务器为主观下线，则该主服务器就会被判定为客观下线(ODown)。
- 选举Leader Sentinel 当一个主服务器被判定为客观下线后，监视这个主服务器的所有Sentinel会通过选举算法(raft)，选
  出一个Leader Sentinel去执行failover(故障转移)操作。
哨兵leader选举
- Raft Raft协议是用来解决分布式系统一致性问题的协议。 Raft协议描述的节点共有三种状态:Leader, Follower, Candidate。
  term:Raft协议将时间切分为一个个的Term(任期)，可以认为是一种“逻辑时间”。
  - 选举流程 Raft采用心跳机制触发Leader选举
    系统启动后，全部节点初始化为Follower，term为0。
    节点如果收到了RequestVote或者AppendEntries，就会保持自己的Follower身份。
    节点如果一段时间内没收到AppendEntries消息，在该节点的超时时间内还没发现Leader，Follower就
    会转换成Candidate，自己开始竞选Leader。
    一旦转化为Candidate，该节点立即开始下面几件事情:
    - 增加自己的term。
    - 启动一个新的定时器。
    - 给自己投一票。
    - 向所有其他节点发送RequestVote，并等待其他节点的回复。
    如果在计时器超时前，节点收到多数节点的同意投票，就转换成Leader。同时向所有其他节点发送 AppendEntries，告知自己成为了Leader。
    每个节点在一个term内只能投一票，采取先到先得的策略，Candidate前面说到已经投给了自己， Follower会投给第一个收到RequestVote的节点。
    Raft协议的定时器采取随机超时时间，这是选举Leader的关键。在同一个term内，先转为Candidate的节点会先发起投票，从而获得多数票。
- Sentinel的leader选举流程 1、某Sentinel认定master客观下线后，该Sentinel会先看看自己有没有投过票，如果自己已经投过票给其他Sentinel了，在一定时间内自己就不会成为Leader。
  2、如果该Sentinel还没投过票，那么它就成为Candidate。
  3、Sentinel需要完成几件事情:
  - 更新故障转移状态为start
  - 当前epoch加1，相当于进入一个新term，在Sentinel中epoch就是Raft协议中的term。
  - 向其他节点发送 is-master-down-by-addr 命令请求投票。命令会带上自己的epoch。
  - 给自己投一票(leader、leader_epoch)
  4、当其它哨兵收到此命令时，可以同意或者拒绝它成为领导者;(通过判断epoch)
  5、Candidate会不断的统计自己的票数，直到他发现认同他成为Leader的票数超过一半而且超过它配置的quorum，这时它就成为了Leader。
  6、其他Sentinel等待Leader从slave选出master后，检测到新的master正常工作后，就会去掉客观下线的标识。
故障转移当选举出Leader Sentinel后，Leader Sentinel会对下线的主服务器执行故障转移操作，主要有三个步骤:
1.它会将失效Master的其中一个slave升级为新的Master,并让失效Master的其他Slave改为复制新的Master;

2.当客户端试图连接失效的Master时，集群也会向客户端返回新Master的地址，使得集群可以使用现在的Master替换失效Master。

3.Master和Slave服务器切换后，Master的redis.conf、Slave的redis.conf和sentinel.conf的配置文件的内容都会发生相应的改变，即，Master主服务器的redis.conf 配置文件中会多一行replicaof的配置，sentinel.conf的监控目标会随之调换。
主服务器的选择哨兵leader根据以下规则从客观下线的主服务器的从服务器中选择出新的主服务器。
1. 过滤掉主观下线的节点
2. 选择slave-priority最高的节点，如果由则返回没有就继续选择
3. 选择出复制偏移量最大的系节点，因为复制偏移量越大则数据复制的越完整，如果有就返回了，没有就继续
4. 选择run_id最小的节点，因为run_id越小说明重启次数越少

集群与分区

分区是将数据分布在多个Redis实例(Redis主机)上，以至于每个实例只包含一部分数据。

分区的意义
- 性能的提升单机Redis的网络I/O能力和计算资源是有限的，将请求分散到多台机器，充分利用多台机器的计算能力
  可网络带宽，有助于提高Redis总体的服务能力。
- 存储能力的横向扩展即使Redis的服务能力能够满足应用需求，但是随着存储数据的增加，单台机器受限于机器本身的存储容量，将数据分散到多台机器上存储使得Redis服务可以横向扩展。
分区的方式
- 范围分区
  根据id数字的范围比如1–10000、100001–20000…..90001-100000，每个范围分到不同的Redis实例中
  
  好处:
  实现简单，方便迁移和扩展
  
  缺陷:
  热点数据分布不均，性能损失
  非数字型key，比如uuid无法使用(可采用雪花算法替代)
  分布式环境主键使用雪花算法生成
- hash分区利用简单的hash算法即可:
  Redis实例=hash(key)%N
  key:要进行分区的键，比如user_id N:Redis实例个数(Redis主机)
  
  好处:
  支持任何类型的key
  热点分布较均匀，性能较好
  
  缺陷:
  迁移复杂，需要重新计算，扩展较差(利用一致性hash环)
- client端分区
  对于一个给定的key，客户端直接选择正确的节点来进行读写。许多Redis客户端都实现了客户端分区
  (JedisPool)，也可以自行编程实现。
  - 客户端选择算法
    - hash 普通hash
      hash(key)%N
      hash:可以采用hash算法，比如CRC32、CRC16等
      N:是Redis主机个数
      比如:
      user_id : u001 hash(u001) : 1844213068 Redis实例=1844213068%3 余数为2，所以选择Redis3。
      优势:
      实现简单，热点数据分布均匀
      
      缺陷
      节点数固定，扩展的话需要重新计算查询时必须用分片的key来查，一旦key改变，数据就查不出了，所以要使用不易改变的key进行分片
    - 一致性hash
      优点
      添加或移除节点时，数据只需要做部分的迁移
      
      缺点
      复杂度高:客户端需要自己处理数据路由、高可用、故障转移等问题
      使用分区，数据的处理会变得复杂，不得不对付多个redis数据库和AOF文件，不得在多个实例和主机之间持久化你的数据。
      不易扩展: 一旦节点的增或者删操作，都会导致key无法在redis中命中，必须重新根据节点计算，并手动迁移全部或部分数据。
      - hash环偏移
        在介绍一致性哈希的概念时，我们理想化的将3台服务器均匀的映射到了hash环上。也就是说数据的范围是2^32/N。但实际情况往往不是这样的。有可能某个服务器的数据会很多，某个服务器的数据会很少，造成服务器性能不平均。这种现象称为hash环偏移。
        
        虚拟节点
        理论上我们可以通过增加服务器的方式来减少偏移，但这样成本较高，所以我们可以采用虚拟节点的方式，也就是虚拟服务器
        
        “虚拟节点”是”实际节点”(实际的物理服务器)在hash环上的复制品,一个实际节点可以对应多个虚拟节点。
        从上图可以看出，A、B、C三台服务器分别虚拟出了一个虚拟节点，当然，如果你需要，也可以虚拟出更多的虚拟节点。引入虚拟节点的概念后，缓存的分布就均衡多了，上图中，1号、3号数据被缓存在服务器A中，5号、4号数据被缓存在服务器B中，6号、2号数据被缓存在服务器C中，如果你还不放心，可以虚拟出更多的虚拟节点，以便减小hash环偏斜所带来的影响，虚拟节点越多，hash环上的节点就越多，缓存被均匀分布的概率就越大。
- 官方cluster分区
  Redis3.0之后，Redis官方提供了完整的集群解决方案。
  方案采用去中心化的方式，包括:sharding(分区)、replication(复制)、failover(故障转移)。称为RedisCluster。
  
  Redis5.0前采用redis-trib进行集群的创建和管理，需要ruby支持
  Redis5.0可以直接使用Redis-cli进行集群的创建和管理
  - 去中心化 RedisCluster由多个Redis节点组构成，是一个P2P无中心节点的集群架构，依靠Gossip协议传播的集群。
  - Gossip协议 Gossip协议是一个通信协议，一种传播消息的方式。
    起源于:病毒传播
    Gossip协议基本思想就是:
    一个节点周期性(每秒)随机选择一些节点，并把信息传递给这些节点。这些收到信息的节点接下来会做同样的事情，即把这些信息传递给其他一些随机选择的节点。信息会周期性的传递给N个目标节点。这个N被称为fanout(扇出)
    - 包含的消息
      通过gossip协议，cluster可以提供集群间状态同步更新、选举自助failover等重要的集群功能。
  - slot
    redis-cluster把所有的物理节点映射到[0-16383]个slot上,基本上采用平均分配和连续分配的方式。比如上图中有5个主节点，这样在RedisCluster创建时
    
    当需要在 Redis 集群中放置一个 key-value 时，redis 先对 key 使用 crc16 算法算出一个结果，然后把结果对 16384 求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽，redis 会根据节点数量大致均等的将哈希槽映射到不同的节点。
    比如:
    set name zhaoyun
    hash(“name”)采用crc16算法，得到值:1324203551%16384=15903 根据上表15903在13088-16383之间，所以name被存储在Redis5节点。 slot槽必须在节点上连续分配，如果出现不连续的情况，则RedisCluster不能工作，详见容错。
  - RedisCluster的优势
    - 高性能 Redis Cluster 的性能与单节点部署是同级别的。多主节点、负载均衡、读写分离
    - 高可用 Redis Cluster 支持标准的主从复制配置来保障高可用和高可靠。failover
      Redis Cluster 也实现了一个类似 Raft 的共识方式，来保障整个集群的可用性。
    - 易扩展向 Redis Cluster 中添加新节点，或者移除节点，都是透明的，不需要停机。
      水平、垂直方向都非常容易扩展。
      数据分区，海量数据，数据存储
    - 原生部署 Redis Cluster 不需要其他的代理或者工具，而且 Redis Cluster 和单机 Redis 几乎完全兼容。
  - 集群搭建 RedisCluster最少需要三台主服务器，三台从服务器。
    这里在一台机器上操作，搭建伪集群。
    端口号分别为:7001~7006
    mkdir /var/redis-cluster/7001 # 在源码的src目录下 make install PREFIX=/var/redis-cluster/7001 # 复制源码目录下的redis.conf cp redis.conf /var/redis-cluster/7001/bin
    - 安装和配置
      - 配置7001
        vim redis.conf port 7001 daemonize yes #bind 127.0.0.01 protected-mode no # 打开集群模式 cluster-enabled yes
      - 配置7002
        cd /var/redis-cluster cp -r 7001 7002 vim 7002/bin/redis.conf # port port 7002
      - 配置7003
        cd /var/redis-cluster cp -r 7001 7003 vim 7003/bin/redis.conf # port port 7003
      - 配置7004
        cd /var/redis-cluster cp -r 7001 7004 vim 7004/bin/redis.conf # port port 7004
      - 配置7005
        cd /var/redis-cluster cp -r 7001 7005 vim 7005/bin/redis.conf # port port 7005
      - 配置7006
        cd /var/redis-cluster cp -r 7001 7006 vim 7006/bin/redis.conf # port port 7006
      - 群起脚本
        vim start.sh #!/bin/bash path="/var/redis-cluster/" for num in {7001..7006..1} do cd ${path}${num}/bin ./redis-server redis.conf done
      - 创建Redis集群(创建时Redis里不要有数据)
        # 本机ip 172.16.134.6 ./redis-cli --cluster create 172.16.134.6:7001 172.16.134.6:7002 172.16.134.6:7003 172.16.134.6:7004 172.16.134.6:7005 172.16.134.6:7006 --cluster-replicas 1
        
        –cluster-replicas : 集群的副本数，1代表1台master对应1台slave
    - 命令行客户端
      ./redis-cli -h 127.0.0.1 -p 7001 -c
      此处需要指定端口，
      -c: 表示以集群的方式连接
      # 查看集群信息 cluster info # 查看nodes信息 cluster nodes # 获取key，set key的时候如果不是分配给当前节点，会跳转
    - Smart智能客户端 JedisCluster是Jedis根据RedisCluster的特性提供的集群智能客户端
      JedisCluster为每个节点创建连接池，并跟节点建立映射关系缓存(Cluster slots)
      JedisCluster将每个主节点负责的槽位一一与主节点连接池建立映射缓存
      JedisCluster启动时，已经知道key,slot和node之间的关系，可以找到目标节点
      JedisCluster对目标节点发送命令，目标节点直接响应给JedisCluster
      如果JedisCluster与目标节点连接出错，则JedisCluster会知道连接的节点是一个错误的节点
      此时节点返回moved异常给JedisCluster
      JedisCluster会重新初始化slot与node节点的缓存关系，然后向新的目标节点发送命令，目标命令执行命令并向JedisCluster响应
      如果命令发送次数超过5次，则抛出异常”Too many cluster redirection!”
      - 示例代码
        @Test public void testCluster() { JedisPoolConfig config = new JedisPoolConfig(); Set<HostAndPort> jedisClusterNode = new HashSet<HostAndPort>(); jedisClusterNode.add(new HostAndPort("172.16.134.6", 7001)); jedisClusterNode.add(new HostAndPort("172.16.134.6", 7002)); jedisClusterNode.add(new HostAndPort("172.16.134.6", 7003)); jedisClusterNode.add(new HostAndPort("172.16.134.6", 7004)); jedisClusterNode.add(new HostAndPort("172.16.134.6", 7005)); jedisClusterNode.add(new HostAndPort("172.16.134.6", 7006)); JedisCluster jcd = new JedisCluster(jedisClusterNode, config); String name = jcd.get("name"); System.out.println(name); jcd.set("gfname", "Lara"); }
  - 分片不同节点分组服务于相互无交集的分片(sharding)，Redis Cluster 不存在单独的proxy或配置服务器，所以需要将客户端路由到目标的分片。
    
    moved和ask的区别
    1、moved:槽已确认转移
    2、ask:槽还在转移过程中
    - moved重定向
      1.每个节点通过通信都会共享Redis Cluster中槽和集群中对应节点的关系
      2.客户端向Redis Cluster的任意节点发送命令，接收命令的节点会根据CRC16规则进行hash运算与 16384取余，计算自己的槽和对应节点
      3.如果保存数据的槽被分配给当前节点，则去槽中执行命令，并把命令执行结果返回给客户端
      4.如果保存数据的槽不在当前节点的管理范围内，则向客户端返回moved重定向异常
      5.客户端接收到节点返回的结果，如果是moved异常，则从moved异常中获取目标节点的信息
      6.客户端向目标节点发送命令，获取命令执行结果
    - ask重定向
      在对集群进行扩容和缩容时，需要对槽及槽中数据进行迁移
      当客户端向某个节点发送命令，节点向客户端返回moved异常，告诉客户端数据对应的槽的节点信息
      如果此时正在进行集群扩展或者缩空操作，当客户端向正确的节点发送命令时，槽及槽中数据已经被迁移到别的节点了，就会返回ask，这就是ask重定向机制
      
      1.客户端向目标节点发送命令，目标节点中的槽已经迁移支别的节点上了，此时目标节点会返回ask转向给客户端
      2.客户端向新的节点发送Asking命令给新的节点，然后再次向新节点发送命令
      3.新节点执行命令，把命令执行结果返回给客户端
  - 迁移在RedisCluster中每个slot 对应的节点在初始化后就是确定的。在某些情况下，节点和分片需要变更:
    - 新的节点作为master加入;
    - 某个节点分组需要下线;
    - 负载不均衡需要调整slot 分布。
    此时需要进行分片的迁移，迁移的触发和过程控制由外部系统完成。包含下面 2 种:
    - 节点迁移状态设置:迁移前标记源/目标节点。
    - key迁移的原子化命令
      - 迁移的步骤
        1、向节点B发送状态变更命令，将B的对应slot 状态置为importing。
        2、向节点A发送状态变更命令，将A对应的slot 状态置为migrating。
        3、向A 发送migrate 命令，告知A 将要迁移的slot对应的key 迁移到B。
        4、当所有key 迁移完成后，cluster setslot 重新设置槽位。
  - 扩容
    - 添加主节点7007, 7008
    - 新增节点7007
      ./redis-cli --cluster add-node 172.16.134.6:7007 172.16.134.6:7001
      这个时候查看节点信息cluster nodes，可以看到新增的节点，但是没有分配slot。添加完主节点需要对主节点进行hash槽分配，这样该主节才可以存储数据。
    - 给刚添加的7007结点分配槽
      ./redis-cli --cluster reshard 172.16.134.6:7007
    - 添加从结点添加7008从结点，将7008作为7007的从结点
      # ./redis-cli --cluster add-node 新节点的ip和端口旧节点ip和端口 --cluster-slave --cluster-master-id 主节点id ./redis-cli --cluster add-node 172.16.134.6:7008 172.16.134.6:7007 --cluster-slave --cluster-master-id 7c8eafb470821a05c6567eb4139c87ee5287ad75
      注意:如果原来该结点在集群中的配置信息已经生成到cluster-config-file指定的配置文件中(如果 cluster-config-file没有指定则默认为nodes.conf)，这时可能会报错:
      [ERR] Node XXXXXX is not empty. Either the node already knows other nodes (check with CLUSTER NODES) or contains some key in database 0
      解决方法是删除生成的配置文件nodes.conf，删除后再执行./redis-cli –cluster add-node 指令
  - 缩容
    ./redis-cli --cluster del-node 172.16.134.6:7008 ba82b702fe38bba3e925a62ab404fbd31a8a4809
    删除已经占有hash槽的结点会失败，报错如下:
    [ERR] Node 192.168.127.128:7008 is not empty! Reshard data away and try again.
    需要将该结点占用的hash槽分配出去。
  - 容灾(failover)
    - 故障检测集群中的每个节点都会定期地(每秒)向集群中的其他节点发送PING消息如果在一定时间内(cluster-node-timeout)，发送ping的节点A没有收到某节点B的pong回应，则A将B标识为pfail。
      A在后续发送ping时，会带上B的pfail信息，通知给其他节点。
      如果B被标记为pfail的个数大于集群主节点个数的一半(N/2 + 1)时，B会被标记为fail，A向整个集群广播，该节点已经下线。
      其他节点收到广播，标记B为fail。
    - 从节点选举 raft，每个从节点，都根据自己对master复制数据的offset，来设置一个选举时间，offset越大(复制数据越多)的从节点，选举时间越靠前，优先进行选举。
      slave 通过向其他master发送FAILVOER_AUTH_REQUEST 消息发起竞选， master 收到后回复FAILOVER_AUTH_ACK 消息告知是否同意。
      slave 发送FAILOVER_AUTH_REQUEST 前会将currentEpoch 自增，并将最新的Epoch 带入到 FAILOVER_AUTH_REQUEST 消息中，如果自己未投过票，则回复同意，否则回复拒绝。
      
      所有的Master开始slave选举投票，给要进行选举的slave进行投票，如果大部分master node(N/2 + 1)都投票给了某个从节点，那么选举通过，那个从节点可以切换成master。
      RedisCluster失效的判定:
      1、集群中半数以上的主节点都宕机(无法投票)
      2、宕机的主节点的从节点也宕机了(slot槽分配不连续)
    - 变更通知当slave 收到过半的master 同意时，会成为新的master。此时会以最新的Epoch 通过PONG 消息广播自己成为master，让Cluster 的其他节点尽快的更新拓扑结构(node.conf)。
    - 主从切换
      - 自动切换就是上面讲的从节点选举
      - 手动切换人工故障切换是预期的操作，而非发生了真正的故障，目的是以一种安全的方式(数据无丢失)将当前 master节点和其中一个slave节点(执行cluster-failover的节点)交换角色
        
        1、向从节点发送cluster failover 命令(slaveof no one)
        2、从节点告知其主节点要进行手动切换(CLUSTERMSG_TYPE_MFSTART) 3、主节点会阻塞所有客户端命令的执行(10s)
        4、从节点从主节点的ping包中获得主节点的复制偏移量
        5、从节点复制达到偏移量，发起选举、统计选票、赢得选举、升级为主节点并更新配置
        6、切换完成后，原主节点向所有客户端发送moved指令重定向到新的主节点
        
        以上是在主节点在线情况下。
        如果主节点下线了，则采用cluster failover force或cluster failover takeover 进行强制切换。
  - 副本漂移我们知道在一主一从的情况下，如果主从同时挂了，那整个集群就挂了。为了避免这种情况我们可以做一主多从，但这样成本就增加了。
    Redis提供了一种方法叫副本漂移，这种方法既能提高集群的可靠性又不用增加太多的从机。
    - 举个例子
      Master1宕机，则Slaver11提升为新的Master1 集群检测到新的Master1是单点的(无从机)
      集群从拥有最多的从机的节点组(Master3)中，选择节点名称字母顺序最小的从机(Slaver31)漂移到单点的主从节点组(Master1)。
      具体流程如下(以上图为例):
      1、将Slaver31的从机记录从Master3中删除
      2、将Slaver31的的主机改为Master1 3、在Master1中添加Slaver31为从节点
      4、将Slaver31的复制源改为Master1 5、通过ping包将信息同步到集群的其他节点

Redis经典问题

缓存问题

缓存穿透一般的缓存系统，都是按照key去缓存查询，如果不存在对应的value，就应该去后端系统查找(比如 DB)。
缓存穿透是指在高并发下查询key不存在的数据，会穿过缓存查询数据库。导致数据库压力过大而宕机
解决方案:
- 缓存不存在的key 对查询结果为空的情况也进行缓存，缓存时间(ttl)设置短一点，或者该key对应的数据insert了之后清理缓存。
  
  问题:缓存太多空值占用了更多的空间
- 使用布隆过滤器在缓存之前在加一层布隆过滤器，在查询的时候先去布隆过滤器查询 key 是否
  存在，如果不存在就直接返回，存在再查缓存和DB
缓存雪崩当缓存服务器重启或者大量缓存集中在某一个时间段失效，这样在失效的时候，也会给后端系统(比如 DB)带来很大压力。
突然间大量的key失效了或redis重启，大量访问数据库，数据库崩溃

解决方案:
- key的失效期分散开不同的key设置不同的有效期
- 设置二级缓存(数据不一定一致)
- 高可用(脏读)
缓存击穿对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题:缓存被“击穿”的问题，这个和缓存雪崩的区别在于这里针对某一key缓存，前者则是很多key。
缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
解决方案:
- 用分布式锁控制访问的进程使用redis的setnx互斥锁先进行判断，这样其他线程就处于等待状态，保证不会有大并发操作去操作数据库。
- 不设超时时间不设超时时间, 但会造成写一致问题
  当数据库数据发生更新时，缓存中的数据不会及时更新，这样会造成数据库中的数据与缓存中的数据的不一致，应用会从缓存中读取到脏数据。可采用延时双删策略处理
数据不一致缓存和DB的数据不一致的根源 : 数据源不一样,DB数据更新了
如何解决：
- 延时双删 1、先更新数据库同时删除缓存项(key)，等读的时候再填充缓存
  2、2秒后再删除一次缓存项(key)
  3、设置缓存过期时间 Expired Time 比如 10秒或1小时
  4、将缓存删除失败记录到日志中，利用脚本提取失败记录再次删除(缓存失效期过长 7*24)
  升级方案：通过数据库的binlog来异步淘汰key，利用工具(canal)将binlog日志采集发送到MQ中，然后通过ACK机制确认处理删除缓存。

数据并发竞争

这里的并发指的是多个redis的client同时set 同一个key引起的并发问题。多客户端(Jedis)同时并发写一个key，一个key的值是1，本来按顺序修改为2,3,4，最后是4，但是顺序变成了4,3,2，最后变成了2。

分布式锁+时间戳
- 分布式锁
  - 整体技术方案
    这种情况，主要是准备一个分布式锁，大家去抢锁，抢到锁就做set操作。加锁的目的实际上就是把并行读写改成串行读写的方式，从而来避免资源竞争。
  - Redis分布式锁的实现主要用到的redis函数是setnx()
    用SETNX实现分布式锁
- 时间戳由于上面举的例子，要求key的操作需要顺序执行，所以需要保存一个时间戳判断set顺序。
```
系统A key 1 {ValueA 7:00}  
系统B key 1 { ValueB 7:05}  
```
  假设系统B先抢到锁，将key1设置为{ValueB 7:05}。接下来系统A抢到锁，发现自己的key1的时间戳早
  于缓存中的时间戳(7:00<7:05)，那就不做set操作了。
利用消息队列在并发量过大的情况下,可以通过消息中间件进行处理,把并行读写进行串行化。把Redis的set操作放在队列中使其串行化,必须的一个一个执行。

Hot Key

当有大量的请求(几十万)访问某个Redis某个key时，由于流量集中达到网络上限，从而导致这个redis的服务器宕机。造成缓存击穿，接下来对这个key的访问将直接访问数据库造成数据库崩溃，或者访问数据库回填Redis再访问Redis，继续崩溃。

如何发现热key 1、预估热key，比如秒杀的商品、火爆的新闻等
2、在客户端进行统计，实现简单，加一行代码即可
3、如果是Proxy，比如Codis，可以在Proxy端收集
4、利用Redis自带的命令，monitor、hotkeys。但是执行缓慢(不要用)
5、利用基于大数据领域的流式计算技术来进行实时数据访问次数的统计，比如 Storm、Spark Streaming、Flink，这些技术都是可以的。发现热点数据后可以写到zookeeper中
如何处理热Key 1、变分布式缓存为本地缓存
发现热key后，把缓存数据取出后，直接加载到本地缓存中。可以采用Ehcache、Guava Cache都可以，这样系统在访问热key数据时就可以直接访问自己的缓存了。(数据不要求时时一致)

2、在每个Redis主节点上备份热key数据，这样在读取时可以采用随机读取的方式，将访问压力负载到每个Redis上。

3、利用对热点数据访问的限流熔断保护措施
每个系统实例每秒最多请求缓存集群读操作不超过 400 次，一超过就可以熔断掉，不让请求缓存集群，
直接返回一个空白信息，然后用户稍后会自行再次重新刷新页面之类的。(首页不行，系统友好性差)
通过系统层自己直接加限流熔断保护措施，可以很好的保护后面的缓存集群。

Big Key

大key指的是存储的值(Value)非常大，常见场景:

热门话题下的讨论
大V的粉丝列表
序列化后的图片
没有及时处理的垃圾数据

大key的影响:

大key会大量占用内存，在集群中无法均衡
Redis的性能下降，主从复制异常
在主动删除或过期删除时会操作时间过长而引起服务阻塞
如何发现大key 1、redis-cli –bigkeys命令。可以找到某个实例5种数据类型(String、hash、list、set、zset)的最大key。
但如果Redis 的key比较多，执行该命令会比较慢
2、获取生产Redis的rdb文件，通过rdbtools分析rdb生成csv文件，再导入MySQL或其他数据库中进行分析统计，根据size_in_bytes统计bigkey
大key的处理优化big key的原则就是string减少字符串长度，list、hash、set、zset等减少成员数。

2、
4、
- string类型存到其他文档数据库 string类型的big key，尽量不要存入Redis中，可以使用文档型数据库MongoDB或缓存到CDN上。
  如果必须用Redis存储，最好单独存储，不要和其他的key一起存储。采用一主一从或多从。
- hash,set,zset,list 元素分拆单个简单的key存储的value很大，可以尝试将对象分拆成几个key-value，使用mget获取值，这样分拆的意义在于分拆单次操作的压力，将操作压力平摊到多次操作中，降低对redis的IO影响。
  hash，set, zset, list 中存储过多的元素，可以将这些元素分拆。(常见)
```
以hash类型举例来说，对于field过多的场景，可以根据field进行hash取模，生成一个新的key，例如原 来的  
hash_key:{filed1:value, filed2:value, filed3:value ...}，可以hash取模后形成如下 key:value形式  
hash_key:1:{filed1:value}  
hash_key:2:{filed2:value}  
hash_key:3:{filed3:value}  
...  
取模后，将原先单个key分成多个key，每个key filed个数为原先的1/N  
```
- 使用unlink删除key 删除大key时不要使用del,因为del是阻塞命令，删除时会影响性能。
  可以使用 unlink命令(lazy delete)
  删除指定的key(s),若key不存在则该key被跳过。该命令会在另一个线程中回收内存，因此它是非阻塞的。这也是该命令名字的由来:仅将keys从key空间中删除，真正的数据删除会在后续异步操作。
```
redis> SET key1 "Hello"  
"OK"  
redis> SET key2 "World"  
"OK"  
redis> UNLINK key1 key2 key3  
(integer) 2  
```

分布式锁

利用Watch实现Redis乐观锁

乐观锁(当前没人就改，有人改就算了)基于CAS(Compare And Swap)思想(比较并替换)，是不具有互斥性，不会产生锁等待而消耗资源，但是需要反复的重试，但也是因为重试的机制，能比较快的响应。因此我们可以利用redis来实现乐观锁。具体思路如下:
1、利用redis的watch功能，监控这个redisKey的状态值
2、获取redisKey的值
3、创建redis事务
4、给这个key的值+1
5、然后去执行这个事务，如果key的值被修改过则提交失败，key不加1

示例代码

public class CasRedisDistributeLock {  
    public static void main(String[] args) {  
        final String redisKey = "lock";  
	    
        ExecutorService executorService = Executors.newFixedThreadPool(20);  
	    
        Jedis jedis = new Jedis("centos7-3", 6379);  
        // 初始值  
        jedis.set(redisKey, "0");  
        jedis.close();  
	    
        for (int i = 0; i < 1000; i++) {  
            final int finalI = i;  
            executorService.execute(new Runnable() {  
                public void run() {  
                    Jedis jedis1 = new Jedis("centos7-3", 6379);  
	    
                    jedis1.watch(redisKey);  
	    
                    Integer value = Integer.valueOf(jedis1.get(redisKey));  
                    String user = String.format("%03d", finalI);  
	    
                    if (value < 20) {  
                        Transaction transaction = jedis1.multi();  
                        transaction.incr(redisKey);  
                        List<Object> list = transaction.exec();  
	    
                        if (list != null && list.size() > 0) {  
	    
                            System.out.printf("user->%s, success! Current user-> %d\n", user, value + 1);  
                        } else {  
                            System.out.printf("user->%s, failed \n", user);  
                        }  
                    }  
	    
                    jedis1.close();  
	    
                }  
            });  
	    
        }  
        executorService.shutdown();  
	    
    }  
}  
	    

利用setnx实现分布式锁
实现原理：共享资源互斥，共享资源串行化
分布式应用中使用锁:(多进程多线程) 分布式锁是控制分布式系统之间同步访问共享资源的一种方式。利用Redis的单线程特性对共享资源进行串行化处理
- 获取锁
```
/**  
* 使用redis的set命令实现获取分布式锁  
* @param lockKey key就是锁  
* @param requestId 请求ID，保证同一性 uuid+threadID  
* @param expireTime 过期时间，避免死锁  
* @return  
*/  
public boolean getLock(String lockKey,String requestId,int expireTime) {   
    // NX:保证互斥性  
    // hset 原子性操作 只要lockKey有效 则说明有进程在使用分布式锁  
    String result = jedis.set(lockKey, requestId, "NX", "EX", expireTime);   
    if("OK".equals(result)) {  
        return true;  
    }  
    return false;  
}  
```
- 释放锁
```
//使用lua脚本可以保证操作的原子性  
public static boolean releaseLock(String lockKey, String requestId) {  
    String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";  
    Object result = jedis.eval(script, Collections.singletonList(lockKey),Collections.singletonList(requestId));  
    if (result.equals(1L)) {  
        return true;  
    }  
    return false;      
}  
```
- 存在问题
  单机: 无法保证高可用
  主–从: 无法保证数据的强一致性，在主机宕机时会造成锁的重复获得。
  无法续租: 超过expireTime后，不能继续使用
- 本质分析 CAP模型
  - 一致性（Consistency)（所有节点在同一时间具有相同的数据）
  - 可用性（Availability）（保证每个请求不管成功或者失败都有响应）
  - 分隔容忍（Partition tolerance）（系统中任意信息的丢失或失败不会影响系统的继续运作）
  在分布式环境下不可能满足三者共存，只能满足其中的两者共存，在分布式下P不能舍弃(舍弃P就是单
  机了)。
  所以只能是CP(强一致性模型)和AP(高可用模型)。
  
  分布式锁是CP模型，Redis集群是AP模型。 (base)
  Redis集群不能保证数据的随时一致性，只能保证数据的最终一致性。
  
  为什么还可以用Redis实现分布式锁?
  与业务有关, 当业务不需要数据强一致性时，比如:社交场景，就可以使用Redis实现分布式锁;
  当业务必须要数据的强一致性，即不允许重复获得锁，比如金融场景(重复下单，重复转账)就不要使用
  
  可以使用CP模型实现，比如:zookeeper和etcd。

Redisson分布式锁 Redisson是架设在Redis基础上的一个Java驻内存数据网格(In-Memory Data Grid)。 Redisson在基于NIO的Netty框架上，生产环境使用分布式锁。

示例代码

导入依赖

<!-- https://mvnrepository.com/artifact/org.redisson/redisson -->  
<dependency>  
    <groupId>org.redisson</groupId>  
    <artifactId>redisson</artifactId>  
    <version>3.12.0</version>  
</dependency>  

Redisson锁工具类

import org.redisson.Redisson;  
import org.redisson.api.RLock;  
import org.redisson.config.Config;  
import java.util.concurrent.TimeUnit;  
		    
		    
public class RedissonDeistributeLock {  
    private static Redisson redisson = RedissonManager.getRedisson();  
    private static final  String LOCK_TITLE = "redisLock_";  
		    
    public static boolean acquire(String lockName) {  
        String key = LOCK_TITLE + lockName;  
		    
        RLock lock = redisson.getLock(key);  
		    
        //加锁，并且设置锁过期时间3秒，防止死锁的产生  
        lock.lock(3, TimeUnit.SECONDS);  
		    
        return true;  
    }  
		    
    public static void release(String lockName) {  
        String key = LOCK_TITLE + lockName;  
        RLock lock = redisson.getLock(key);  
        lock.unlock();  
    }  
		    
    /**  
     * 获取Redisson客户端的工具类  
     */  
    static class RedissonManager {  
        private static Config config = new Config();  
        private static Redisson redisson = null;  
		    
        static {  
            config.useClusterServers()  
                    // 扫描集群的周期  
                    .setScanInterval(2000)  
                    .addNodeAddress("redis://centos7-3:7001")  
                    .addNodeAddress("redis://centos7-3:7002")  
                    .addNodeAddress("redis://centos7-3:7003")  
                    .addNodeAddress("redis://centos7-3:7004")  
                    .addNodeAddress("redis://centos7-3:7005")  
                    .addNodeAddress("redis://centos7-3:7006")  
                    .addNodeAddress("redis://centos7-3:7007");  
		    
            redisson = (Redisson) Redisson.create(config);  
        }  
		    
        public static Redisson getRedisson() {  
            return redisson;  
        }  
    }  
}  
		    

测试类

import redis.clients.jedis.Jedis;  
import java.util.concurrent.ExecutorService;  
import java.util.concurrent.Executors;  
		    
public class RedissonLockTest {  
    public static void main(String[] args) {  
        final String key = "lock001";  
		    
        Jedis jedis = new Jedis("centos7-3", 6379);  
        jedis.set("num", "0");  
        jedis.close();  
		    
        ExecutorService executorService = Executors.newFixedThreadPool(20);  
		    
        for (int i = 0; i < 1000; i++) {  
            final int finalI = i;  
            executorService.execute(new Runnable() {  
                public void run() {  
                    // 加锁  
                    RedissonDeistributeLock.acquire(key);  
		    
                    Jedis jedis1 = new Jedis("centos7-3", 6379);  
                    Integer num = Integer.valueOf(jedis1.get("num"));  
		    
                    if (num < 20) {  
                        jedis1.incr("num");  
                        System.out.println(finalI + " : " + Integer.valueOf(jedis1.get("num")));  
                    } else {  
                        System.out.println(finalI);  
                    }  
		    
                    // 解锁  
                    RedissonDeistributeLock.release(key);  
                }  
            });  
		    
        }  
		    
        executorService.shutdown();  
    }  
}  
		    

实现原理

加锁机制如果该客户端面对的是一个redis cluster集群，他首先会根据hash节点选择一台机器。
发送lua脚本到redis服务器上，脚本如下:
```
"if (redis.call('exists',KEYS[1])==0) then "+ //看有没有锁   
"redis.call('hset',KEYS[1],ARGV[2],1) ; "+ //无锁 加锁   
"redis.call('pexpire',KEYS[1],ARGV[1]) ; "+  
"return nil; end ;" +  
"if (redis.call('hexists',KEYS[1],ARGV[2]) ==1 ) then "+ //我加的锁   
"redis.call('hincrby',KEYS[1],ARGV[2],1) ; "+ //重入锁   
"redis.call('pexpire',KEYS[1],ARGV[1]) ; "+  
"return nil; end ;" +  
"return redis.call('pttl',KEYS[1]) ;" //不能加锁，返回锁的时间  
```
lua的作用:保证这段复杂业务逻辑执行的原子性。

KEYS[1]) : 加锁的key
ARGV[1] : key的生存时间，默认为30秒
ARGV[2] : 加锁的客户端ID (UUID.randomUUID() + “:” + threadId)
- 锁互斥机制
- 自动延时机制只要客户端1一旦加锁成功，就会启动一个watch dog看门狗，他是一个后台线程，会每隔10秒检查一下，如果客户端1还持有锁key，那么就会不断的延长锁key的生存时间
- 可重入锁机制

释放锁机制执行lua脚本如下:

// 如果key已经不存在，说明已经被解锁，直接发布(publish)redis消息   
"if (redis.call('exists', KEYS[1]) == 0) then " +  
                        "redis.call('publish', KEYS[2], ARGV[1]); " +  
                        "return 1; " +  
                    "end;" +  
// key和field不匹配，说明当前客户端线程没有持有锁，不能主动解锁。 不是我加的锁 不能解锁  
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +  
// 将value减1  
    "return nil;" +  
"end; " +  
"local counter = redis.call('hincrby', KEYS[1], ARGV[3],  
-1); " +  
// 如果counter>0说明锁在重入，不能删除key  
                    "if (counter > 0) then " +  
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +  
                        "return 0; " +  
// 删除key并且publish 解锁消息 "else " +  
"redis.call('del', KEYS[1]); " + //删除锁   
"redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+  
                    "end; " +  
                    "return nil;",  

– KEYS[1] :需要加锁的key，这里需要是字符串类型。
– KEYS[2] :redis消息的ChannelName,一个分布式锁对应唯一的一个channelName:“redisson_lockchannel{” + getName() + “}”
– ARGV[1] :reids消息体，这里只需要一个字节的标记就可以，主要标记redis的key已经解锁，再结合redis的Subscribe，能唤醒其他订阅解锁消息的客户端线程申请锁。
– ARGV[2] :锁的超时时间，防止死锁
– ARGV[3] :锁的唯一标识，也就是刚才介绍的 id(UUID.randomUUID()) + “:” + threadId

如果执行lock.unlock()，就可以释放分布式锁，此时的业务逻辑也是非常简单的。其实说白了，就是每次都对myLock数据结构中的那个加锁次数减1。如果发现加锁次数是0了，说明这个客户端已经不再持有锁了，此时就会用:
“del myLock”命令，从redis里删除这个key。然后呢，另外的客户端2就可以尝试完成加锁了。

分布式锁特性
- 互斥性任意时刻，只能有一个客户端获取锁，不能同时有两个客户端获取到锁。
  Redis : setnx set key value NX 如果key存在就不设置
- 同一性锁只能被持有该锁的客户端删除，不能由其它客户端删除。
  Redis : lua 实现原子性
- 可重入性持有某个锁的客户端可继续对该锁加锁，实现锁的续租
- 容错性锁失效后(超过生命周期)自动释放锁(key失效)，其他客户端可以继续获得该锁，防止死锁
  expire 设置超时时间
  set key value NX PX

底层数据结构

基本架构

Redis没有表的概念，Redis实例所对应的db以编号区分，db本身就是key的命名空间。比如:user:1000作为key值，表示在user这个命名空间下id为1000的元素，类似于user表的id=1000的行。

RedisDB结构

Redis中存在“数据库”的概念，该结构由server.h中的redisDb定义。当redis 服务器初始化时，会预先分配 16 个数据库, 所有数据库保存到结构 redisServer 的一个成员 redisServer.db 数组中, redisClient中存在一个名叫db的指针指向当前使用的数据库
RedisDB结构体源码:

typedef struct redisDb {  
    dict *dict;                 /* The keyspace for this DB */  
    dict *expires;              /* Timeout of keys with a timeout set */  
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP)*/  
    dict *ready_keys;           /* Blocked keys that received a PUSH */  
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */  
    int id;                     /* Database ID 为0-15(默认Redis有16个数据库)*/  
    long long avg_ttl;          /* Average TTL, just for stats */  
    list *defrag_later;         /* List of key names to attempt to defrag one by one, gradually. */  
} redisDb;  

RedisObject结构

Value是一个包含字符串，列表，哈希，集合和有序集合等结构的对象

typedef struct redisObject {  
    unsigned type:4;  
    unsigned encoding:4;  
    unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or  
                            * LFU data (least significant 8 bits frequency  
                            * and most significant 16 bits access time). */  
    int refcount;  
    void *ptr;  
} robj;  

4位type type 字段表示对象的类型，占 4 位;
REDIS_STRING(字符串)、REDIS_LIST (列表)、REDIS_HASH(哈希)、REDIS_SET(集合)、REDIS_ZSET(有序集合)。
当我们执行 type 命令时，便是通过读取 RedisObject 的 type 字段获得对象的类型
```
127.0.0.1:6379> type a1  
string  
```
4位encoding encoding 表示对象的内部编码，占 4 位, 每个对象有不同的实现编码
Redis 可以根据不同的使用场景来为对象设置不同的编码，大大提高了 Redis 的灵活性和效率。通过 object encoding 命令，可以查看对象采用的编码方式
```
127.0.0.1:6379> set name april  
OK  
127.0.0.1:6379> set age 19  
OK  
127.0.0.1:6379> TYPE name  
string  
127.0.0.1:6379> type age  
string  
127.0.0.1:6379> object encoding name  
"embstr"  
127.0.0.1:6379> object encoding age  
"int"  
```
24位LRU lru 记录的是对象最后一次被命令程序访问的时间，( 4.0 版本占 24 位，2.6 版本占 22 位)。
高16位存储一个分钟数级别的时间戳，低8位存储访问计数(lfu : 最近访问次数)
- lru—-> 高16位: 最后被访问的时间
- lfu—–>低8位:最近访问次数
refcount refcount 记录的是该对象被引用的次数，类型为整型。
refcount 的作用，主要在于对象的引用计数和内存回收。
当对象的refcount>1时，称为共享对象
Redis 为了节省内存，当有一些对象重复出现时，新的程序不会创建新的对象，而是仍然使用原来的对
象。
ptr ptr 指针指向具体的数据，比如:set hello world，ptr 指向包含字符串 world 的 SDS。

7种type

字符串对象
C语言: 字符数组 “\0”
Redis 使用了 SDS(Simple Dynamic String)。用于存储字符串和整型数据。
```
struct sdshdr{   
//记录buf数组中已使用字节的数量 int len;  
//记录 buf 数组中未使用字节的数量 int free;  
//字符数组，用于保存字符串  
char buf[];  
}  
```
buf[] 的长度=len+free+1
- SDS的优势:
  
  1、SDS 在 C 字符串的基础上加入了 free 和 len 字段，获取字符串长度:SDS 是 O(1)，C 字符串是 O(n)。
  buf数组的长度=free+len+1
  2、 SDS 由于记录了长度，在可能造成缓冲区溢出时会自动重新分配内存，杜绝了缓冲区溢出。
  3、可以存取二进制数据，以字符串长度len来作为结束标识
  C: \0 空字符串二进制数据包括空字符串，所以没有办法存取二进制数据
  SDS : 非二进制 \0 二进制: 字符串长度可以存二进制数据
- 使用场景:
  
  SDS的主要应用在:存储字符串和整型数据、存储key、AOF缓冲区和用户输入缓冲。
跳跃表跳跃表是有序集合(sorted-set)的底层实现，效率高，实现简单。
跳跃表的基本思想:将有序链表中的部分节点分层，每一层都是一个有序链表。
- 查找在查找时优先从最高层开始向后查找，当到达某个节点时，如果next节点值大于要查找的值或next指针指向null，则从当前节点下降一层继续向后查找。
  这种数据结构，就是跳跃表，它具有二分查找的功能。
- 插入上面例子中，9个结点，一共4层，是理想的跳跃表。概率性插入，通过抛硬币(概率1/2)的方式来决定新插入结点跨越的层数:
  正面:插入上层
  背面:不插入
- 删除找到指定元素并删除每层的该元素即可
- 跳跃表特点每层都是一个有序链表，查找次数近似于层数(1/2) 。底层包含所有元素，
  空间复杂度 O(n) 扩充了一倍
- Redis跳跃表的实现
```
//跳跃表节点  
typedef struct zskiplistNode {  
    sdsele;/*存储字符串类型数据redis3.0版本中使用robj类型表示，  
但是在redis4.0.1中直接使用sds类型表示*/  
    double score;//存储排序的分值  
    struct zskiplistNode *backward;//后退指针，指向当前节点最底层的前一个节点  
    struct zskiplistLevel {  
        struct zskiplistNode *forward;//指向本层下一个节点  
        unsigned int span; //本层下个节点到本节点的元素个数  
        }level [];  
} zskiplistNode;  
	    
//链表  
typedef struct zskiplist{  
    //表头节点和表尾节点  
    struct zskiplistNode *header, *tail;   
    //表中节点的数量  
    unsigned long length;  
    //表中层数最大的节点的层数  
    int level;  
}zskiplist;  
```
- 完整的跳跃表结构体
- 跳跃表的优势 1、可以快速查找到需要的节点
  2、可以在O(1)的时间复杂度下，快速获得跳跃表的头节点、尾结点、长度和高度。
  应用场景:有序集合的实现
字典字典dict又称散列表(hash)，是用来存储键值对的一种数据结构。
Redis整个数据库是用字典来存储的。(K-V结构)
对Redis进行CURD操作其实就是对字典中的数据进行CURD操作。
- 数组数组:用来存储数据的容器，采用头指针+偏移量的方式能够以O(1)的时间复杂度定位到数据所在的内存地址。
  Redis 海量存储->快
- Hash函数 Hash(散列)，作用是把任意长度的输入通过散列算法转换成固定类型、固定长度的散列值。 hash函数可以把Redis里的key:包括字符串、整数、浮点数统一转换成整数。
  key=100.1 String “100.1” 5位长度的字符串
  
  Redis-cli :times 33
  Redis-Server : siphash
- 数组下标 hash(key)%数组容量(hash值%数组容量得到的余数)
- Hash冲突不同的key经过计算后出现数组下标一致，称为Hash冲突。采用单链表在相同的下标位置处存储原始key和value,当根据key找Value时，找到数组下标，遍历单链表可以找出key相同的value
- 存取原理
- Redis字典的实现
  Redis字典实现包括:字典(dict)、Hash表(dictht)、Hash表节点(dictEntry)。
  - Redis字典数据结构
  - Hash表
    typedef struct dictht { dictEntry **table; //哈希表数组 unsigned long size; //哈希表数组的大小 unsigned long sizemask; //用于映射位置的掩码，值永远等于(size-1) unsigned long used; //哈希表已有节点的数量,包含next单链表数据 }dictht;
    1、hash表的数组初始容量为4，随着k-v存储量的增加需要对hash表数组进行扩容，新扩容量为当前量的一倍，即4,8,16,32
    2、索引值=Hash值&掩码值(Hash值与Hash表容量取余)
  - Hash表节点
    typedef struct dictEntry { void *key; //键 union { //值v的类型可以是以下4种类型 void *val; uint64_tu64; int64_ts64; double d; } v; struct dictentry *next; //指向下一个哈希表节点，形成单向链表解决hash冲突 } dictEntry;
    key字段存储的是键值对中的键, v字段是个联合体，存储的是键值对中的值。 next指向下一个哈希表节点，用于解决hash冲突
  - dict字典
    typedef struct dict { //该字典对应的特定操作函数 dictType *type; //上述类型函数对应的可选参数 void *privdata; /*两张哈希表，存储键值对数据，ht[0]为原生哈希表， ht[1]为rehash哈希表 */ dictht ht[2]; /*rehash标识当等于-1时表示没有在rehash, 否则表示正在进行rehash操作，存储的值表示 hash表ht[0]的rehash进行到哪个索引值(数组下标)*/ long rehashidx; //当前运行的迭代器数量 int iterators; } dict;
  - type字段指向dictType结构体，里边包括了对该字典操作的函数指针
    typedef struct dictType { //计算哈希值的函数 unsigned int (*hashFunction) (const void *key); //复制键的函数 void *(*keyDup) (void *privdata, const void *key); //复制值的函数 void *(*va1Dup) (void *privdata, const void *obj); //比较键的函数 int (*keyCompare) (void *privdata, const void *key1, const void *key2); //销毁键的函数 void (*keyDestructor) (void *privdata, void *key); // 销毁值的函数 void (*valDestructor) (void *privdata, void *obj); } dictType;
    Redis字典除了主数据库的K-V数据存储以外，还可以用于:散列表对象、哨兵模式中的主从节点管理等在不同的应用中，字典的形态都可能不同，dictType是为了实现各种形态的字典而抽象出来的操作函数 (多态)。
  - 字典扩容
    字典达到存储上限，需要rehash(扩容)
    1. 初次申请默认容量为4个dictEntry，非初次申请为当前hash表容量的一倍。
    2. rehashidx=0表示要进行rehash操作。
    3. 新增加的数据在新的hash表h[1]
    4. 修改、删除、查询在老hash表h[0]、新hash表h[1]中(rehash中)
    5. 将老的hash表h[0]的数据重新计算索引值后全部迁移到新的hash表h[1]中，这个过程称为 rehash。
    - 渐进式rehash 当数据量巨大时rehash的过程是非常缓慢的，所以需要进行优化。服务器忙，则只对一个节点进行rehash, 服务器闲，可批量rehash(100节点)
- 应用场景 1、主数据库的K-V数据存储
  2、散列表对象(hash)
  3、哨兵模式中的主从节点管理
压缩列表压缩列表(ziplist)是由一系列特殊编码的连续内存块组成的顺序型数据结构
-> 节省内存
是一个字节数组，可以包含多个节点(entry)。每个节点可以保存一个字节数组或一个整数。
- 数据结构
  zlbytes:压缩列表的字节长度
  zltail:压缩列表尾元素相对于压缩列表起始地址的偏移量
  zllen:压缩列表的元素个数
  entry1..entryX : 压缩列表的各个节点 zlend:压缩列表的结尾，占一个字节，恒为0xFF(255)
  - entryX元素
    previous_entry_length:前一个元素的字节长度
    encoding:表示当前元素的编码 content:数据内容
- 代码实现
```
typedef struct zlentry {  
    //previous_entry_length字段的长度  
    unsigned int prevrawlensize;   
    //previous_entry_length字段存储的内容  
    unsigned int prevrawlen;  
    //encoding字段的长度  
    unsigned int lensize;   
    //数据内容长度  
    unsigned int len;  
    //当前元素的首部长度，即previous_entry_length字段长度与encoding字段长度之和。  
    unsigned int headersize;   
    //数据类型  
    unsigned char encoding;  
    //当前元素首地址  
    unsigned char *p;   
} zlentry;  
```
- 应用场景
  - sorted-set和hash元素个数少且是小整数或短字符串(直接使用)
  - list用快速链表(quicklist)数据结构存储，而快速链表是双向列表与压缩列表的组合。(间接使用)
整数集合整数集合(intset)是一个有序的(整数升序)、存储整数的连续存储结构。当Redis集合类型的元素都是整数并且都处在64位有符号整数范围内(2^64)，使用该结构体存储。
- intset的结构图
```
typedef struct intset{   
	//编码方式  
	uint32_t encoding;   
	//集合包含的元素数量   
	uint32_t length;   
	//保存元素的数组   
	int8_t contents[];  
} intset;  
```
- 应用场景可以保存类型为int16_t、int32_t 或者int64_t 的整数值，并且保证集合中不会出现重复元素。

快速列表快速列表(quicklist)是Redis底层重要的数据结构。是列表的底层实现。(在Redis3.2之前，Redis采用双向链表(adlist)和压缩列表(ziplist)实现)在Redis3.2以后结合adlist和ziplist的优势Redis设计出了quicklist。

双向列表(adlist)
双向链表优势:
1. 双向:链表具有前置节点和后置节点的引用，获取这两个节点时间复杂度都为O(1)。
2. 普通链表(单链表):节点类保留下一节点的引用。链表类只保留头节点的引用，只能从头节点插入删除
3. 无环:表头节点的 prev 指针和表尾节点的 next 指针都指向 NULL,对链表的访问都是以 NULL 结束。
  环状:头的前一个节点指向尾节点
4. 带链表长度计数器:通过 len 属性获取链表长度的时间复杂度为 O(1)。
5. 多态:链表节点使用 void* 指针来保存节点值，可以保存各种不同类型的值。

quicklist的结构
quicklist是一个双向链表，链表中的每个节点时一个ziplist结构。quicklist中的每个节点ziplist都能够存储多个数据元素。

quicklist

typedef struct quicklist {  
    //指向quicklist的头部  
    quicklistNode *head;  
    //指向quicklist的尾部  
    quicklistNode *tail;  
    //列表中所有数据项的个数总和  
    unsigned long count;  
    // quicklist节点的个数，即ziplist的个数  
    unsigned int len;  
    // ziplist大小限定，由1ist-max-ziplist-size给定(Redis设定)  
    int fill:16;  
    //节点压缩深度设置，由list-compress-depth给定(Redis 设定)  
    unsigned int compress:16;  
} quicklist;  

quicklistNode

typedef struct quicklistNode {  
    //指向上一个ziplist节点  
    struct quicklistNode *prev;   
    //指向下一个ziplist节点  
    struct quicklistNode *next;   
    //数据指针，如果没有被压缩，就指向ziplist结构，反之指向quick1istLZF结构  
    unsigned char *z1;  
    //表示指向ziplist结构的总长度(内存占用长度)  
    unsigned int sz;  
    //表示ziplist中的数据项个数  
    unsigned int count:16;   
    //编码方式，1--ziplist, 2--quicklistLZF  
    unsigned int encoding:2;   
    //预留字段，存放数据的方式，1--NONE，2--ziplistunsigned  
    unsigned int container:2;   
    //解压标记，当查看一个被压缩的数据时，需要暂时解压，标记此参数为1，之后再重新进行压缩  
    int recompress:1;   
    //测试相关  
    unsigned int attempted_compress:1;  
    //扩展字段，暂时没用  
    unsigned int extra:10;  
} quicklistNode;  

quicklistLZF quicklist每个节点的实际数据存储结构为ziplist，这种结构的优势在于节省存储空间。为了进一步降低 ziplist的存储空间，还可以对ziplist进行压缩。Redis采用的压缩算法是LZF。其基本思想是:数据与前面重复的记录重复位置及长度，不重复的记录原始数据。
压缩过后的数据可以分成多个片段，每个片段有两个部分:解释字段和数据字段。
```
typedef struct quicklistLZF {  
	unsigned int sz; // LZF压缩后占用的字节数   
	char compressed[]; // 柔性数组，指向数据部分  
} quicklistLZF;  
```

应用场景列表(List)的底层实现、发布与订阅、慢查询、监视器等功能。

流对象
stream主要由:消息、生产者、消费者和消费组构成。
Redis Stream的底层主要使用了listpack(紧凑列表)和Rax树(基数树)
- listpack
  listpack表示一个字符串列表的序列化，listpack可用于存储字符串或整数。用于存储stream的消息内容。
- Rax树
  Rax 是一个有序字典树 (基数树 Radix Tree)，按照 key 的字典序排列，支持快速地定位、插入和删除操作。
  Rax 被用在 Redis Stream 结构里面用于存储消息队列，在 Stream 里面消息 ID 的前缀是时间戳 + 序号，这样的消息可以理解为时间序列消息。使用 Rax 结构进行存储就可以快速地根据消息 ID 定位到具体的消息，然后继续遍历指定消息之后的所有消息。
- 应用场景 stream的底层实现

10种encoding

encoding 表示对象的内部编码，占 4 位。
Redis通过 encoding 属性为对象设置不同的编码
对于少的和小的数据，Redis采用小的和压缩的存储方式，体现Redis的灵活性
大大提高了 Redis 的存储量和执行效率

String 底层是可能是int，embstr，raw
- int REDIS_ENCODING_INT(int类型的整数)
- embstr REDIS_ENCODING_EMBSTR(编码的简单动态字符串) 小字符串，长度小于44个字节
- raw REDIS_ENCODING_RAW (简单动态字符串) 大字符串，长度大于44个字节
list
- quicklist 列表的编码是quicklist。 REDIS_ENCODING_QUICKLIST(快速列表)
hash 散列的编码是字典和压缩列表
- ziplist REDIS_ENCODING_ZIPLIST(压缩列表) 。当散列表元素的个数比较少，且元素都是小整数或短字符串时。
- hashtable REDIS_ENCODING_HT(字典)
  当散列表元素的个数比较多或元素不是小整数或短字符串时。
set 集合的编码是整形集合和字典
- hashtable REDIS_ENCODING_HT(字典)
  当散列表元素的个数比较多或元素不是小整数或短字符串时。
- intset REDIS_ENCODING_INTSET(整数集合) 当Redis集合类型的元素都是整数并且都处在64位有符号整数范围内(<18446744073709551616)
zset 有序集合的编码是压缩列表和跳跃表+字典
- ziplist REDIS_ENCODING_ZIPLIST(压缩列表) 。当散列表元素的个数比较少，且元素都是小整数或短字符串时。
- skiplist + hashtable REDIS_ENCODING_SKIPLIST(跳跃表+字典) 当元素的个数比较多或元素不是小整数或短字符串时。