123

[catcatbai]

redis

应用场景：

缓存分布式会话分布式锁最新列表消息系统

Redis支持多个数据库，并且每个数据库的数据是隔离的不能共享，并且基于单机才有，如果是集群就没有数据库的概念。

Redis是一个字典结构的存储服务器，而实际上一个Redis实例提供了多个用来存储数据的字典，客户端可以指定将数据存储在哪个字典中。这与我们熟知的在一个关系数据库实例中可以创建多个数据库类似，所以可以将其中的每个字典都理解成一个独立的数据库。

每个数据库对外都是一个从0开始的递增数字命名，Redis默认支持16个数据库（可以通过配置文件支持更多，无上限），可以通过配置databases来修改这一数字。客户端与Redis建立连接后会自动选择0号数据库，不过可以随时使用SELECT命令更换数据库，如要选择1号数据库：

redis> SELECT 1
OK
redis [1] > GET foo
(nil)

然而这些以数字命名的数据库又与我们理解的数据库有所区别。首先Redis不支持自定义数据库的名字，每个数据库都以编号命名，开发者必须自己记录哪些数据库存储了哪些数据。另外Redis也不支持为每个数据库设置不同的访问密码，所以一个客户端要么可以访问全部数据库，要么连一个数据库也没有权限访问。最重要的一点是多个数据库之间并不是完全隔离的，比如FLUSHALL命令可以清空一个Redis实例中所有数据库中的数据。综上所述，这些数据库更像是一种命名空间，而不适宜存储不同应用程序的数据。比如可以使用0号数据库存储某个应用生产环境中的数据，使用1号数据库存储测试环境中的数据，但不适宜使用0号数据库存储A应用的数据而使用1号数据库B应用的数据，不同的应用应该使用不同的Redis实例存储数据。由于Redis非常轻量级，一个空Redis实例占用的内在只有1M左右，所以不用担心多个Redis实例会额外占用很多内存。

# 下载代码，要换到 tag 发布分支。
# 运行服务器， daemonize 表示在后台运行
> ./redis-server --daemonize yes

多线程机制

http://www.redis.cn/topics/distlock.html

Redis 6.0 多线程连环13问

https://www.cnblogs.com/mumage/p/12832766.html

以下内容，都是看的上边链接的内容，哈哈，感觉挺有用的。特别是那个彩蛋，估计很多人都不知道吧。

文章下边的那些引用的文章，感觉也可以看看，因该收益也很大。例如：看敖炳的视频的时候，看到它写文章还是做视频来着，说的那些数，感觉看了真的很不错里。《Redis设计与实现》、《Redis深度历险：核心原理和应用实践》都很有用。当时很多数，但是当时就下了这两个的pdf。哈哈

Redis6.0之前的版本真的是单线程吗？

并不是单线程，只是Redis在处理客户端的请求时，包括获取 (socket 读)、解析、执行、内容返回 (socket 写) 等都由一个顺序串行的主线程处理，这就是所谓的“单线程”。但如果严格来讲从Redis4.0之后并不是单线程，除了主线程外，它也有后台线程在处理一些较为缓慢的操作，例如清理脏数据、无用连接的释放、大 key 的删除等等。还有用于备份的子进程。

Redis6.0之前为什么一直不使用多线程？

官方曾做过类似问题的回复：使用Redis时，几乎不存在CPU成为瓶颈的情况， Redis主要受限于内存和网络。例如在一个普通的Linux系统上，Redis通过使用pipelining每秒可以处理100万个请求，所以如果应用程序主要使用O(N)或O(log(N))的命令，它几乎不会占用太多CPU。

使用了单线程后，可维护性高。多线程模型虽然在某些方面表现优异，但是它却引入了程序执行顺序的不确定性，带来了并发读写的一系列问题，增加了系统复杂度、同时可能存在线程切换、甚至加锁解锁、死锁造成的性能损耗。Redis通过AE事件模型以及IO多路复用等技术，处理性能非常高，因此没有必要使用多线程。单线程机制使得 Redis 内部实现的复杂度大大降低，Hash 的惰性 Rehash、Lpush 等等 “线程不安全” 的命令都可以无锁进行。

Redis6.0为什么要引入多线程呢？

Redis将所有数据放在内存中，内存的响应时长大约为100纳秒，对于小数据包，Redis服务器可以处理80,000到100,000 QPS，这也是Redis处理的极限了，对于80%的公司来说，单线程的Redis已经足够使用了。

但随着越来越复杂的业务场景，有些公司动不动就上亿的交易量，因此需要更大的QPS。常见的解决方案是在分布式架构中对数据进行分区并采用多个服务器，但该方案有非常大的缺点，例如要管理的Redis服务器太多，维护代价大；某些适用于单个Redis服务器的命令不适用于数据分区；数据分区无法解决热点读/写问题；数据偏斜，重新分配和放大/缩小变得更加复杂等等。

从Redis自身角度来说，因为读写网络的read/write系统调用占用了Redis执行期间大部分CPU时间，瓶颈主要在于网络的 IO 消耗, 优化主要有两个方向:

• 提高网络 IO 性能，典型的实现比如使用 DPDK 来替代内核网络栈的方式

• 使用多线程充分利用多核，典型的实现比如 Memcached。

协议栈优化的这种方式跟 Redis 关系不大，支持多线程是一种最有效最便捷的操作方式。所以总结起来，redis支持多线程主要就是两个原因：

• 可以充分利用服务器 CPU 资源，目前主线程只能利用一个核

• 多线程任务可以分摊 Redis 同步 IO 读写负荷

Redis6.0默认是否开启了多线程？

Redis6.0的多线程默认是禁用的，只使用主线程。如需开启需要修改redis.conf配置文件：io-threads-do-reads yes

# 默认值6.0版本
io-threads-do-reads no

Redis6.0多线程开启时，线程数如何设置？

开启多线程后，还需要设置线程数，否则是不生效的。同样修改redis.conf配置文件

io-threads 4

关于线程数的设置，官方有一个建议：4核的机器建议设置为2或3个线程，8核的建议设置为6个线程，线程数一定要小于机器核数。还需要注意的是，线程数并不是越大越好，官方认为超过了8个基本就没什么意义了。

Redis6.0采用多线程后，性能的提升效果如何？

Redis 作者 antirez 在 RedisConf 2019分享时曾提到：Redis 6 引入的多线程 IO 特性对性能提升至少是一倍以上。国内也有大牛曾使用unstable版本在阿里云esc进行过测试，GET/SET 命令在4线程 IO时性能相比单线程是几乎是翻倍了。

详见：https://zhuanlan.zhihu.com/p/76788470

说明1：这些性能验证的测试并没有针对严谨的延时控制和不同并发的场景进行压测。数据仅供验证参考而不能作为线上指标。

说明2：如果开启多线程，至少要4核的机器，且Redis实例已经占用相当大的CPU耗时的时候才建议采用，否则使用多线程没有意义。所以估计80%的公司开发人员看看就好。

Redis6.0多线程的实现机制？

流程简述如下：

1. 主线程负责接收建立连接请求，获取 socket 放入全局等待读处理队列
2. 主线程处理完读事件之后，通过 RR(Round Robin) 将这些连接分配给这些 IO 线程
3. 主线程阻塞等待 IO 线程读取 socket 完毕
4. 主线程通过单线程的方式执行请求命令，请求数据读取并解析完成，但并不执行
5. 主线程阻塞等待 IO 线程将数据回写 socket 完毕
6. 解除绑定，清空等待队列

（图片来源：https://ruby-china.org/topics/38957）

该设计有如下特点： 1、IO 线程要么同时在读 socket，要么同时在写，不会同时读或写 2、IO 线程只负责读写 socket 解析命令，不负责命令处理

开启多线程后，是否会存在线程并发安全问题？

从上面的实现机制可以看出，Redis的多线程部分只是用来处理网络数据的读写和协议解析，执行命令仍然是单线程顺序执行。所以我们不需要去考虑控制 key、lua、事务，LPUSH/LPOP 等等的并发及线程安全问题。

Linux环境上如何安装Redis6.0.1（6.0的正式版是6.0.1）？

这个和安装其他版本的redis没有任何区别，整个流程跑下来也没有任何的坑，所以这里就不做描述了。唯一要注意的就是配置多线程数一定要小于cpu的核心数，查看核心数量命令：

[root@centos7.5 ~]# lscpu
Architecture: x86_64
CPU op-mode(s): 32-bit, 64-bit
Byte Order: Little Endian
CPU(s): 4
On-line CPU(s) list: 0-3

Redis6.0的多线程和Memcached多线程模型进行对比

如上图所示：Memcached 服务器采用 master-woker 模式进行工作，服务端采用 socket 与客户端通讯。主线程、工作线程 采用 pipe管道进行通讯。主线程采用 libevent 监听 listen、accept 的读事件，事件响应后将连接信息的数据结构封装起来，根据算法选择合适的工作线程，将连接任务携带连接信息分发出去，相应的线程利用连接描述符建立与客户端的socket连接 并进行后续的存取数据操作。

Redis6.0与Memcached多线程模型对比：

相同点：都采用了 master线程-worker 线程的模型

不同点：Memcached 执行主逻辑也是在 worker 线程里，模型更加简单，实现了真正的线程隔离，符合我们对线程隔离的常规理解。而 Redis 把处理逻辑交还给 master 线程，虽然一定程度上增加了模型复杂度，但也解决了线程并发安全等问题。

Redis作者是如何点评 “多线程”这个新特性的？

关于多线程这个特性，在6.0 RC1时，Antirez曾做过说明：

Redis支持多线程有2种可行的方式：第一种就是像“memcached”那样，一个Redis实例开启多个线程，从而提升GET/SET等简单命令中每秒可以执行的操作。这涉及到I/O、命令解析等多线程处理，因此，我们将其称之为“I/O threading”。另一种就是允许在不同的线程中执行较耗时较慢的命令，以确保其它客户端不被阻塞，我们将这种线程模型称为“Slow commands threading”。

经过深思熟虑，Redis不会采用“I/O threading”，redis在运行时主要受制于网络和内存，所以提升redis性能主要是通过在多个redis实例，特别是redis集群。接下来我们主要会考虑改进两个方面： 1.Redis集群的多个实例通过编排能够合理地使用本地实例的磁盘，避免同时重写AOF。 2.提供一个Redis集群代理，便于用户在没有较好的集群协议客户端时抽象出一个集群。

补充说明一下，Redis和memcached一样是一个内存系统，但不同于Memcached。多线程是复杂的，必须考虑使用简单的数据模型，执行LPUSH的线程需要服务其他执行LPOP的线程。

我真正期望的实际是“slow operations threading”，在redis6或redis7中，将提供“key-level locking”，使得线程可以完全获得对键的控制以处理缓慢的操作。

详见：http://antirez.com/news/126

Redis线程中经常提到IO多路复用，如何理解？

这是IO模型的一种，即经典的Reactor设计模式，有时也称为异步阻塞IO。

多路指的是多个socket连接，复用指的是复用一个线程。多路复用主要有三种技术：select，poll，epoll。epoll是最新的也是目前最好的多路复用技术。采用多路 I/O 复用技术可以让单个线程高效的处理多个连接请求（尽量减少网络IO的时间消耗），且Redis在内存中操作数据的速度非常快（内存内的操作不会成为这里的性能瓶颈），主要以上两点造就了Redis具有很高的吞吐量。

你知道Redis的彩蛋LOLWUT吗？

这个其实从Redis5.0就开始有了，但是原谅我刚刚知道。作者是这么描述这个功能的《LOLWUT: a piece of art inside a database command》，“数据库命令中的一件艺术品”。你可以把它称之为情怀，也可以称之为彩蛋，具体是什么，我就不透露了。和我一样不清楚是什么的小伙伴可以参见：http://antirez.com/news/123，每次运行都会随机生成的噢。

参考、致谢

Rdis作者Antirez的博客：http://antirez.com

1. https://www.zhihu.com/question/26943938/answer/68773398
2. https://zhuanlan.zhihu.com/p/76788470
3. http://www.web-lovers.com/redis-source-6-rc-mult-thread.html
4. https://ruby-china.org/topics/38957
5. https://redis.io/topics/faq#redis-is-single-threaded-how-can-i-exploit-multiple-cpu--cores
6. https://juejin.im/post/5e9ae485f265da47b04d95d2
7. https://www.cnblogs.com/gattaca/p/6929361.html

select、poll、epoll之间的区别总结

多路复用的三种方式，总结的很好，自己需要看看并理解。

https://www.cnblogs.com/Anker/p/3265058.html

https://www.cnblogs.com/Anker/archive/2013/08/14/3258674.html

https://www.cnblogs.com/Anker/archive/2013/08/15/3261006.html

https://www.cnblogs.com/Anker/archive/2013/08/17/3263780.html

Redis Pipelining

请求/响应协议和RTT

pipeline选择客户端缓冲

1. 使用管道技术可以显著提升Redis处理命令的速度，其原理就是将多条命令打包，只需要一次网络开销，在服务器端和客户端各一次read()和write()系统调用，以此来节约时间。
2. 管道中的命令数量要适当，并不是越多越好。
3. Redis2.6版本以后，脚本在大部分场景中的表现要优于管道。

TODOReactor 模式

在处理web请求时，通常有两种体系结构，分别为：thread-based architecture（基于线程）、event-driven architecture（事件驱动）

事件驱动在很多地方用到，

https://zhuanlan.zhihu.com/p/93612337

https://www.jianshu.com/p/eef7ebe28673

TODORedis的事件模型（ae epoll实现方式）

https://cloud.tencent.com/developer/article/1477190

五种基础数据类型

string（字符串），hash（哈希），list（列表），set（集合）及zset(sorted set：有序集合)。

类型	简介	特性	场景
String(字符串)	二进制安全	可以包含任何数据,比如jpg图片或者序列化的对象,一个键最大能存储512M	---
Hash(字典)	键值对集合,即编程语言中的Map类型	适合存储对象,并且可以像数据库中update一个属性一样只修改某一项属性值(Memcached中需要取出整个字符串反序列化成对象修改完再序列化存回去)	存储、读取、修改用户属性
List(列表)	链表(双向链表)	增删快,提供了操作某一段元素的API	1,最新消息排行等功能(比如朋友圈的时间线) 2,消息队列
Set(集合)	哈希表实现,元素不重复	1、添加、删除,查找的复杂度都是O(1) 2、为集合提供了求交集、并集、差集等操作	1、共同好友 2、利用唯一性,统计访问网站的所有独立ip 3、好友推荐时,根据tag求交集,大于某个阈值就可以推荐
Sorted Set(有序集合)	将Set中的元素增加一个权重参数score,元素按score有序排列	数据插入集合时,已经进行天然排序	1、排行榜 2、带权重的消息队列

String

string 类型是二进制安全的。意思是 redis 的 string 可以包含任何数据。比如jpg图片或者序列化的对象。

string 类型的值最大能存储 512MB。

# 键值对
set name "hello"
get name
exists name
del name
# 批量键值对
mset name1 boy name2 girl name3 unknown
mget name1 name2 name3
# 5s 后过期，等价于 set+expire
set name "codehole"
expire name 5
setex name 5 codehole
# 计数,整数，最大。十进制、64位、有符号、整数-9223372036854775808~9223372036854775807。
# 因为 Redis 没有专用的整数类型，所以 key 内储存的字符串被解释为十进制 64 位有符号整数来执行 INCR 操作。
set age 30
incr age
decr age
incrby age 5
incrby age -5
# 锁
setnx name "hello"

字符串结构使用非常广泛，一个常见的用途就是缓存用户信息。我们将用户信息结构体使用 JSON 序列化成字符串，然后将序列化后的字符串塞进 Redis 来缓存。同样，取用户信息会经过一次反序列化的过程。

SDS:简单动态字符串

sds (Simple Dynamic String)，Simple的意思是简单，Dynamic即动态，意味着其具有动态增加空间的能力，扩容不需要使用者关心。

• 二进制安全的数据结构，不会产生数据的丢失
• 内存预分配机制，避免了频繁的内存分配。当字符串长度小于 1M 时，扩容都是加倍现有的空间，如果超过 1M，扩容时一次只会多扩 1M 的空间。(字符串最大长度为 512M)
• 兼容c语言函数库，可以使用部分c的字符串函数

底层结构

sds 有两个版本，在Redis 3.2之前使用的是第一个版本，其数据结构如下所示：

typedef char *sds;      //注意，sds其实不是一个结构体类型，而是被typedef的char*

struct sdshdr {
    unsigned int len;   //buf中已经使用的长度
    unsigned int free;  //buf中未使用的长度
    char buf[];         //柔性数组buf
};

但是在Redis 3.2 版本中，对数据结构做出了修改，针对不同的长度范围定义了不同的结构，如下，这是目前的结构：

typedef char *sds;      

struct __attribute__ ((__packed__)) sdshdr5 {     // 对应的字符串长度小于 1<<5
    unsigned char flags; 
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr8 {     // 对应的字符串长度小于 1<<8
    uint8_t len; /* used */                       // 目前字符创的长度,使用1个byte
    uint8_t alloc;                                // 已经分配的总长度,使用1个byte
    unsigned char flags;                          // flag用3bit来标明类型，类型后续解释，其余5bit目前没有使用。使用1byte。
    char buf[];                                   // 柔性数组，以'\0'结尾
};
struct __attribute__ ((__packed__)) sdshdr16 {    // 对应的字符串长度小于 1<<16
    uint16_t len; /* used，使用2byte */
    uint16_t alloc; /* excluding the header and null terminator,使用2byte */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr32 {    // 对应的字符串长度小于 1<<32
    uint32_t len; /* used,使用4byte */
    uint32_t alloc; /* excluding the header and null terminator,使用4byte */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};
struct __attribute__ ((__packed__)) sdshdr64 {    // 对应的字符串长度小于 1<<64
    uint64_t len; /* used */
    uint64_t alloc; /* excluding the header and null terminator */
    unsigned char flags; /* 3 lsb of type, 5 unused bits */
    char buf[];
};

根据字符串的长度，分成了5种类型sdshdr5、sdshdr8、sdshdr16、sdshdr32、sdshdr64。这里可以很明显的看出sdshdr5居然没有了头部（len和free），而其他四种数据结构，多了一个flags字段。

如上图所示，sdshdr5结构中，flags占1个字符，其低3位表示type，高5位表示长度，能表示的长度区间为0~31，flags后面就是字符串的内容。 而长度大于31的字符串，1个字节存不下，那么就要将len和free单独存放。sdshdr8、sdshdr16、sdshdr32和sdshdr64的结构相同，如下图所示：

这4种结构的成员变量类似，唯一的区别是len和alloc的类型不同。结构体中4个字段的具体含义如下：

1. len表示buf中已占用字节数。
2. alloc表示buf中已分配字节数，记录的是为buf分配的总长度，不同于free。
3. flags标识当前结构体的类型，低3位用作标识位，高5位预留。
4. buf柔性数组，真正存储字符串的数据空间。

创建SDS的大致流程是，首先计算好不同类型的头部和初始长度，然后动态分配内存。不过，需要注意一下3点：

1. 创建空字符串是SDS_TYPE_5会被强制转换为SDS_TYPE_8。
2. 长度计算时有“+1”操作，是为了算上结束符“0”。
3. 返回值是指向sds结构buf字段的指针。

什么是二进制安全？

通俗的讲，C语言中，用“0”表示字符串的结束，如果字符串中本身就有“0”字符，那么这个字符串就会被截断，即非二进制安全；若通过某种机制，保证读写字符串时不损害其内容，则是二进制安全。 Redis 3.2 之前的SDS主要是通过int len; int free; char buf[];这三个字段来确定一个字符串的。其中len表示buf中已占用字节数，free表示buf中剩余可用字节数，buf是数据空间。

这样设计有什么优点？

有单独的统计变量len和free（称为头部），可以很方便的得到字符串的长度。 2、内容存放在柔性数组buf中，SDS对上层暴露的指针不是指向结构体SDS的指针，而是直接指向柔性数组buf的指针。上层可以像读取C字符串一样读取SDS的内容，兼容C语言处理字符串的各种函数。 3、由于有长度的统计变量len的存在，读写字符串时不依赖“0”终止符，保证了二进制安全。

为什么要用柔性数组？

柔性数组的地址和结构体是连续的，这样查找内存更快（因为不需要额外通过指针找到字符串的位置）；可以更快的通过柔性数组的首地址偏移得到结构体首地址，进而能很方便的获取其余变量。

思考：这样的设计有个缺点，不同长度的字符串需要占用相同大小的头部，显然是浪费了空间。

SDS如何兼容C语言字符串？如何保证二进制安全？

SDS对象中的buf是一个柔性数组，上层调用时，SDS直接返回了buf。由于buf是直接指向内容的指针，所以兼容C语言函数。而当真正读取内容时，SDS会通过len来限制读取长度，而非“0”，所以保证了二进制安全。

sdshdr5的特殊之处是什么？

sdshdr5只负责存储小于32字节的字符串。一般情况下，小字符串的存储更普遍，所以Redis进一步压缩了sdshdr5的数据结构，将sdshdr5的类型和长度放入了同一个属性中，用flags的低3位存储类型，高5位存储长度。创建空字符串时，sdshdr5会被sdshdr8替代。

SDS是如何扩容的？

SDS在涉及字符串修改时会调用sdsMakeroomFor函数进行检查，会根据空闲长度和新增内容的长度进行比较判断，然后根据不同情况动态扩容，该操作对上层透明。

Hash

Redis 的字典相当于 Java 语言里面的 HashMap，它是无序字典。内部实现结构上同Java 的 HashMap 也是一致的，同样的数组 + 链表二维结构。 hash 碰撞，用链表串接起来。

hash 特别适合用于存储对象。

每个 hash 可以存储 2^32 - 1 键值对（40多亿），sh 特别适合用于存储对象

rehash 采用的是渐进式hash。

# 命令行的字符串如果包含空格，要用引号括起来
hset books java "think in java"
hget books java
hlen books
hmset books java "effective java" python "learning python" golang "modern golang programming"
hmget books java python golang
# entries()， key 和 value 间隔出现
hgetall books

# 计数
hincrby xiaoming age 5

数据少的时候是压缩表，多了才是标准结构。

# 查看存储结构 encoding
> debug object programmings
Value at:0x7fec2de00020 refcount:1 encoding:ziplist serializedlength:36 lru:6022374 lru_seconds_idle:6

压缩表

字典

List

Redis 的列表相当于 Java 语言里面的 LinkedList，注意它是链表而不是数组。这意味着list 的插入和删除操作非常快，时间复杂度为 O(1)，但是索引定位很慢，时间复杂度为O(n)。

当列表弹出了最后一个元素之后，该数据结构自动被删除，内存被回收。

Redis 的列表结构常用来做异步队列使用。 将需要延后处理的任务结构体序列化成字符串塞进 Redis 的列表，另一个线程从这个列表中轮询数据进行处理。

字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部（左边）或者尾部（右边）。

最多可以包含 2^32 - 1 个元素 (4294967295, 每个列表超过40亿个元素)。

# 右边进左边出：队列
rpush books python java golang
llen books # 查看长度，不存在 rlen 命令，哈哈
lpop books

# 右边进右边出：栈
rpush books python java golang
rpop books
lrange books 0 10

慢操作：小心慢操作，

# O(n) 慎用,因为是链表结构
lindex books 1 
# 获取所有元素， O(n) 慎用。因为要遍历所有元素
lrange books 0 -1 
# 保留区间内的值。# O(n) 慎用
ltrim books 1 -1 
# 这其实是清空了整个列表，因为区间范围长度为负
ltrim books 1 0 

TODOquicklist快速列表

127.0.0.1:6379> debug object test
Value at:0x7f20762adc60 refcount:1 encoding:quicklist serializedlength:16 lru:16765476 lru_seconds_idle:2 ql_nodes:1 ql_avg_node:1.00 ql_ziplist_max:-2 ql_compressed:0 ql_uncompressed_size:14

Set

Redis 的集合相当于 Java 语言里面的 HashSet，它内部的键值对是无序的唯一的。它的内部实现相当于一个特殊的字典，字典中所有的 value 都是一个值 NULL。

Set 是 String 类型的无序集合。集合成员是唯一的，这就意味着集合中不能出现重复的数据。

Redis 中集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是 O(1)。

最大的成员数为 2^32 - 1 (4294967295, 每个集合可存储40多亿个成员)。

# 添加一个 string 元素到 key 对应的 set 集合中，成功返回 1，如果元素已经在集合中返回 0。
sadd books python
sadd books java golang
# 查询全部，注意顺序，和插入的并不一致，因为 set 是无序的
smembers books 
# 查询某个 value 是否存在，相当于 contains(o)
sismember books java 
# 获取长度相当于 count()
scard books 
# 默认弹出一个 spop key, 可以指定弹出的个数
spop books n

ZSet

（Sorted Set），跳跃表，有序集合和集合一样也是 string 类型元素的集合,且不允许重复的成员。

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

有序集合的成员是唯一的,但分数(score)却可以重复。内部 score 使用 double 类型进行存储，所以存在小数点精度问题。

集合是通过哈希表实现的。

zadd books 9.0 "think in java"
# 按 score 排序列出，参数区间为排名范围
zrange books 0 -1 
# 按 score 逆序列出，参数区间为排名范围
zrevrange books 0 -1 
# 相当于 count()
zcard books
# 获取指定 value 的 score
zscore books "java concurrency"
# 获取排名
zrank books "java concurrency" 
# 根据分值区间遍历 zset
zrangebyscore books 0 8.91 
# 根据分值区间 (-∞, 8.91] 遍历 zset，同时返回分值。 inf 代表 infinite，无穷大的意思。
zrangebyscore books -inf 8.91 withscores 
# 删除 value
zrem books "java concurrency" 

ZSet底层结构

参考：https://www.jianshu.com/p/fb7547369655

https://blog.csdn.net/weichi7549/article/details/107335133 （这篇文章总结的真棒，让我恍然顿悟）

zset底层的存储结构包括ziplist或skiplist，在同时满足以下两个条件的时候使用ziplist，其他时候使用skiplist，两个条件如下：

• 元素数量小于128个。zset-max-ziplist-entries
• 所有元素的长度小于64字节。 zset-max-ziplist-value

当ziplist作为zset的底层存储结构时候，每个集合元素使用两个紧挨在一起的压缩列表节点来保存，第一个节点保存元素的成员，第二个元素保存元素的分值。

当跳跃表作为zset的底层存储结构的时候，使用skiplist按序保存元素及分值，使用dict来保存元素和分值的映射关系。

ziplist

ziplist 编码的 Zset 使用紧挨在一起的压缩列表节点来保存，第一个节点保存 member，第二个保存 score。ziplist 内的集合元素按 score 从小到大排序，其实质是一个双向链表。虽然元素是按 score 有序排序的， 但对 ziplist 的节点指针只能线性地移动，所以在 REDIS_ENCODING_ZIPLIST 编码的 Zset 中， 查找某个给定元素的复杂度为 O(N)。

压缩表包含 5 部分：

• zlbytes： 存储一个无符号整数，固定四个字节长度（32bit），用于存储压缩列表所占用的字节（也包括本身占用的4个字节），当重新分配内存的时候使用，不需要遍历整个列表来计算内存大小。
• zltail： 存储一个无符号整数，固定四个字节长度（32bit），表示ziplist表中最后一项（entry）在ziplist中的偏移字节数。的存在，使得我们可以很方便地找到最后一项（不用遍历整个ziplist），从而可以在ziplist尾端快速地执行push或pop操作。
• zllen： 压缩列表包含的节点个数，固定两个字节长度（16bit）， 表示ziplist中数据项（entry）的个数。由于zllen字段只有16bit，所以可以表达的最大值为2^16-1。注意点：如果ziplist中数据项个数超过了16bit能表达的最大值，ziplist仍然可以表示。ziplist是如何做到的？如果小于等于2^16-2（也就是不等于2^16-1），那么就表示ziplist中数据项的个数；否则，也就是等于16bit全为1的情况，那么就不表示数据项个数了，这时候要想知道ziplist中数据项总数，那么必须对ziplist从头到尾遍历各个数据项，才能计数出来。
• entry，表示真正存放数据的数据项，长度不定，节点的长度由节点保存的内容决定。一个数据项（entry）也有它自己的内部结构。
• zlend， ziplist最后1个字节，值固定等于255，其是一个结束标记，用于标记压缩列表的末端。

优点：

缺点：

• 因为压缩表是紧凑存储的，没有多余的空间。这就意味着插入一个新的元素就需要调用函数扩展内存。过程中可能需要重新分配新的内存空间，并将之前的内容一次性拷贝到新的地址。

• 级联更新

skiplist

skiplist 编码的 Zset 底层为一个被称为 zset 的结构体，这个结构体中包含一个字典和一个跳跃表。跳跃表按 score 从小到大保存所有集合元素，查找时间复杂度为平均 O(logN)，最坏 O(N) 。字典则保存着从 member 到 score 的映射，这样就可以用 O(1)的复杂度来查找 member 对应的 score 值。虽然同时使用两种结构，但它们会通过指针来共享相同元素的 member 和 score，因此不会浪费额外的内存。

跳跃表的基础层链表，是双向链表，为了方便逆序查找。

跳表(skip List)是一种随机化的数据结构，基于并联的链表，实现简单，插入、删除、查找的复杂度均为O(logN)。简单说来跳表也是链表的一种，只不过它在链表的基础上增加了跳跃功能，正是这个跳跃的功能，使得在查找元素时，跳表能够提供O(logN)的时间复杂度。

多层链表，从基础链表中抽出相邻节点的一个，提取的上一次，如果当前层节点数大于二，继续向上抽取。这样上面每一层链表的节点个数，是下面一层的节点个数的一半，这样查找过程就非常类似于一个二分查找，使得链表查找的时间复杂度可以降低到O(log n)。

但是插入的时候，就会打乱上下相邻两层链表上节点个数严格的2:1的对应关系。如果要维持这种对应关系，就必须把新插入的节点后面的所有节点（也包括新插入的节点）重新进行调整，这会让时间复杂度重新蜕化成O(n)。删除数据也有同样的问题。

skiplist 为了避免关键节点索引这一问题，它不要求上下相邻两层链表之间的节点个数有严格的对应关系，而是为每个节点随机出一个层数(level)。比如，一个节点随机出的层数是3，那么就把它链入到第1层到第3层这三层链表中。（这个很关键，我一直疑惑的就是这一点。原来逻辑就是，每个节点，都随机给个层级，然后放到，该层和所有下层链表中。Redis 最大层级32，概率 0.25，即 1/4。哈哈哈）

从上面skiplist的创建和插入过程可以看出，每一个节点的层数（level）是随机出来的，而且新插入一个节点不会影响其它节点的层数。因此，插入操作只需要修改插入节点前后的指针，而不需要对很多节点都进行调整。这就降低了插入操作的复杂度。实际上，这是skiplist的一个很重要的特性，这让它在插入性能上明显优于平衡树的方案。

skiplist，指的就是除了最下面第1层链表之外，它会产生若干层稀疏的链表（和严格相邻节点的多层链表对比），这些链表里面的指针故意跳过了一些节点（而且越高层的链表跳过的节点越多）。这就使得我们在查找数据的时候能够先在高层的链表中进行查找，然后逐层降低，最终降到第1层链表来精确地确定数据位置。在这个过程中，我们跳过了一些节点，从而也就加快了查找速度。

刚刚创建的这个skiplist总共包含4层链表，现在假设我们在它里面依然查找23，下图给出了查找路径：

需要注意的是，前面演示的各个节点的插入过程，实际上在插入之前也要先经历一个类似的查找过程，在确定插入位置后，再完成插入操作。

实际应用中的skiplist每个节点应该包含key和value两部分。前面的描述中我们没有具体区分key和value，但实际上列表中是按照key(score)进行排序的，查找过程也是根据key在比较。

执行插入操作时计算随机数的过程，是一个很关键的过程，它对skiplist的统计特性有着很重要的影响。这并不是一个普通的服从均匀分布的随机数，它的计算过程如下：

• 首先，每个节点肯定都有第1层指针（每个节点都在第1层链表里）。
• 如果一个节点有第i层(i>=1)指针（即节点已经在第1层到第i层链表中），那么它有第(i+1)层指针的概率为p。
• 节点最大的层数不允许超过一个最大值，记为MaxLevel。

随机层数的计算方法（源码）:

//zskiplist max level,Should be enough for 2^32 elements
#define ZSKIPLIST_MAXLEVEL 32

//Skiplist P = 1/4
#define ZSKIPLIST_P 0.25

int zslRandomLevel(void)
{
    int level = 1;
    while ((rand()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

skiplist的数据结构定义：

#define ZSKIPLIST_MAXLEVEL 32
#define ZSKIPLIST_P 0.25

typedef struct zskiplistNode {
    robj *obj;
    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;

跳表rank原理：跳表计算rank实际是经历了一次对目标值的查找过程，并在这个过程中累加出来的。在跳表中，会为每个节点在每一level维护下一跳的距离span值。们在不同level向右移动的过程中就只需要累加span，最总 span 的累加值就是排名。

跳跃表 原理剖析

参考：https://blog.csdn.net/qq_24047659/article/details/88042998

什么是跳跃表？跳跃表(Skip List)是一种基于【有序链表】的扩展，简称【跳表】。其实就是使用【关键节点】作为【索引】的一种结构。

层级极限是什么？当节点足够多的时候，不止能提出2层索引，还可以向更高层次提取，保证每一层是上一层节点数的【一半】至于提取的【极限】，当提出的新一层只有【两个节点】的时候（因为继续提出新的一层，只会有一个节点没有比较的意义）。这样的【多层链表】结构，就是所谓的【跳跃表】

跳跃表【删除节点】的流程？自上而下，查找第一次出现节点的索引，并逐层找到每一层对应的节点O(logN)。删除每一层查找到的节点，如果该层只剩下1个节点，删除整个一层（原链表除外）。O(logN)

为什么要从最高层链表开始呢？因为高层链表串联的节点之间稀疏，跨度大，所以可以快速推进；一旦发现高层链表没有线索了，则需要下降高度到更稠密的链表索引中，继续向目标推进；直到某一个高度的链表索引中找到了目标；或者到最低层链表也没有找到目标，则说明目标值不存在。

skiplist与平衡树、哈希表的比较：

• 有序性：skiplist和各种平衡树（如AVL、红黑树等）的元素是有序排列的，而哈希表不是有序的。因此，在哈希表上只能做单个key的查找，不适宜做范围查找。所谓范围查找，指的是查找那些大小在指定的两个值之间的所有节点。

• 范围查找：平衡树比skiplist操作要复杂。在平衡树上，我们找到指定范围的小值之后，还需要以中序遍历的顺序继续寻找其它不超过大值的节点。如果不对平衡树进行一定的改造，这里的中序遍历并不容易实现。而在skiplist上进行范围查找就非常简单，只需要在找到小值之后，对第1层链表进行若干步的遍历就可以实现。类似mysql中B+树，叶子节点，链表连接，方便范围查找。

• 插入、删除：平衡树的插入和删除操作可能引发子树的调整（rebalance重新调整结构平衡），逻辑复杂，而skiplist的插入和删除只需要修改相邻节点的指针，操作简单又快速（插入时，需要先随机出该节点层级）。

• 内存占用：skiplist比平衡树更灵活一些。一般来说，平衡树每个节点包含2个指针（分别指向左右子树），而skiplist每个节点包含的指针数目平均为1/(1-p)，具体取决于参数p的大小。如果像Redis里的实现一样，取p=1/4，那么平均每个节点包含1.33个指针，比平衡树更有优势。

• 查找单个key：skiplist和平衡树的时间复杂度都为O(log n)，大体相当；而哈希表在保持较低的哈希值冲突概率的前提下，查找时间复杂度接近O(1)，性能更高一些。所以我们平常使用的各种Map或dictionary结构，大都是基于哈希表实现的。

• 从算法实现难度上来比较，skiplist比平衡树要简单得多。

跳跃表的实际运用？Redis当中的Sorted-set这种【有序集合】，正是对跳跃表的改进和应用。对于关系型数据库如何维护有序的记录集合呢？ 使用的是B+树，特点是聚簇索引，n叉树（减少IO），非叶子节点只存主键(减少内存占用，一个页可以存更多的键，减少IO)，叶子节点存储主键和行完整数据，叶子节点链表相连（方便范围查找）。

Redis为什么用skiplist而不用平衡树？

• 也不是非常耗费内存，实际上取决于生成层数函数里的概率 p，取决得当的话其实和平衡树差不多。
• 因为有序集合经常会进行 ZRANGE 或 ZREVRANGE 这样的范围查找操作，跳表里面的双向链表可以十分方便地进行这类操作。
• 实现简单，ZRANK 操作还能达到 o(logn)的时间复杂度O

需要注意的点

容器型数据结构的通用规则 ：

list/set/hash/zset 这四种数据结构是容器型数据结构，它们共享下面两条通用规则：

1. create if not exists 如果容器不存在，那就创建一个，再进行操作。比如 rpush 操作刚开始是没有列表的，Redis 就会自动创建一个，然后再 rpush 进去新元素
2. drop if no elements 如果容器里元素没有了，那么立即删除元素，释放内存。这意味着 lpop 操作到最后一个元素，列表就消失了。

过期时间 ：

• Redis 所有的数据结构都可以设置过期时间，时间到了， Redis 会自动删除相应的对象。 需要注意的是过期是以对象为单位，比如一个 hash 结构的过期是整个 hash 对象的过期， 而不是其中的某个子 key。
• 还有一个需要特别注意的地方是如果一个字符串已经设置了过期时间，然后你调用了 set 方法修改了它，它的过期时间会消失。

三种特殊的数据类型

geospatial

Geospatial 地理位置

应用场景：

• 附近的人：可以根据定位，查找附近的人
• 两点之间，距离计算：但是计算出来的位置只算水平距离，不算垂直距离

Hyperloglog

pf 它是 HyperLogLog 这个数据结构的发明人 Philippe Flajolet 的首字母缩写。

pfmerge 适合什么场合用？ 比如在网站中我们有两个内容差不多的页面，运营说需要这两个页面的数据进行合并。其中页面的 UV 访问量也需要合并，那这个时候 pfmerge 就可以派上用场了。

pf 的内存占用为什么是 12k？ 在 Redis 的 HyperLogLog实现中用到的是 16384 个桶，也就是 2^14，每个桶的 maxbits 需要 6 个 bits 来存储，最大可以表示 maxbits=63，于是总共占用内存就是 2^14 * 6 / 8 = 12k 字节。

注意事项：它需要占据一定 12k 的存储空间，所以它不适合统计单个用户相关的数据。 不过你也不必过于当心，因为 Redis 对 HyperLogLog 的存储进行了优化，在计数比较小时，它的存储空间采用稀疏矩阵存储，空间占用很小，仅仅在计数慢慢变大，稀疏矩阵占用空间渐渐超过了阈值时才会一次性转变成稠密矩阵，才会占用 12k 的空间。

说明：

• 基数不大，数据量不大就用不上，会有点大材小用浪费空间
• 有局限性，就是只能统计基数数量，而没办法去知道具体的内容是什么
• 和bitmap相比，属于两种特定统计情况，简单来说，HyperLogLog 去重比 bitmap 方便很多
• 一般可以bitmap和hyperloglog配合使用，bitmap标识哪些用户活跃，hyperloglog计数

使用场景，一般使用：

• 统计注册 IP 数
• 统计每日访问 IP 数
• 统计页面实时 UV 数
• 统计在线用户数
• 统计用户每天搜索不同词条的个数

每个网页每天的 UV 数据

一个爆款页面几千万的 UV ，上百个页面。使用set存储用户ID再 scard，太浪费内存，肯定是大key。

Redis 提供了 HyperLogLog 数据结构就是用来解决这种统计问题的。 HyperLogLog 提供不精确的去重计数方案，虽然不精确但是也不是非常不精确，标准误差是 0.81%，这样的精确度已经可以满足上面的 UV 统计需求了。

pfadd codehole user1
pfcount codehole
pfadd codehole user2 user3 user4

实现原理

HyperLogLog 的使用非常简单，但是实现原理比较复杂。

使用概率论，稀疏矩阵，稠密矩阵。伯努利原理。

https://zhuanlan.zhihu.com/p/26562588

Bitmap

字符串是由多个字节组成，每个字节又是由 8 个 bit 组成，如此便可以将一个字符串看成很多 bit 的组合，这便是 bitmap「位图」数据结构 。

位图，setbit等命令只不过是在set上的扩展。在bitmap上可执行AND,OR,XOR以及其它位操作。

# 统计每日登录用户，或者签到用户 
setbit daily_active_users:2019-03-27 10 1
getbit daily_active_users:2019-03-27 9
bitcount daily_active_users:2019-03-27

#bitpos 遗憾的是， start 和 end 参数是字节索引，也就是说指定的位范围必须是 8 的倍数，而不能任意指定。
set w hello
bitcount w
bitcount w 0 0 # 第一个字符中 1 的位数
bitcount w 0 1 # 第一个字符中 1 的位数

位图不是特殊的数据结构，它的内容其实就是普通的字符串，也就是 byte 数组。我们可以使用普通的 get/set 直接获取和设置整个位图的内容，也可以使用位图操作 getbit/setbit等将 byte 数组看成「位数组」来处理。

# 零存整取，根据每个字符的 ASCII 码的二进制值
# 例如：字母A。01000001   65  41  A
# 零存整取
setbit s 1 1
setbit s 7 1
get s
"A"
# 整存零取
set s A
getbit s 1
getbit s 7
# 如果对应位的字节是不可打印字符， redis-cli 会显示该字符的 16 进制形式
setbit x 0 1
setbit x 1 1
get s
"\xc0"

「零存整取」，同样我们还也可以「零存零取」，「整存零取」。「 零存」就是使用 setbit 对位值进行逐个设置，「整存」就是使用字符串一次性填充所有位数组，覆盖掉旧值。

• 值得注意的是位数组的顺序和字符的位顺序是相反的。
• 如果对应位的字节是不可打印字符， redis-cli 会显示该字符的 16 进制形式
• 零存整取，根据每个字符的 ASCII 码的二进制值

魔术指令 bitfield

前文我们设置 (setbit) 和获取 (getbit) 指定位的值都是单个位的，如果要一次操作多个位，就必须使用管道来处理。 不过 Redis 的 3.2 版本以后新增了一个功能强大的指令，有了这条指令，不用管道也可以一次进行多个位的操作。 bitfield 有三个子指令，分别是get/set/incrby，它们都可以对指定位片段进行读写，但是最多只能处理 64 个连续的位，如果超过 64 位，就得使用多个子指令， bitfield 可以一次执行多个子指令。

set w hello
bitfield w get u4 0 # 从第一个位开始取 4 个位，结果是无符号数 (u)
bitfield w get u3 2 # 从第三个位开始取 3 个位，结果是无符号数 (u)
bitfield w get i4 0 # 从第一个位开始取 4 个位，结果是有符号数 (i)
bitfield w get i3 2 # 从第三个位开始取 3 个位，结果是有符号数 (i)

所谓有符号数是指获取的位数组中第一个位是符号位，剩下的才是值。如果第一位是1，那就是负数。无符号数表示非负数，没有符号位，获取的位数组全部都是值。有符号数最多可以获取 64 位，无符号数只能获取 63 位 (因为 Redis 协议中的 integer 是有符号数，最大 64 位，不能传递 64 位无符号值)。如果超出位数限制， Redis 就会告诉你参数错误。

biffield一次执行多个子指令:

127.0.0.1:6379> bitfield w get u4 0 get u3 2 get i4 0 get i3 2
1) (integer) 6
2) (integer) 5
3) (integer) 6
4) (integer) -3

我们使用 set 子指令将第二个字符 e 改成 a， a 的 ASCII 码是 97。

127.0.0.1:6379> bitfield w set u8 8 97 # 从第 8 个位开始，将接下来的 8 个位用无符号数 97 替换
1) (integer) 101
127.0.0.1:6379> get w
"hallo"

再看第三个子指令 incrby，它用来对指定范围的位进行自增操作。既然提到自增，就有可能出现溢出。如果增加了正数，会出现上溢，如果增加的是负数，就会出现下溢出。 Redis默认的处理是折返。如果出现了溢出，就将溢出的符号位丢掉。如果是 8 位无符号数 255，加 1 后就会溢出，会全部变零。如果是 8 位有符号数 127，加 1 后就会溢出变成 -128。

bitfield 指令提供了溢出策略子指令 overflow，用户可以选择溢出行为，默认是折返(wrap)，还可以选择失败 (fail) 报错不执行，以及饱和截断 (sat)，超过了范围就停留在最大最小值。 overflow 指令只影响接下来的第一条指令，这条指令执行完后溢出策略会变成默认值折返 (wrap)。

# 饱和截断 SAT，保持最大值
127.0.0.1:6379> bitfield w overflow sat incrby u4 2 1 # 保持最大值
1) (integer) 15
# 失败不执行 FAIL
127.0.0.1:6379> bitfield w overflow fail incrby u4 2 1 # 不执行
1) (nil)

应用场景：

1、位图计数统计

位图计数统计的是bitmap中值为1的位的个数。位图计数的效率很高，例如，一个bitmap包含10亿个位，90%的位都置为1，在一台MacBook Pro上对其做位图计数需要21.1ms。

例子：日活跃用户

为了统计今日登录的用户数，我们建立了一个bitmap,每一位标识一个用户ID。当某个用户访问我们的网页或执行了某个操作，就在bitmap中把标识此用户的位置为1。

每次用户登录时会执行一次redis.setbit(daily_active_users, user_id, 1)。将bitmap中对应位置的位置为1，时间复杂度是O(1)。统计bitmap结果显示有今天有9个用户登录。Bitmap的key是daily_active_users，它的值是1011110100100101。

因为日活跃用户每天都变化，所以需要每天创建一个新的bitmap。我们简单地把日期添加到key后面，实现了这个功能。例如要统计某一天有多少个用户访问，可以把这个bitmap的key设计为daily_active_users:2019-03-27。当用户访问进来，我们只是简单地在bitmap中把标识这个用户的位置为1，时间复杂度是O(1)。

Redis布隆过滤器

https://oss.redislabs.com/redisbloom/Quick_Start/

可以知道一个元素一定不存在或者可能存在，占用空间更少，缺点是可能会误判。但是只要参数设置的合理，它的精确度也可以控制的相对精确，只会有小小的误判概率。

安装布隆过滤器插件：

# 下载，下载发布的版本(Tags中选一个)，master不行。不然编译报错
https://github.com/RedisBloom/RedisBloom/tree/v2.2.5
# 上到服务器 、解压 unzip、 编译 make、加载模块 
unzip RedisBloom-2.2.5.zip
cd RedisBloom-2.2.5
make
loadmodule /root/RedisBloom-2.2.5/redisbloom.so # 修改 redis.conf 配置文件，追加
# 启动需要指定配置文件，不然使用的是默认配置，修改的配置文件内容不生效
redis-server /root/redis-6.2.5/redis.conf

# 布隆过滤器的使用
# 添加，第一次添加元素，会自动创建key。默认的error_rate是 0.01，capacity是 100。
bf.add martin-bloom user1 
bf.exists martin-bloom user1 # 判断是否存在
bf.madd martin-bloom user4 user5 user6 # 批量添加
bf.mexists martin-bloom user4 user5 user6 user7 # 批量判断

# 自己创建 布隆滤器 bf.reserve
bf.reserve one-more-filter 0.0001 1000000
key：# 键
error_rate：# 期望错误率，期望错误率越低，需要的空间就越大。
capacity：# 初始容量，当实际元素的数量超过这个初始化容量时，误判率上升。但是不会急剧上升，重建布隆过滤器。

Redis提供了自定义参数的布隆过滤器，需要我们在add之前使用bf.reserve指令显式创建。bf.reserve有三个参数，分别是key，error_rate和initial_size。initial_size参数表示预计放入的元素数量，当实际数量超出这个数值时，误判率就会上升。所以需要提前设置一个较大的数值避免超出导致误判率升高，如果不使用bf.reserve，默认的error_rate是0.01，默认的initial_size为100。

https://krisives.github.io/bloom-calculator/（元素数量，错误率，计算所需要的指纹数量，空间大小）

使用注意事项：

布隆过滤器的initial_size估计的过大，会浪费存储空间，估计的过小，就会影响准确率，用户在使用之前一定要尽可能地精确估计好元素数量，还需要加上一定的冗余空间以避免实际元素可能会意外高出估计值很多。

实际中，布隆过滤器的error_rate设置的越小，需要的存储空间就会越大，对于不需要过于精确的场合，error_rate设置稍大一点也无伤大雅。比如在新闻去重上，误判率高一点只会让小部分文章不能让合适的人看到，文章的整体阅读量不会因为这点误判率就带来巨大的改变。

如果误差比较大，重建

使用bf.reserve指令显式创建。bf.reserve有三个参数，分别是key，error_rate和initial_size。

使用时不要让实际元素远大于初始化大小，当实际元素开始超出初始化大小时，应该对布隆过滤器进行重建，重新分配一个size更大的过滤器，再将所有的历史元素批量add进去（这就要求我们在其他的存储器中记录所有的历史元素）。因为error_rate不会因为数量超出就急剧增加，这就会给我们重建过滤器提供了较为宽松的时间。

原理

每个布隆过滤器对应到Redis的数据结构里面就是一个大型的位数组和几个不一样的无偏hash函数。所谓无偏就是能够把元素的hash值算的比较均匀。

向布隆过滤器中添加key时，会使用多个hash函数对key进行hash计算，计算出一个整数索引值然后对位数组长度进行取模运算得到一个位置，每个hash函数都会算得一个不同的位置。再把位数组的这几个位置都置为1就完成了add操作。

向布隆过滤器询问key是否存在时，跟add一样，也会把hash的几个位置都算出来，看看位数组中的这几个位置是否都为1，只要有一个位置是0，那么说明布隆过滤器中这个key不存在。如果都为1，这并不能说明这个key就一定存在，只是极有可能存在，因为这些位置都置为1可能是因为其他的key存在导致。如果这个位数组比较拥挤，这个概率就会很大，如果这个位数组比较稀疏，这个概率就会降低。

应用场景

主要就是需要去重的场景。

• 判断网站是否已经爬过：比如需要爬虫的网站有很多，碰到一个新的网站url，判断是否爬过
• 解决缓存穿透的问题：一般情况下，先查询缓存是否有该条数据，缓存中没有时，再查询数据库。当数据库也不存在该条数据时，每次查询都要访问数据库，这就是缓存穿透。缓存穿透带来的问题是，当有大量请求查询数据库不存在的数据时，就会给数据库带来压力，甚至会拖垮数据库。可以使用布隆过滤器解决缓存穿透的问题，把已存在数据的key存在布隆过滤器中。当有新的请求时，先到布隆过滤器中查询是否存在，如果不存在该条数据直接返回；如果存在该条数据再查询缓存查询数据库。
• 黑名单校验:发现存在黑名单中的，就执行特定操作。比如：识别垃圾邮件，只要是邮箱在黑名单中的邮件，就识别为垃圾邮件。假设黑名单的数量是数以亿计的，存放起来就是非常耗费存储空间的，布隆过滤器则是一个较好的解决方案。把所有黑名单都放在布隆过滤器中，再收到邮件时，判断邮件地址是否在布隆过滤器中即可。平时也会遇到某个正常的邮件被放到了垃圾邮件目录中，这个就是误判所致，概率极低。
• 邮箱系统的垃圾邮件过滤功能也普遍用到了布隆过滤器

渐进式 rehash

redis 的 rehash 采用的是渐进式 rehash。

背景：

redis字典（hash表）当数据越来越多的时候，就会发生扩容，也就是rehash， 扩展或收缩哈希表需要将 ht[0] 里面的所有键值对 rehash 到 ht[1] 里面， 但是， 这个 rehash 动作并不是一次性、集中式地完成的， 而是分多次、渐进式地完成的。

这样做的原因在于， 如果 ht[0] 里只保存着四个键值对， 那么服务器可以在瞬间就将这些键值对全部 rehash 到 ht[1] ； 但是， 如果哈希表里保存的键值对数量不是四个， 而是四百万、四千万甚至四亿个键值对， 那么要一次性将这些键值对全部 rehash 到 ht[1] 的话， 庞大的计算量可能会导致服务器在一段时间内停止服务。

因此， 为了避免 rehash 对服务器性能造成影响， 服务器不是一次性将 ht[0] 里面的所有键值对全部 rehash 到 ht[1] ， 而是分多次、渐进式地将 ht[0] 里面的键值对慢慢地 rehash 到 ht[1] 。

对比：java中的hashmap，当数据数量达到阈值的时候(0.75)，就会发生rehash，hash表长度变为原来的二倍，将原hash表数据全部重新计算hash地址，重新分配位置，达到rehash目的

详细步骤：

1. 为 ht[1] 分配空间， 让字典同时持有 ht[0] 和 ht[1] 两个哈希表。
2. 在字典中维持一个索引计数器变量 rehashidx ， 并将它的值设置为 0 ， 表示 rehash 工作正式开始。
3. 在 rehash 进行期间， 每次对字典执行添加、删除、查找或者更新操作时， 程序除了执行指定的操作以外， 还会顺带将 ht[0] 哈希表在 rehashidx 索引上的所有键值对 rehash 到 ht[1] ， 当 rehash 工作完成一行之后， 程序将 rehashidx 属性的值增一。
4. 随着字典操作的不断执行， 最终在某个时间点上， ht[0] 的所有键值对都会被 rehash 至 ht[1] ， 这时程序将 rehashidx 属性的值设为 -1 ， 表示 rehash 操作已完成。

渐进式 rehash 的好处在于它采取分而治之的方式， 将 rehash 键值对所需的计算工作均滩到对字典的每个添加、删除、查找和更新操作上， 从而避免了集中式 rehash 而带来的庞大计算量。

在后续的定时任务中以及 hash 的子指令中，循序渐进地将旧 hash 的内容一点点迁移到新的 hash 结构中

图 4-12 至图 4-17 展示了一次完整的渐进式 rehash 过程， 注意观察在整个 rehash 过程中， 字典的 rehashidx 属性是如何变化的。

渐进式 rehash 执行期间的哈希表操作

修改、删除、查询：因为在进行渐进式 rehash 的过程中， 字典会同时使用 ht[0] 和 ht[1] 两个哈希表， 所以在渐进式 rehash 进行期间， 字典的删除（delete）、查找（find）、更新（update）等操作会在两个哈希表上进行： 比如说， 要在字典里面查找一个键的话， 程序会先在 ht[0] 里面进行查找， 如果没找到的话， 就会继续到 ht[1] 里面进行查找， 诸如此类。

增加：另外， 在渐进式 rehash 执行期间， 新添加到字典的键值对一律会被保存到 ht[1] 里面， 而 ht[0] 则不再进行任何添加操作： 这一措施保证了 ht[0] 包含的键值对数量会只减不增， 并随着 rehash 操作的执行而最终变成空表。

redis rehash

https://blog.csdn.net/yuanrxdu/article/details/24779693 (Redis的字典(dict)rehash过程源码解析)

使dict出发rehash的条件有两个:

1. 总的元素个数 除 DICT桶的个数得到每个桶平均存储的元素个数(pre_num),如果 pre_num > dict_force_resize_ratio,就会触发dict 扩大操作。dict_force_resize_ratio = 5。
2. 在总元素 * 10 < 桶的个数，也就是,填充率必须<10%, DICT便会进行收缩，让total / bk_num 接近 1:1。

# 负载因子 = 哈希表已保存节点数量 / 哈希表大小

load_factor = ht[0].used / ht[0].size

因为已保存节点数量是包括冲突的节点数量，所以已保存节点数量是有可能大于哈希表大小的，所以也就可以达到5。但是java中hashmap的负载因子，指的是已经使用的桶的数量，而不是所有元素的大小

（平常 dict_force_resize_ratio=1，在后台持久化的时候dict_force_resize_ratio =5，为了减少备份的rehash对写时复制的印象）

服务器目前没有在执行 BGSAVE 命令或者 BGREWRITEAOF 命令， 并且哈希表的负载因子大于等于 1 ； 服务器目前正在执行BGSAVE 命令或者 BGREWRITEAOF 命令， 并且哈希表的负载因子大于等于 5 ；

持久化机制

在 Redis 中允许使用其中的一种、同时使用两种，或者两种都不用。

RDB的原理是什么？

你给出两个词汇就可以了，fork和cow。fork是指redis通过创建子进程来进行RDB操作，cow指的是copy on write，子进程创建后，父子进程共享数据段，父进程继续提供读写服务，写入的页面数据会逐渐和子进程分离开来。

快照恢复（RDB），通过快照（snapshotting）实现的，它是备份当前瞬间 Redis 在内存中的数据记录。如果当前 Redis 的数据量大，备份可能造成 Redis 卡顿，但是恢复重启是比较快速的。

# 当 900 秒执行 1 个写命令时，启用快照备份。
# 当 300 秒执行 10 个写命令时，启用快照备份。
# 当 60 秒内执行 10000 个写命令时，启用快照备份
save 900 1
save 300 10
save 60 10000
#  bgsave 异常的时候，将禁止写入命令。
stop-writes-on-bgsave-error yes 

save命令，会阻塞客户端的写入。生产建议禁用该命令，容易造成阻塞。

bgsave 命令，它和 save 命令最大的不同是它不会阻塞客户端的写入，也就是在执行 bgsave 的时候，允许客户端继续读/写 Redis。最终的实现还是调用rdbSave(char *filename)，只不过是通过fork()出的子进程来执行罢了，所以bgsave和save的实现是殊途同归。

在默认情况下，如果 Redis 执行 bgsave 失败后，Redis 将停止接受写操作，这样以一种强硬的方式让用户知道数据不能正确的持久化到磁盘，否则就会没人注意到灾难的发生，如果后台保存进程重新启动工作了，Redis 也将自动允许写操作。然而如果安装了靠谱的监控，可能不希望 Redis 这样做，那么你可以将其修改为 no。stop-writes-on-bgsave-error yes

备份的过程：根据RDB协议，把所有库逐个遍历，写入dump.rdb 文件中。

1. 首先创建一个临时文件。
2. 创建并初始化rio，rio是redis对io的一种抽象，提供了read、write、flush、checksum……等方法。
3. 调用 rdbSaveRio()，将当前 Redis 的内存信息全量写入到临时文件中。
4. 调用 fflush、 fsync 和 fclose 接口将文件写入磁盘中。
5. 使用 rename 将临时文件改名为 正式的 RDB 文件。
6. 将server.dirty清零，server.dirty是用了记录在上次生成rdb后有多少次数据变更，会在serverCron中用到。

Redis每次快照持久化都是将内存数据完整写入到磁盘一次，并不是增量的只同步脏数据。如果数据量大的话，而且写操作比较多，必然会引起大量的磁盘io操作，可能会严重影响性能，因此redis会fork一个子进程出来干活，这也是为什么线上禁止使用SAVE命令的原因：可能会导致Redis阻塞！

写时复制机制 COW

参考：Redis-关于RDB的几点顿悟-COW（Copy On Write）

COW（Copy On Write）

核心思路：fork一个子进程，只有在父进程发生写操作修改内存数据时，才会真正去分配内存空间，并复制内存数据，而且也只是复制被修改的内存页中的数据，并不是全部内存数据。

因为子进程没有数据变化，它看到的内存里的数据在进程产生的一瞬间就确定了。这也是redis的持久化叫 快照 的原因。

• Redis中执行BGSAVE命令生成RDB文件时，本质就是调用Linux中的fork()命令，Linux下的fork()系统调用实现了copy-on-write写时复制；
• fork()是类Unix操作系统上创建线程的主要方法，fork用于创建子进程（等同于当前进程的副本）；
• 传统的普通进程复制，会直接将父进程的数据拷贝到子进程中，拷贝完成后，父进程和子进程之间的数据段和堆栈是相互独立的；
• copy-on-write技术，在fork出子进程后，与父进程共享内存空间，两者只是虚拟空间不同，但是其对应的物理空间是同一个；

Linux中CopyOnWrite实现原理

fork()之后，kernel把父进程中所有的内存页的权限都设为read-only，然后子进程的地址空间指向父进程。当父子进程都只读内存时，相安无事。当其中某个进程写内存时，CPU硬件检测到内存页是read-only的，于是触发页异常中断（page-fault），陷入kernel的一个中断例程。中断例程中，kernel就会把触发的异常的页复制一份，于是父子进程各自持有独立的一份。

CopyOnWrite的好处：

1. 减少分配和复制资源时带来的瞬时延迟；可以保证子进程快速构建，不耗费性能
2. 减少不必要的资源分配；可以保证数据完整性，不会因为断电后导致原有数据错乱

CopyOnWrite的缺点：

1. 如果父子进程都需要进行大量的写操作，会产生大量的分页错误（页异常中断page-fault）;

Redis中的CopyOnWrite

• Redis在持久化时，如果是采用BGSAVE命令或者BGREWRITEAOF的方式，那Redis会fork出一个子进程来读取数据，从而写到磁盘中。
• 总体来看，Redis还是读操作比较多。如果子进程存在期间，发生了大量的写操作，那可能就会出现很多的分页错误(页异常中断page-fault)，这样就得耗费不少性能在复制上。
• 而在rehash阶段上，写操作是无法避免的。所以Redis在fork出子进程之后，将负载因子阈值提高，尽量减少写操作，避免不必要的内存写入操作，最大限度地节约内存。

RDB的FAQ

问题：

1. RDB的过程中是否会停止对外提供服务？（会提供服务）
2. RDB的过程中数据修改了，备份的是修改前的还是修改后的？（只备份，页修改，出现page-fail异常，然后父子线程各保存一份要修改的数据页）
3. RDB时是不是先把内容中的所有KV复制一份，保证数据不会被修改？（不是，是写时复制的那部分）
4. Redis是单线程的，那在RDB的过程中，是不是就没法对外提供服务了？Redis操作快的一个重要原因是Redis的数据是在内存中存储和操作的，持久化本身是磁盘的IO操作，IO操作又是特别耗时的，RDB备份的过程对Redis来说是挺漫长的，如果Redis没法对外提供服务的话，对Redis的影响是很大的吧？（bgsave 会fork子进程来持久化，主进程继续提供读写服务。因为有 COW 写时复制技术，所以影响不大。）
5. 知道备份时不会阻塞对外服务，那在数据备份的过程中，有新的数据变更的操作发生时，备份的是变更前的数据还是变更后的数据呢？（是精确的一个时刻的，所有数据页是只读的，写的时候会触发page-fail，然后父子进程各拷贝一份）
   • 另一个角度：RDB快照的是精确的一个时刻的内存数据呢？还是一段时间内的内存数据？
   • 另一个角度：RDB快照是精确的还是模糊的？
6. 既然是数据备份，在开始备份的时候，是不是要把Redis的所有数据现在内存中拷贝一份呢？那样的话平时Redis服务器的内存利用率就不能大于50%了啊？（不会的，核心就是 COW 写时复制机制的强大）
7. 写时复制，复制页面的数量问题？（肯定不会超过两倍内存的占用，另外一个 Redis 实例里冷数据占的比例往往是比较高的，所以很少会出现所有的页面都会被分离，被分离的往往只有其中一部分页 面。每个页面的大小只有 4K，一个 Redis 实例里面一般都会有成千上万的页面。 ）

解答：

1. RDB过程中会fork一个子进程，子进程做数据备份操作，主进程继续对外提供服务，所有Redis服务不会阻塞；
2. Copy On Write 机制，备份的是开始那个时刻内存中的数据；
3. Copy On Write 机制不需要把整个内存的数据都复制一份；

为什么在执行 BGSAVE 命令或者 BGREWRITEAOF 命令， 负载因子要大于等于 5？而未执行时大于等于1？

根据 BGSAVE 命令或 BGREWRITEAOF 命令是否正在执行， 服务器执行扩展操作所需的负载因子并不相同， 这是因为在执行BGSAVE 命令或BGREWRITEAOF 命令的过程中， Redis 需要创建当前服务器进程的子进程， 而大多数操作系统都采用写时复制（copy-on-write）技术来优化子进程的使用效率， 所以在子进程存在期间， 服务器会提高执行扩展操作所需的负载因子， 从而尽可能地避免在子进程存在期间进行哈希表扩展操作， 这可以避免不必要的内存写入操作，最大限度地节约内存。

如果进行rehash操作的话，会将整个字典中的数据全部重新分配一次内存，导致产生大量复制。

为什么负载因子可以大于1，并且达到5？HashMap也就0.75而已。

# 负载因子 = 哈希表已保存节点数量 / 哈希表大小
load_factor = ht[0].used / ht[0].size

因为已保存节点数量是包括冲突的节点数量，所以已保存节点数量是有可能大于哈希表大小的，所以也就可以达到5。java中是桶的使用量，不是总元素的个数。

RDB实现源码阅读

https://www.jianshu.com/p/131cf929a262

RDB相关源码在rdb.c中；通过saveCommand(redisClient c) 和bgsaveCommand(redisClient c) 两个方法可知，RDB持久化业务逻辑在rdbSave(server.rdb_filename)和rdbSaveBackground(server.rdb_filename这两个方法中；一个通过执行"save"触发，另一个通过执行"bgsave"或者save seconds changes条件满足时(在redis.c的serverCron中)触发：

redis.c里serverCron中通过调用rdbSaveBackground(server.rdb_filename)触发bgsave的部分代码：

if (server.dirty >= sp->changes &&
    server.unixtime-server.lastsave > sp->seconds &&
    (server.unixtime-server.lastbgsave_try >
     REDIS_BGSAVE_RETRY_DELAY ||
     server.lastbgsave_status == REDIS_OK))
{
    redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
        sp->changes, (int)sp->seconds);
    rdbSaveBackground(server.rdb_filename);
    break;
}

通过阅读rdbSaveBackground(char filename)的源码可知，其最终的实现还是调用rdbSave(char filename)，只不过是通过fork()出的子进程来执行罢了，所以bgsave和save的实现是殊途同归：

int rdbSaveBackground(char *filename) {
    pid_t childpid;
    long long start;

    // 如果已经有RDB持久化任务，那么rdb_child_pid的值就不是-1，那么返回REDIS_ERR；
    if (server.rdb_child_pid != -1) return REDIS_ERR;

    server.dirty_before_bgsave = server.dirty;
    server.lastbgsave_try = time(NULL);

    // 记录RDB持久化开始时间
    start = ustime();
    //fork一个子进程，
    if ((childpid = fork()) == 0) {
        // 如果fork()的结果childpid为0，即当前进程为fork的子进程，那么接下来调用rdbSave()进程持久化；
        int retval;

        /* Child */
        closeListeningSockets(0);
        redisSetProcTitle("redis-rdb-bgsave");
        // bgsave事实上就是通过fork的子进程调用rdbSave()实现, rdbSave()就是save命令业务实现；
        retval = rdbSave(filename);
        if (retval == REDIS_OK) {
            size_t private_dirty = zmalloc_get_private_dirty();

            if (private_dirty) {
                // RDB持久化成功后，如果是notice级别的日志，那么log输出RDB过程中copy-on-write使用的内存
                redisLog(REDIS_NOTICE,
                    "RDB: %zu MB of memory used by copy-on-write",
                    private_dirty/(1024*1024));
            }
        }
        exitFromChild((retval == REDIS_OK) ? 0 : 1);
    } else {
        // 父进程更新redisServer记录一些信息，例如：fork进程消耗的时间stat_fork_time, 
        /* Parent */
        server.stat_fork_time = ustime()-start;
       // 更新redisServer记录fork速率：每秒多少G；zmalloc_used_memory()的单位是字节，所以通过除以(1024*1024*1024),得到GB；由于记录的fork_time即fork时间是微妙，所以*1000000，得到每秒钟fork多少GB的速度；
        server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
        latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
        // 如果fork子进程出错，即childpid为-1，更新redisServer，记录最后一次bgsave状态是REDIS_ERR;
        if (childpid == -1) {
            server.lastbgsave_status = REDIS_ERR;
            redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
                strerror(errno));
            return REDIS_ERR;
        }
        redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
        // 最后在redisServer中记录的save开始时间重置为空，并记录执行bgsave的子进程id，即child_pid；
        server.rdb_save_time_start = time(NULL);
        server.rdb_child_pid = childpid;
        server.rdb_child_type = REDIS_RDB_CHILD_TYPE_DISK;
        updateDictResizePolicy();
        return REDIS_OK;
    }
    return REDIS_OK; /* unreached */
}

RDB持久化实现：

/* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success. */
int rdbSave(char *filename) {
    char tmpfile[256];
    FILE *fp;
    rio rdb;
    int error;
    // 文件临时文件名为temp-${pid}.rdb
    snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
    fp = fopen(tmpfile,"w");
    if (!fp) {
        redisLog(REDIS_WARNING, "Failed opening .rdb for saving: %s",
            strerror(errno));
        return REDIS_ERR;
    }

    rioInitWithFile(&rdb,fp);
    // RDB持久化的核心实现；
    if (rdbSaveRio(&rdb,&error) == REDIS_ERR) {
        errno = error;
        goto werr;
    }

    /* Make sure data will not remain on the OS's output buffers */
    if (fflush(fp) == EOF) goto werr;
    if (fsync(fileno(fp)) == -1) goto werr;
    if (fclose(fp) == EOF) goto werr;

    // 重命名rdb文件的命名；
    /* Use RENAME to make sure the DB file is changed atomically only
     * if the generate DB file is ok. */
    if (rename(tmpfile,filename) == -1) {
        redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
        unlink(tmpfile);
        return REDIS_ERR;
    }
    redisLog(REDIS_NOTICE,"DB saved on disk");
    server.dirty = 0;
    server.lastsave = time(NULL);
    server.lastbgsave_status = REDIS_OK;
    return REDIS_OK;

werr:
    redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
    fclose(fp);
    unlink(tmpfile);
    return REDIS_ERR;
}

rdbSaveRio--RDB持久化实现的核心代码--根据RDB文件协议将所有redis中的key-value写入rdb文件中:

/* Produces a dump of the database in RDB format sending it to the specified
 * Redis I/O channel. On success REDIS_OK is returned, otherwise REDIS_ERR
 * is returned and part of the output, or all the output, can be
 * missing because of I/O errors.
 *
 * When the function returns REDIS_ERR and if 'error' is not NULL, the
 * integer pointed by 'error' is set to the value of errno just after the I/O
 * error. */
int rdbSaveRio(rio *rdb, int *error) {
    dictIterator *di = NULL;
    dictEntry *de;
    char magic[10];
    int j;
    long long now = mstime();
    uint64_t cksum;

    if (server.rdb_checksum)
        rdb->update_cksum = rioGenericUpdateChecksum;
    // rdb文件中最先写入的内容就是magic，magic就是REDIS这个字符串+4位版本号
    snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);
    if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;

    // 遍历所有db重写rdb文件；
    for (j = 0; j < server.dbnum; j++) {
        redisDb *db = server.db+j;
        dict *d = db->dict;
        // 如果db的size为0，即没有任何key，那么跳过，遍历下一个db；
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);
        if (!di) return REDIS_ERR;

        // 写入REDIS_RDB_OPCODE_SELECTDB，这个值redis定义为254，即FE，再通过rdbSaveLen合入当前dbnum，例如当前db为0，那么写入FE 00
        /* Write the SELECT DB opcode */
        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_SELECTDB) == -1) goto werr;
        if (rdbSaveLen(rdb,j) == -1) goto werr;

        // 如注释所表达的，迭代遍历db这个dict的每一个entry；
        /* Iterate this DB writing every entry */
        while((de = dictNext(di)) != NULL) {
            // 先得到当前entry的key(sds类型)和value（redisObject类型）；
            sds keystr = dictGetKey(de);
            robj key, *o = dictGetVal(de);
            long long expire;

            initStaticStringObject(key,keystr);
            // 从redisDb的expire这个dict中查询过期时间属性值；
            expire = getExpire(db,&key);
            // 每个entry（redis中的key和其value）rdb持久化的核心代码
            if (rdbSaveKeyValuePair(rdb,&key,o,expire,now) == -1) goto werr;
        }
        dictReleaseIterator(di);
    }
    di = NULL; /* So that we don't release it again on error. */

    // 遍历所有db后，写入EOF这个opcode，REDIS_RDB_OPCODE_EOF申明为255，即FF，所以是写入FF到rdb文件中；FF是redis对rdb文件结束的定义；
    /* EOF opcode */
    if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EOF) == -1) goto werr;

    // 最后写入8个字节长度的checksum值到rdb文件尾部；
    /* CRC64 checksum. It will be zero if checksum computation is disabled, the
     * loading code skips the check in this case. */
    cksum = rdb->cksum;
    memrev64ifbe(&cksum);
    if (rioWrite(rdb,&cksum,8) == 0) goto werr;
    return REDIS_OK;

werr:
    if (error) *error = errno;
    if (di) dictReleaseIterator(di);
    return REDIS_ERR;
}

每个entry（key-value）rdb持久化的核心代码：

/* Save a key-value pair, with expire time, type, key, value.
 * On error -1 is returned.
 * On success if the key was actually saved 1 is returned, otherwise 0
 * is returned (the key was already expired). */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val,
                        long long expiretime, long long now)
{
    /* Save the expire time */
    if (expiretime != -1) {
        // 如果过期时间少于当前时间，那么表示该key已经失效，返回不做任何保存；
        /* If this key is already expired skip it */
        if (expiretime < now) return 0;
        // 如果当前遍历的entry有失效时间属性，那么保存REDIS_RDB_OPCODE_EXPIRETIME_MS即252，即"FC"以及失效时间到rdb文件中，
        if (rdbSaveType(rdb,REDIS_RDB_OPCODE_EXPIRETIME_MS) == -1) return -1;
        if (rdbSaveMillisecondTime(rdb,expiretime) == -1) return -1;
    }

    // 接下来保存redis key的类型，key，以及value到rdb文件中；
    /* Save type, key, value */
    if (rdbSaveObjectType(rdb,val) == -1) return -1;
    if (rdbSaveStringObject(rdb,key) == -1) return -1;
    if (rdbSaveObject(rdb,val) == -1) return -1;
    return 1;
}

RDB文件内容解析

有数据的文件：

REDIS // RDB协议约束的固定字符串,证明是rdb文件，5字节
0006 // redis的版本号，1字节
376 // SELECTDB 选择库
\0 // 0号数据库，默认有16个数据库，总数可在配置中修改，databases 16
\0 // 是数据类型type，0是string
004 // 数据长度
name // 是字符串key
004 // 值长度
jack // key的值
377 // EOF
377 后面的是checknum // 用来校验rdb文件的完整性

备注:
#define REDIS_RDB_TYPE_STRING 0 // 字符串 0
#define REDIS_RDB_TYPE_LIST 1 // list 1
#define REDIS_RDB_TYPE_SET 2 // set集合 2
#define REDIS_RDB_TYPE_ZSET 3 // zset 3
#define REDIS_RDB_TYPE_HASH 4 // hash 4

AOF

# 当前AOF文件比上次重写后的AOF文件大小的增长比例超过100 
auto-aof-rewrite-percentage 100  
# 当前AOF文件的文件大小大于64MB 
auto-aof-rewrite-min-size 64mb

追加文件（Append-Only File，AOF），其作用就是当 Redis 执行写命令后，在一定的条件下将执行过的写命令依次保存在 Redis 的文件中，将来就可以依次执行那些保存的命令恢复 Redis 的数据了。

对于 AOF 备份而言，它只是追加写入命令，所以备份一般不会造成 Redis 卡顿，但是恢复重启要执行更多的命令，备份文件可能也很大，使用者使用的时候要注意。

redis 还支持了 BGREWRITEAOF 指令，对appendonly.aof 进行重新整理。如果不经常进行数据迁移操作，推荐生产环境下的做法为关闭镜像，开启 appendonly.aof，同时可以选择在访问较少的时间每天对 appendonly.aof进行重写一次。

另外，对 master 机器,主要负责写，建议使用 AOF,对于 slave,主要负责读，挑选出 1-2 台开启 AOF，其余的建议关闭。

只进行追加文件操作。这里的文件追加记录是记录数据操作的改变记录，用以异常情况的数据恢复的。

我们都知道，redis作为一个内存数据库，数据的每次操作改变是先放在内存中，等到内存数据满了，在刷新到磁盘文件中，达到持久化的目的。

AOF实现源码阅读

源码解读：https://blog.csdn.net/androidlushangderen/article/details/40304889

aof的操作模式，也是采用了这样的方式。这里引入了一个block块的概念，其实就是一个缓冲区块。关于块的一些定义如下:

/* AOF的下面的一些代码都用到了一个简单buffer缓存块来进行存储，存储了数据的一些改变操作记录，等到
    缓冲中的达到一定的数据规模时，在持久化地写入到一个文件中，redis采用的方式是append追加的形式，这意味
    每次追加都要调整存储的块的大小，但是不可能会有无限大小的块空间，所以redis在这里引入了块列表的概念，
    设定死一个块的大小，超过单位块大小，存入另一个块中,这里定义每个块的大小为10M. */
#define AOF_RW_BUF_BLOCK_SIZE (1024*1024*10)    /* 10 MB per block */

/* 标准的aof文件读写块 */
typedef struct aofrwblock {
    //当前文件块被使用了多少，空闲的大小
    unsigned long used, free;
    //具体存储内容，大小10M
    char buf[AOF_RW_BUF_BLOCK_SIZE];
} aofrwblock;

也就是说，每个块的大小默认为10M,这个大小说大不大，说小不小了，如果填入的数据超出长度了，系统会动态申请一个新的缓冲块，在server端是通过一个块链表的形式，组织整个块的:

/* Append data to the AOF rewrite buffer, allocating new blocks if needed. */
/* 在缓冲区中追加数据，如果超出空间，会新申请一个缓冲块 */
void aofRewriteBufferAppend(unsigned char *s, unsigned long len) {
    listNode *ln = listLast(server.aof_rewrite_buf_blocks);
    //定位到缓冲区的最后一块，在最后一块的位置上进行追加写操作
    aofrwblock *block = ln ? ln->value : NULL;

    while(len) {
        /* If we already got at least an allocated block, try appending
         * at least some piece into it. */
        if (block) {
            //如果当前的缓冲块的剩余空闲能支持len长度的内容时，直接写入
            unsigned long thislen = (block->free < len) ? block->free : len;
            if (thislen) {  /* The current block is not already full. */
                memcpy(block->buf+block->used, s, thislen);
                block->used += thislen;
                block->free -= thislen;
                s += thislen;
                len -= thislen;
            }
        }

        if (len) { /* First block to allocate, or need another block. */
            int numblocks;
            //如果不够的话，需要新创建，进行写操作
            block = zmalloc(sizeof(*block));
            block->free = AOF_RW_BUF_BLOCK_SIZE;
            block->used = 0;
            //还要把缓冲块追加到服务端的buffer列表中
            listAddNodeTail(server.aof_rewrite_buf_blocks,block);

            /* Log every time we cross more 10 or 100 blocks, respectively
             * as a notice or warning. */
            numblocks = listLength(server.aof_rewrite_buf_blocks);
            if (((numblocks+1) % 10) == 0) {
                int level = ((numblocks+1) % 100) == 0 ? REDIS_WARNING :
                                                         REDIS_NOTICE;
                redisLog(level,"Background AOF buffer size: %lu MB",
                    aofRewriteBufferSize()/(1024*1024));
            }
        }
    }
}

当想要主动的将缓冲区中的数据刷新到持久化到磁盘中时，调用下面的方法:

/* Write the append only file buffer on disk.
 *
 * Since we are required to write the AOF before replying to the client,
 * and the only way the client socket can get a write is entering when the
 * the event loop, we accumulate all the AOF writes in a memory
 * buffer and write it on disk using this function just before entering
 * the event loop again.
 *
 * About the 'force' argument:
 *
 * When the fsync policy is set to 'everysec' we may delay the flush if there
 * is still an fsync() going on in the background thread, since for instance
 * on Linux write(2) will be blocked by the background fsync anyway.
 * When this happens we remember that there is some aof buffer to be
 * flushed ASAP, and will try to do that in the serverCron() function.
 *
 * However if force is set to 1 we'll write regardless of the background
 * fsync. */
#define AOF_WRITE_LOG_ERROR_RATE 30 /* Seconds between errors logging. */
/* 刷新缓存区的内容到磁盘中 */
void flushAppendOnlyFile(int force) {
    ssize_t nwritten;
    int sync_in_progress = 0;
    mstime_t latency;

    if (sdslen(server.aof_buf) == 0) return;

    if (server.aof_fsync == AOF_FSYNC_EVERYSEC)
        sync_in_progress = bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC) != 0;

    if (server.aof_fsync == AOF_FSYNC_EVERYSEC && !force) {
        /* With this append fsync policy we do background fsyncing.
         * If the fsync is still in progress we can try to delay
         * the write for a couple of seconds. */
        if (sync_in_progress) {
            if (server.aof_flush_postponed_start == 0) {
                /* No previous write postponinig, remember that we are
                 * postponing the flush and return. */
                server.aof_flush_postponed_start = server.unixtime;
                return;
            } else if (server.unixtime - server.aof_flush_postponed_start < 2) {
                /* We were already waiting for fsync to finish, but for less
                 * than two seconds this is still ok. Postpone again. */
                return;
            }
            /* Otherwise fall trough, and go write since we can't wait
             * over two seconds. */
            server.aof_delayed_fsync++;
            redisLog(REDIS_NOTICE,"Asynchronous AOF fsync is taking too long (disk is busy?). Writing the AOF buffer without waiting for fsync to complete, this may slow down Redis.");
        }
    }
    /* We want to perform a single write. This should be guaranteed atomic
     * at least if the filesystem we are writing is a real physical one.
     * While this will save us against the server being killed I don't think
     * there is much to do about the whole server stopping for power problems
     * or alike */

    //在进行写入操作的时候，还监听了延迟
    latencyStartMonitor(latency);
    nwritten = write(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
    latencyEndMonitor(latency);
    /* We want to capture different events for delayed writes:
     * when the delay happens with a pending fsync, or with a saving child
     * active, and when the above two conditions are missing.
     * We also use an additional event name to save all samples which is
     * useful for graphing / monitoring purposes. */
    if (sync_in_progress) {
        latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
    } else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) {
        latencyAddSampleIfNeeded("aof-write-active-child",latency);
    } else {
        latencyAddSampleIfNeeded("aof-write-alone",latency);
    }
    latencyAddSampleIfNeeded("aof-write",latency);

    /* We performed the write so reset the postponed flush sentinel to zero. */
    server.aof_flush_postponed_start = 0;

    if (nwritten != (signed)sdslen(server.aof_buf)) {
        static time_t last_write_error_log = 0;
        int can_log = 0;

        /* Limit logging rate to 1 line per AOF_WRITE_LOG_ERROR_RATE seconds. */
        if ((server.unixtime - last_write_error_log) > AOF_WRITE_LOG_ERROR_RATE) {
            can_log = 1;
            last_write_error_log = server.unixtime;
        }

        /* Lof the AOF write error and record the error code. */
        if (nwritten == -1) {
            if (can_log) {
                redisLog(REDIS_WARNING,"Error writing to the AOF file: %s",
                    strerror(errno));
                server.aof_last_write_errno = errno;
            }
        } else {
            if (can_log) {
                redisLog(REDIS_WARNING,"Short write while writing to "
                                       "the AOF file: (nwritten=%lld, "
                                       "expected=%lld)",
                                       (long long)nwritten,
                                       (long long)sdslen(server.aof_buf));
            }

            if (ftruncate(server.aof_fd, server.aof_current_size) == -1) {
                if (can_log) {
                    redisLog(REDIS_WARNING, "Could not remove short write "
                             "from the append-only file.  Redis may refuse "
                             "to load the AOF the next time it starts.  "
                             "ftruncate: %s", strerror(errno));
                }
            } else {
                /* If the ftrunacate() succeeded we can set nwritten to
                 * -1 since there is no longer partial data into the AOF. */
                nwritten = -1;
            }
            server.aof_last_write_errno = ENOSPC;
        }

        /* Handle the AOF write error. */
        if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
            /* We can't recover when the fsync policy is ALWAYS since the
             * reply for the client is already in the output buffers, and we
             * have the contract with the user that on acknowledged write data
             * is synched on disk. */
            redisLog(REDIS_WARNING,"Can't recover from AOF write error when the AOF fsync policy is 'always'. Exiting...");
            exit(1);
        } else {
            /* Recover from failed write leaving data into the buffer. However
             * set an error to stop accepting writes as long as the error
             * condition is not cleared. */
            server.aof_last_write_status = REDIS_ERR;

            /* Trim the sds buffer if there was a partial write, and there
             * was no way to undo it with ftruncate(2). */
            if (nwritten > 0) {
                server.aof_current_size += nwritten;
                sdsrange(server.aof_buf,nwritten,-1);
            }
            return; /* We'll try again on the next call... */
        }
    } else {
        /* Successful write(2). If AOF was in error state, restore the
         * OK state and log the event. */
        if (server.aof_last_write_status == REDIS_ERR) {
            redisLog(REDIS_WARNING,
                "AOF write error looks solved, Redis can write again.");
            server.aof_last_write_status = REDIS_OK;
        }
    }
    server.aof_current_size += nwritten;

    /* Re-use AOF buffer when it is small enough. The maximum comes from the
     * arena size of 4k minus some overhead (but is otherwise arbitrary). */
    if ((sdslen(server.aof_buf)+sdsavail(server.aof_buf)) < 4000) {
        sdsclear(server.aof_buf);
    } else {
        sdsfree(server.aof_buf);
        server.aof_buf = sdsempty();
    }

    /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
     * children doing I/O in the background. */
    if (server.aof_no_fsync_on_rewrite &&
        (server.aof_child_pid != -1 || server.rdb_child_pid != -1))
            return;

    /* Perform the fsync if needed. */
    if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
        /* aof_fsync is defined as fdatasync() for Linux in order to avoid
         * flushing metadata. */
        latencyStartMonitor(latency);
        aof_fsync(server.aof_fd); /* Let's try to get this data on the disk */
        latencyEndMonitor(latency);
        latencyAddSampleIfNeeded("aof-fsync-always",latency);
        server.aof_last_fsync = server.unixtime;
    } else if ((server.aof_fsync == AOF_FSYNC_EVERYSEC &&
                server.unixtime > server.aof_last_fsync)) {
        if (!sync_in_progress) aof_background_fsync(server.aof_fd);
        server.aof_last_fsync = server.unixtime;
    }
}

当然有操作会对数据库中的所有数据，做操作记录，便宜用此文件进行全盘恢复:

/* Write a sequence of commands able to fully rebuild the dataset into
 * "filename". Used both by REWRITEAOF and BGREWRITEAOF.
 *
 * In order to minimize the number of commands needed in the rewritten
 * log Redis uses variadic commands when possible, such as RPUSH, SADD
 * and ZADD. However at max REDIS_AOF_REWRITE_ITEMS_PER_CMD items per time
 * are inserted using a single command. */
/* 将数据库的内容按照键值，再次完全重写入文件中 */
int rewriteAppendOnlyFile(char *filename) {
    dictIterator *di = NULL;
    dictEntry *de;
    rio aof;
    FILE *fp;
    char tmpfile[256];
    int j;
    long long now = mstime();

    /* Note that we have to use a different temp name here compared to the
     * one used by rewriteAppendOnlyFileBackground() function. */
    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
    fp = fopen(tmpfile,"w");
    if (!fp) {
        redisLog(REDIS_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
        return REDIS_ERR;
    }

    rioInitWithFile(&aof,fp);
    if (server.aof_rewrite_incremental_fsync)
        rioSetAutoSync(&aof,REDIS_AOF_AUTOSYNC_BYTES);
    for (j = 0; j < server.dbnum; j++) {
        char selectcmd[] = "*2\r\n$6\r\nSELECT\r\n";
        redisDb *db = server.db+j;
        dict *d = db->dict;
        if (dictSize(d) == 0) continue;
        di = dictGetSafeIterator(d);
        if (!di) {
            fclose(fp);
            return REDIS_ERR;
        }

        /* SELECT the new DB */
        if (rioWrite(&aof,selectcmd,sizeof(selectcmd)-1) == 0) goto werr;
        if (rioWriteBulkLongLong(&aof,j) == 0) goto werr;

        /* Iterate this DB writing every entry */
        //遍历数据库中的每条记录，进行日志记录
        while((de = dictNext(di)) != NULL) {
            sds keystr;
            robj key, *o;
            long long expiretime;

            keystr = dictGetKey(de);
            o = dictGetVal(de);
            initStaticStringObject(key,keystr);

            expiretime = getExpire(db,&key);

            /* If this key is already expired skip it */
            if (expiretime != -1 && expiretime < now) continue;

            /* Save the key and associated value */
            if (o->type == REDIS_STRING) {
                /* Emit a SET command */
                char cmd[]="*3\r\n$3\r\nSET\r\n";
                if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                /* Key and value */
                if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
                if (rioWriteBulkObject(&aof,o) == 0) goto werr;
            } else if (o->type == REDIS_LIST) {
                if (rewriteListObject(&aof,&key,o) == 0) goto werr;
            } else if (o->type == REDIS_SET) {
                if (rewriteSetObject(&aof,&key,o) == 0) goto werr;
            } else if (o->type == REDIS_ZSET) {
                if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
            } else if (o->type == REDIS_HASH) {
                if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
            } else {
                redisPanic("Unknown object type");
            }
            /* Save the expire time */
            if (expiretime != -1) {
                char cmd[]="*3\r\n$9\r\nPEXPIREAT\r\n";
                if (rioWrite(&aof,cmd,sizeof(cmd)-1) == 0) goto werr;
                if (rioWriteBulkObject(&aof,&key) == 0) goto werr;
                if (rioWriteBulkLongLong(&aof,expiretime) == 0) goto werr;
            }
        }
        dictReleaseIterator(di);
    }

    /* Make sure data will not remain on the OS's output buffers */
    if (fflush(fp) == EOF) goto werr;
    if (fsync(fileno(fp)) == -1) goto werr;
    if (fclose(fp) == EOF) goto werr;

    /* Use RENAME to make sure the DB file is changed atomically only
     * if the generate DB file is ok. */
    if (rename(tmpfile,filename) == -1) {
        redisLog(REDIS_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
        unlink(tmpfile);
        return REDIS_ERR;
    }
    redisLog(REDIS_NOTICE,"SYNC append only file rewrite performed");
    return REDIS_OK;

werr:
    fclose(fp);
    unlink(tmpfile);
    redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
    if (di) dictReleaseIterator(di);
    return REDIS_ERR;
}

系统同样开放了后台的此方法操作(原理就是和RDB分析的一样，用的是fork(),创建子线程)：

/* This is how rewriting of the append only file in background works:
 *
 * 1) The user calls BGREWRITEAOF
 * 2) Redis calls this function, that forks():
 *    2a) the child rewrite the append only file in a temp file.
 *    2b) the parent accumulates differences in server.aof_rewrite_buf.
 * 3) When the child finished '2a' exists.
 * 4) The parent will trap the exit code, if it's OK, will append the
 *    data accumulated into server.aof_rewrite_buf into the temp file, and
 *    finally will rename(2) the temp file in the actual file name.
 *    The the new file is reopened as the new append only file. Profit!
 */
/* 后台进行AOF数据文件写入操作 */
int rewriteAppendOnlyFileBackground(void)

appendfsync同步频率

# 刷新频率配置
# appendfsync always
appendfsync everysec
# appendfsync no

AOF 内存数据块的同步频率：

• always，其含义为当 Redis 执行命令的时候，则同时同步到 AOF 文件，这样会使得 Redis 同步刷新 AOF 文件，造成缓慢（因为涉及到I/O操作）。每次命令都会持久化，它的好处在于安全，坏处在于每次都持久化性能较差（影响读写性能）。

• evarysec 则代表每秒同步一次命令到 AOF 文件，这个同步指的是把缓冲块中的数据落盘。类似mysql中 redo log buffer 同步刷新策略，多久持久化一次。 备份可能会丢失 1 秒以内的命令，建议默认此选项。

• no 的时候，则由客户端调用命令执行备份，Redis 本身不备份文件。对于采用 always 配置的时候，每次命令都会持久化，它的好处在于安全，坏处在于每次都持久化性能较差。

采用 evarysec 则每秒同步，安全性不如 always，备份可能会丢失 1 秒以内的命令，但是隐患也不大，安全度尚可，性能可以得到保障。采用 no，则性能有所保障，但是由于失去备份，所以安全性比较差。建议采用默认配置 everysec，这样在保证性能的同时，也在一定程度上保证了安全性。

这个参数实际就是控制redis，调用Linux 的 glibc 提供了 fsync(int fd)函数 ，将指定文件的内容强制从内核缓存刷到磁盘。 fsync 是一个磁盘 IO 操作，它很慢！

AOF模式下evarysec频率，断电导致 AOF 文件出现不完整的情况，可以使用 redis-check-aof 工具来修复这一问题，这个工具会将 AOF 文件中不完整的信息移除，确保 AOF 文件完整可用。

AOF重写

想象一个场景，一个字符串被修改了10次，在普通的AOF持久化策略中，10次命令执行都会被保存，这无疑是不必要的开销。AOF重写功能能避免这个问题，字符串被修改了10次，在AOF重写之后的新AOF文件中只有一条命令。

再来一个例子：再数据库中有个list，list中有100条数据，是通过100个命令写入的，普通的AOF持久化会保存100条命令，而AOF重写可以把100条插入命令合为一条命令存储到新的AOF文件。

首先要说明：AOF重写功能的实现是基于数据库的，不是基于旧的AOF文件。简单来说就是执行AOF重写时会读取数据库中的数据，把各个数据的操作还原为高效的命令存储到新的AOF文件中。

AOF重写是发生在Redis子进程中的，就像BGSAVE一样。重写过程中产生的命令，会放入AOF重写缓冲区，重写期间的新命令会添加到这个缓冲池中，在重写完毕之后追加到AOF尾部即可。

混合持久化4.0

# 开启混合持久化,4中默认关闭，5中默认开启了。
aof-use-rdb-preamble yes

注意：混合持久化只发生于 AOF 重写过程。使用了混合持久化，重写后的新 AOF 文件前半段是 RDB 格式的全量数据，后半段是 AOF 格式的增量数据。

AOF文件重写过程与RDB快照bgsave工作过程有点相似，都是通过fork子进程，由子进程完成相应的操作，同样的在fork子进程简短的时间内，redis是阻塞的。

（1）开始bgrewriteaof，判断当前有没有bgsave命令(RDB持久化)/bgrewriteaof在执行，倘若有，则这些命令执行完成以后在执行。

（2）主进程fork出子进程，在这一个短暂的时间内，redis是阻塞的。

（3）主进程fork完子进程继续接受客户端请求。此时，客户端的写请求不仅仅写入aof_buf缓冲区，还写入aof_rewrite_buf重写缓冲区。一方面是写入aof_buf缓冲区并根据appendfsync策略同步到磁盘，保证原有AOF文件完整和正确。另一方面写入aof_rewrite_buf重写缓冲区，保存fork之后的客户端的写请求，防止新AOF文件生成期间丢失这部分数据。

（4.1）子进程写完新的AOF文件后，向主进程发信号，父进程更新统计信息。

（4.2）主进程把aof_rewrite_buf中的数据写入到新的AOF文件。

（5.）使用新的AOF文件覆盖旧的AOF文件，标志AOF重写完成。

AOF重写过程中的数据变更问题

Redis 引入了 AOF 重写缓冲区（aof_rewrite_buf_blocks），这个缓冲区在服务器创建子进程之后开始使用，当 Redis 服务器执行完一个写命令之后，它会同时将这个写命令追加到 AOF 缓冲区和 AOF 重写缓冲区。

这样一来可以保证：

1、现有 AOF 文件的处理工作会如常进行。这样即使在重写的中途发生停机，现有的 AOF 文件也还是安全的。

2、从创建子进程开始，也就是 AOF 重写开始，服务器执行的所有写命令会被记录到 AOF 重写缓冲区里面。

这样，当子进程完成 AOF 重写工作后，父进程会在 serverCron 中检测到子进程已经重写结束，则会执行以下工作：

1、将 AOF 重写缓冲区中的所有内容写入到新 AOF 文件中，这时新 AOF 文件所保存的数据库状态将和服务器当前的数据库状态一致。

2、对新的 AOF 文件进行改名，原子的覆盖现有的 AOF 文件，完成新旧两个 AOF 文件的替换。

AOF 重写缓冲区内容过多怎么办

将 AOF 重写缓冲区的内容追加到新 AOF 文件的工作是由主进程完成的，所以这一过程会导致主进程无法处理请求，如果内容过多，可能会使得阻塞时间过长，显然是无法接受的。

Redis 中已经针对这种情况进行了优化：（让子进程做一部分重写缓冲区中写入AOF文件的工作，减少，主进程的阻塞时间。）

1、在进行 AOF 后台重写时，Redis 会创建一组用于父子进程间通信的管道，同时会新增一个文件事件，该文件事件会将写入 AOF 重写缓冲区的内容通过该管道发送到子进程。

2、在重写结束后，子进程会通过该管道尽量从父进程读取更多的数据，每次等待可读取事件1ms，如果一直能读取到数据，则这个过程最多执行1000次，也就是1秒。如果连续20次没有读取到数据，则结束这个过程。

通过这些优化，Redis 尽量让 AOF 重写缓冲区的内容更少，以减少主进程阻塞的时间。

区别，优缺点，及时性

RDB 的优点：

1. RDB 文件是是经过压缩的二进制文件，占用空间很小，它保存了 Redis 某个时间点的数据集，很适合用于做备份。 RDB文件可能存在版本不兼容。
2. RDB 在恢复大数据集时的速度比 AOF 的恢复速度要快。

RDB 的缺点：

1. 可能会丢失部分数据。因为备份不及时。
2. DB 保存时使用 fork 子进程进行数据的持久化，如果数据比较大的话，fork 可能会非常耗时。

AOF优点：

1. 如果是everysec，最多只会丢失一秒钟的数据。
2. AOF文件是一个纯追加的日志文件。可以使用 redis-check-aof 工具也可以轻易地修复这种问题。
3. 当 AOF文件太大时，Redis 会自动在后台进行重写。优化AOF文件内容。多次修改，变成记录最新值。多次插入，合并成一条命令。
4. 文件可读性好。

AOF 的缺点

1. 对于相同的数据集，AOF 文件的大小一般会比 RDB 文件大。
2. fsync 策略，会对性能有一定影响。但是everysec 是可以满足需求的。

混合的优点：

1. 结合 RDB 和 AOF 的优点, 更快的重写和恢复。

混合的缺点：

1. ​ AOF 文件里面的 RDB 部分不再是 AOF 格式，可读性差。

慢操作日志

慢日志(Slow log) 是 Redis 用来记录命令执行时间的日志系统。例如线上Redis突然出现堵塞，使用该命令可以查询Redis服务器耗时的命令列表，快速定位问题。

#  配置对执行时间大于多少微秒(microsecond, 1秒=10^6微秒) 的命令进行记录（内存中）。线上可以设置为1000微秒，也就是1毫秒。其中"操作时间"不包括网络 IO 开支,只包括请求达到 server 后进行"内存实施"的时间."0"表示记录全部操作。
slowlog-log-slower-than 10000
# 设置最大记录多少条记录。 slow log 本身是一个先进先出(FIFO) 队列，当队列大小超过该配置的值时，最旧的一条日志将被删除。线上可以设置为1000以上。
slowlog-max-len 128
# 通过命令动态修改
CONFIG SET slowlog-log-slower-than 10000
CONFIG SET slowlog-max-len 128
# 查看是否生效
CONFIG GET slowlog-log-slower-than
CONFIG GET slowlog-max-len
# 查看慢日志
SLOWLOG GET
# 查询慢日志数量
SLOWLOG LEN
# 清空慢查询日志
SLOWLOG RESET

日志结果：

127.0.0.1:6379> SLOWLOG GET
1) 1) (integer) 4
   2) (integer) 1627143167
   3) (integer) 2
   4) 1) "SLOWLOG"
      2) "LEN"
   5) "127.0.0.1:40852"
   6) ""
2) 1) (integer) 3
   2) (integer) 1627143164
   3) (integer) 3
   4) 1) "SLOWLOG"
      2) "LEN"
   5) "127.0.0.1:40852"
   6) ""

输出的结果含义：

1. 唯一性(unique)的日志标识符。日志的唯一 id 只有在 Redis 服务器重启的时候才会重置，这样可以避免对日志的重复处理。
2. 被记录命令的执行时间点，以 UNIX 时间戳格式表示
3. 查询执行时间，单位为微秒
4. 执行的命令，以数组的形式排列

TODO主从复制/哨兵/集群

• 主从复制（数据是同步的，类似于MySQL Replication）
• 哨兵 sentinel（数据是同步的）
• 集群（数据是分片的，sharing）
• Codis
• twemproxy

主从复制

https://redis.io/topics/replication

哨兵 Sentinel

redis主从复制模式下，主挂了怎么办？redis提供了哨兵模式（高可用）

脑裂

何谓哨兵模式？就是通过哨兵节点进行自主监控主从节点以及其他哨兵节点，发现主节点故障时自主进行故障转移。

单机架构

集群

其结构特点：

1、所有的redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽。 2、节点的fail是通过集群中超过半数的节点检测失效时才生效。 3、客户端与redis节点直连,不需要中间proxy层.客户端不需要连接集群所有节点,连接集群中任何一个可用节点即可。 4、redis-cluster把所有的物理节点映射到[0-16383]slot上（不一定是平均分配）,cluster 负责维护node<->slot<->value。

5、Redis集群预分好16384个桶，当需要在 Redis 集群中放置一个 key-value 时，根据 CRC16(key) mod 16384的值，决定将一个key放到哪个桶中。

ping-pong 机制

过期策略，内存淘汰策略

过期策略

①、EXPIRE <key> <ttl> ：表示将键 key 的生存时间设置为 ttl 秒。
②、PEXPIRE <key> <ttl> ：表示将键 key 的生存时间设置为 ttl 毫秒。
③、EXPIREAT <key> <timestamp> ：表示将键 key 的生存时间设置为 timestamp 所指定的秒数时间戳。
④、PEXPIREAT <key> <timestamp> ：表示将键 key 的生存时间设置为 timestamp 所指定的毫秒数时间戳。
PS：在Redis内部实现中，前面三个设置过期时间的命令最后都会转换成最后一个PEXPIREAT 命令来完成。

PERSIST <key> ：表示将key的过期时间移除。
TTL <key> ：以秒的单位返回键 key 的剩余生存时间。
PTTL <key> ：以毫秒的单位返回键 key 的剩余生存时间。

# 字符串独有
 SETEX KEY_NAME TIMEOUT VALUE

在Redis内部，每当我们设置一个键的过期时间时，Redis就会将该键带上过期时间存放到一个过期字典中。当我们查询一个键时，Redis便首先检查该键是否存在过期字典中，如果存在，那就获取其过期时间。然后将过期时间和当前系统时间进行比对，比系统时间大，那就没有过期；反之判定该键过期。

过期删除策略

定时删除:

• 在设置某个key 的过期时间同时，我们创建一个定时器，让定时器在该过期时间到来时，立即执行对其进行删除的操作。
• 优点：定时删除对内存是最友好的，能够保存内存的key一旦过期就能立即从内存中删除。
• 缺点：对CPU最不友好，在过期键比较多的时候，删除过期键会占用一部分 CPU 时间，对服务器的响应时间和吞吐量造成影响。

惰性删除:

• 设置该key 过期时间后，我们不去管它，当需要该key时，我们在检查其是否过期，如果过期，我们就删掉它，反之返回该key。
• 优点：对 CPU友好，我们只会在使用该键时才会进行过期检查，对于很多用不到的key不用浪费时间进行过期检查。
• 缺点：对内存不友好，如果一个键已经过期，但是一直没有使用，那么该键就会一直存在内存中，如果数据库中有很多这种使用不到的过期键，这些键便永远不会被删除，内存永远不会释放。从而造成内存泄漏。

定期删除:

• 每隔一段时间，我们就对一些key进行检查，删除里面过期的key。
• 优点：可以通过限制删除操作执行的时长和频率来减少删除操作对 CPU 的影响。另外定期删除，也能有效释放过期键占用的内存。
• 缺点：
  • 难以确定删除操作执行的时长和频率。
  • 如果执行的太频繁，定期删除策略变得和定时删除策略一样，对CPU不友好。
  • 如果执行的太少，那又和惰性删除一样了，过期键占用的内存不会及时得到释放。
  • 另外最重要的是，在获取某个键时，如果某个键的过期时间已经到了，但是还没执行定期删除，那么就会返回这个键的值，这是业务不能忍受的错误。(过了而没有删除)

Redis过期删除策略

前面讨论了删除过期键的三种策略，发现单一使用某一策略都不能满足实际需求，聪明的你可能想到了，既然单一策略不能满足，那就组合来使用吧。

没错，Redis的过期删除策略就是：惰性删除和定期删除两种策略配合使用。惰性删除，解决了定期删除时，还没来得及删除的问题。

惰性删除：Redis的惰性删除策略由 db.c/expireIfNeeded 函数实现，所有键读写命令执行之前都会调用 expireIfNeeded 函数对其进行检查，如果过期，则删除该键，然后执行键不存在的操作；未过期则不作操作，继续执行原有的命令。

定期删除：由redis.c/activeExpireCycle 函数实现，函数以一定的频率运行，每次运行时，都从一定数量的数据库中随机取出一定数量的随机键进行检查，并删除其中的过期键。

# 默认每秒运行10次，大概100ms运行一次。频率太高容易对CPU造成压力
hz 10

内存淘汰策略

Eviction policies(驱逐策略)，当maxmemory达到限制时，Redis 的确切行为是使用maxmemory-policy配置指令配置的。在 64bit 系统下，maxmemory设置为 0 表示不限制 Redis 内存使用（但是通常会设定其为物理内存的四分之三），在 32bit 系统下，maxmemory隐式不能超过 3GB。 当 Redis 内存使用达到指定的限制时，就需要选择一个置换的策略。

内存淘汰方式配置：

# 64位默认配置
config get maxmemory # 查看配置
maxmemory 0
maxmemory-policy noeviction
# 可选策略
noeviction：当内存使用超过配置的时候会返回错误，不会驱逐任何键，直接返回 OOM 异常。(OOM，不驱逐)
allkeys-lru：加入键的时候，如果过限，首先通过LRU算法驱逐最久没有使用的键（所有键LRU）
volatile-lru：加入键的时候如果过限，首先从设置了过期时间的键集合中驱逐最久没有使用的键(设置了key过期字典中，LRU)
allkeys-random：加入键的时候如果过限，从所有key随机删除(所有键，随机)
volatile-random：加入键的时候如果过限，从过期键的集合中随机驱逐（过期字典中，随机）
volatile-ttl：从配置了过期时间的键中驱逐马上就要过期的键（过期字典中，马上要过期的键）
volatile-lfu：从所有配置了过期时间的键中驱逐使用频率最少的键（过期字典中，频率最少的）
allkeys-lfu：从所有键中驱逐使用频率最少的键（所有键，频率最少的）

LRU 存储结构原理？

volatile-xxx 策略只会针对带过期时间的 key 进行淘汰，allkeys-xxx 策略会对所有的key 进行淘汰。如果你只是拿 Redis 做缓存，那应该使用 allkeys-xxx，客户端写缓存时不必携带过期时间。如果你还想同时使用 Redis 的持久化功能，那就使用 volatile-xxx策略，这样可以保留没有设置过期时间的 key，它们是永久的 key 不会被 LRU 算法淘汰。

近似LRU

LRU算法需要在原有结构上附加一个链表。当某个元素被访问时，它在链表中的位置就会被移动到表头，这样位于链表尾部的元素就是最近最少使用的元素，优先被踢掉；位于链表头部的元素就是最近刚被使用过的元素，暂时不会被踢。

Redis 使用近似LRU的原因：

• Redis 使用的是一种近似 LRU 算法，之所以不用传统的 LRU 是因为它引入了链表，会占用较多的内存。
• 近似 LRU 算法在现有数据结构的基础上采用随机采样的方式来淘汰元素，它为每个 key 增加了一个最后一次被访问的时间戳，当内存不足时，就执行一次近似 LRU 算法，具体步骤是随机采样 5 个 key，这个采样个数默认为 5，然后根据时间戳淘汰掉最旧的那个 key，如果淘汰后内存还是不足，就继续随机采样来淘汰。这里的淘汰策略如果设置的是 allkeys，就从所有 key 中随机采样，如果设置的是 volatile，就从设有过期时间的 key 中随机采样，采样值越大，效果就越接近传统的 LRU 算法。
• redis 3.0 在算法中增加了淘汰池，进一步提升了近似 LRU 算法的效果。具体原理是构建一个淘汰池数组，在每一次淘汰循环中，新随机采样的 key 会和淘汰池中的 key 进行融合，淘汰掉最旧的那个 key 后，保留剩余的 key 放入淘汰池中等待下一次循环。

# 采样 key 的数量，默认是 5，增大这个可以逼近 真正的LRU算法。
maxmemory-samples 5

LFU 原理

实现lfu需要两个关键的字段，一个是key创建时间戳，另一个是访问总数。为了复用字段，redis复用了key的时间戳字段，将时间戳字段一分为二，高16位用于存储分钟级别的时间戳，低八位用于记录访问总数counter值，八位二进制最大值为255。但是key高并发情况下，1000以上的qps也不足为奇。为了将访问次数缩放到255以内，redis引入了server.lfu_log_factor配置值，通过这个配置值，即使是千万级别的访问量，redis也能将其缩放到255以内，redis是通过如下这个方法实现缩放counter值。

防止低八位的的counter 大于255的算法：

uint8_t LFULogIncr(uint8_t counter) {
if (counter == 255) return 255; 
//取一随机小数
double r = (double)rand()/RAND_MAX; 
//counter减去初始值5，设置初始值的目的是防止key刚被放入就被淘汰
    double baseval = counter - LFU_INIT_VAL;
if (baseval < 0) baseval = 0;
//server.lfu_log_factor默认为10
double p = 1.0/(baseval*server.lfu_log_factor+1);
// counter越大,则p越小，counter获得增长的机会也越小
    if (r < p) counter++; 
    return counter;
}

新生key问题，对于新加入缓存的key，因为还没有被访问过，计数器的值如果为0，就算这个key是热点key，因为计数器值太小，也会被淘汰机制淘汰掉。为了解决这个问题，Redis会为新生key的计数器设置一个初始值。

如何避免临时高频访问的key常驻内存呢？

redis采用了一种策略，它会让key的访问次数随着时间衰减。

unsigned long LFUDecrAndReturn(robj *o) {
    //分钟时间戳
unsigned long ldt = o->lru >> 8; 
//当前counter值
unsigned long counter = o->lru & 255;
// 默认每经过一分钟counter衰减1
unsigned long num_periods = server.lfu_decay_time ? LFUTimeElapsed(ldt) / server.lfu_decay_time : 0;
if (num_periods)
    //计算衰减后的值
        counter = (num_periods > counter) ? 0 : counter - num_periods; 
    return counter;
}

为了避免排序过程，redis采用了如下的设计方案。

redis新增了pool机制， redis每次都将随机选择的10个key放在pool中，但是随机选择的key的时间戳必须小于pool中最小的key的时间戳才会继续放入，直到pool放满了，如果有新的key需要放入，那么需要将池中最大的一个时间戳的key取出。

Redis 懒惰删除

删除指令 del 会直接释放对象的内存，大部分情况下，这个指令非常快，没有明显延迟。不过如果删除的 key 是一个非常大的对象，比如一个包含了千万元素的 hash，那么删除操作就会导致单线程卡顿。

Redis 为了解决这个卡顿问题，在 4.0 版本引入了 unlink 指令，它能对删除操作进行懒处理，丢给后台线程来异步回收内存。

> unlink key

可以将整个 Redis 内存里面所有有效的数据想象成一棵大树。当 unlink 指令发出时，它只是把大树中的一个树枝别断了，然后扔到旁边的火堆里焚烧 (异步线程池)。树枝离开大树的一瞬间，它就再也无法被主线程中的其它指令访问到了，因为主线程只会沿着这颗大树来访问。

flush

Redis 提供了 flushdb 和 flushall 指令，用来清空数据库，这也是极其缓慢的操作。Redis 4.0 同样给这两个指令也带来了异步化，在指令后面增加 async 参数就可以将整棵大树连根拔起，扔给后台线程慢慢焚烧。

> flushall async

flushdb、flushall 区别

• flushall 清空数据库并执行持久化操作,也就是rdb文件会发生改变,变成76个字节大小(初始状态下为76字节),所以执行flushall之后数据库真正意义上清空了.
• flushdb 清空数据库,但是不执行持久化操作,也就是说rdb文件不发生改变.而redis的数据是从rdb快照文件中读取加载到内存的,所以在flushdb之后,如果想恢复数据库,则可以直接kill掉redis-server进程,然后重新启动服务,这样redis重新读取rdb文件,数据恢复到flushdb操作之前的状态.

注意:要直接kill 掉redis-server服务,因为shutdown操作会触发持久化.

Redis事务

官方：Redis 事务

事务是指一个完整的动作，要么全部执行，要么什么也没有做。

Redis 事务不是严格意义上的事务，只是用于帮助用户在一个步骤中执行多个命令。单个 Redis 命令的执行是原子性的，但 Redis 没有在事务上增加任何维持原子性的机制，所以 Redis 事务的执行并不是原子性的。

注： Redis 的事务根本不能算「原子性」，而仅仅是满足了事务的「隔离性」，隔离性中的串行化——当前执行的事务有着不被其它事务打断的权利。

Redis 事务可以理解为一个打包的批量执行脚本，但批量指令并非原子化的操作，中间某条指令的失败不会导致前面已做指令的回滚，也不会造成后续的指令不做。

DISCARD #取消事务，放弃执行事务块内的所有命令
EXEC    #执行所有事务块内的命令
MULTI   #标记一个事务块的开始
UNWATCH #取消 WATCH 命令对所有 key 的监视
WATCH   #监视一个(或多个) key

Redis 事务错误

https://www.redis.com.cn/redis-transaction.html

multi选择服务端缓冲

两类错误：

1. 调用 EXEC 之前的错误
2. 调用 EXEC 之后的错误

调用 EXEC 之前的错误，有可能是由于语法有误导致的，也可能时由于内存不足导致的。只要出现某个命令无法成功写入缓冲队列的情况，redis 都会进行记录，在客户端调用 EXEC 时，redis 会拒绝执行这一事务。

127.0.0.1:6379> multi
OK
127.0.0.1:6379> haha //一个明显错误的指令
(error) ERR unknown command 'haha'
127.0.0.1:6379> ping
QUEUED
127.0.0.1:6379> exec
# redis无情的拒绝了事务的执行，原因是“之前出现了错误”
(error) EXECABORT Transaction discarded because of previous errors.

调用 EXEC 之后的错误，redis 则采取了完全不同的策略，即 redis 不会理睬这些错误，而是继续向下执行事务中的其他命令。这是因为，对于应用层面的错误，并不是 redis 自身需要考虑和处理的问题，所以一个事务中如果某一条命令执行失败，并不会影响接下来的其他命令的执行。

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set age 23
QUEUED
#age不是集合，所以如下是一条明显错误的指令
127.0.0.1:6379> sadd age 15 
QUEUED
127.0.0.1:6379> set age 29
QUEUED
127.0.0.1:6379> exec #执行事务时，redis不会理睬第2条指令执行错误
1) OK
2) (error) WRONGTYPE Operation against a key holding the wrong kind of value
3) OK
127.0.0.1:6379> get age
"29" #可以看出第3条指令被成功执行了

Redis 乐观锁（WATCH）

指令WATCH，这是一个很好用的指令，它可以帮我们实现类似于“乐观锁”的效果，即CAS（check and set）。

WATCH 本身的作用是监视 key 是否被改动过，而且支持同时监视多个 key，只要还没真正触发事务，WATCH 都会尽职尽责的监视，一旦发现某个 key 被修改了，在执行 EXEC 时就会返回 nil，表示事务无法触发。

# 客户端1
127.0.0.1:6379> set age 23
OK
127.0.0.1:6379> watch age //开始监视age
OK
127.0.0.1:6379> set age 24 //在EXEC之前，age的值被修改了
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set age 25
QUEUED
127.0.0.1:6379> get age
QUEUED
127.0.0.1:6379> exec //触发EXEC
(nil) //事务无法被执行

# 客户端2，在客户端一 watch age，之后修改 age
set age 30

缺点：watch&multi&exec 并不是一个好主意，因为可能会不断循环重试，在竞争激烈时性能很差。

注意：Redis 禁止在 multi 和 exec 之间执行 watch 指令，而必须在 multi 之前做好盯住关键变量，否则会出错。

pipeline和multi对比

1. pipeline选择客户端缓冲，multi选择服务端缓冲；

2. 请求次数的不一致，multi需要每个命令都发送一次给服务端，pipeline最后一次性发送给服务端，请求次数相对于multi减少

3. multi/exec可以保证原子性，而pipeline不保证原子性
4. pipeline需要客户端自己开发配合，而且命令数不能太多，multi是redis服务端支持的，不需要client做任何事情复制代码

缓存问题

缓存穿透

缓存穿透的原因：

缓存穿透是指查询一个一定不存在的数据，由于缓存是不命中时被动写的，并且出于容错考虑，如果从存储层查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。在流量大时，可能 DB 就挂掉了。要是有人利用不存在的 key 频繁的攻击我们的应用，这就是漏洞。

方法1：布隆过滤器：最常见的方法就是采用布隆过滤器，将所有可能存在的数据哈希到一个足够大的 bitmap 中，一个一定不存在的数据会被 bitmap 拦截掉，从而避免了对底层存储系统的查询压力。

方法2：缓存空结果：另一个更为简单粗暴的方法（我们采用的就是这种），如果一个查询返回的数据为空（不管是数据不存在，还是系统故障），把这个空结果进行缓存，但是它的过期时间会很短，最长不超过五分钟。这种处理方式肯定是有问题的，假如传进来的这个不存在的Key值每次都是随机的，那存进Redis也没有意义。

缓存击穿

其实跟缓存雪崩有点类似，缓存雪崩是大规模的key失效，而缓存击穿是一个热点的Key，有大并发集中对其进行访问，突然间这个Key失效了，导致大并发全部打在数据库上，导致数据库压力剧增。这种现象就叫做缓存击穿。可以从两个方面解决，第一是否可以考虑热点key不设置过期时间，第二是否可以考虑降低打在数据库上的请求数量。

热点key的重建问题：

如果keyA是一个热点数据，那么可能在高并发下，多个线程都在重建keyA。重构的时间可能很长。

为了避免反复的重建缓存keyA提出下面两种方案：

1. 互斥锁：只有一个线程可以构建keyA，其他线程都被阻塞。构建完成以后，其他线程都可以直接获得该数据无需再重建keyA了。 缺点：在构建key的过程中，可能查询该数据的操作无法进行。

   

2. 永不过期：（对互斥锁的一个优化）redis中key不设置过期时间，而是给它一个逻辑过期时间。T2时间，如果发现keyA发生了变化需要重建，那就在逻辑过期时间上标记为过期，并fork一个新的线程去异步重建，在重建期间（T3）,T3会不断尝试去获取缓存无需等待，但是获取的是老的数据，直到T4时间，缓存重建成功，再去获取数据就是最新的数据了。从实战看，这种方法对于性能非常友好，唯一不足的就是构建缓存的时候，其余线程（非构建缓存的线程）可能访问的是老数据，但是对于一般的互联网功能来说这个还是可以忍受的。

// 永不过期的实现逻辑
String get(final String key) {
    V v = redis.get(key);
    String value = v.getValue();
    long timeout = v.getTimeout();
    if (v.timeout <= System.currentTimeMillis()) {
// 异步更新后台异常执行  
        threadPool.execute(new Runnable() {
            public void run() {
                String keyMutex = "mutex:" + key;
                if (redis.setnx(keyMutex, "1")) {
// 3 min timeout to avoid mutex holder crash  
                    redis.expire(keyMutex, 3 * 60);
                    String dbValue = db.get(key);
                    redis.set(key, dbValue);
                    redis.delete(keyMutex);
                }
            }
        });
    }
    return value;
}

缓存击穿和缓存雪崩的区别：

区别在于，缓存击穿是对某一个 “超级热点” key 缓存，而缓存雪崩是很多 key 某一时刻同时失效（可能时宕机，或过期时间一样，同时失效）。

相同的是，key 缓存失效，同时 恰好大量并发请求过来，这些请求发现缓存过期一般都是从后端 DB 加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端 DB 压垮。

缓存雪崩

缓存雪崩的原因

缓存雪崩指的是我们设置缓存时采用了相同的过期时间（或者宕机），导致缓存在某一时刻同时失效，请求全部转发到 DB，DB 瞬时压力过重雪崩。

缓存雪崩的解决方法

对于缓存雪崩，没有完美的解决方案，但是可以分析用户行为，尽量让失效时间均匀分布。大多数系统设计者考虑用枷锁、队列的方式，保证缓存的单线程（进程）写，从而避免失效时大量的并发请求落到底层存储系统上。

1. 缓存时间增加随机值：一个简单的方案，就是将缓存失效时间分散开，比如我们可以在原有的失效时间的基础上增加一个随机值，比如 1-5 分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。
2. 使用互斥锁(mutex key)，排队：思路：使用 mutex。就是在缓存失效的时候（判断拿出来的值为空），不是立即去 load db，而是先使用缓存工具的某些带成功操作返回值的操作（比如 Redis 的 setnx 或者 Memcache 的 add）去 set 一个 mutex key，当操作返回成功时，再进行 load db 的操作并回设缓存；否则，休眠一会儿（例如 50ms），重试 get 缓存的方法。
3. "提前"使用互斥锁(mutex key)：就是在 key 对应的 value 中，存一个比 实际缓存小的过期时间。每次取值，比较 value 中的过期时间，是否到了，如果到了，就进行枷锁，并更新值。
4. 做二级缓存，或者双缓存策略：A1 为原始缓存，A2 为拷贝缓存，A1 失效时，可以访问 A2， A1 缓存失效时间设置为短期，A2 设置为长期。

缓存永远不过期和 ”提前“ 使用互斥锁 思路类似，区别是前者是异步构建，后者是加锁构建，不会有旧数据的问题。

注：Redis setnx（SET if Not eXists）该命令在指定的 key 不存在时，为 key 设置指定的值；如果存在，则设置失败。

注：Memcache add 该命令指定的 key 已经存在，则不会更新数据（过期的 key 会更新），返回相应 NOT_STORED。保存成功 STORED。

// 使用互斥锁，重建缓存的逻辑
public String get(key) {
    String value = redis.get(key);
    if (value == null) { //代表缓存值过期
        //设置3min的超时，防止del操作失败的时候，下次缓存过期一直不能load db
        if (redis.setnx(key_mutex, 1, 3 * 60) == 1) {  //代表设置成功
            value = db.get(key);
            redis.set(key, value, expire_secs);
            redis.del(key_mutex);
        } else {  //这个时候代表同时候的其他线程已经load db并回设到缓存了，这时候重试获取缓存值即可
            sleep(50);
            get(key);  //重试
        }
    } else {
        return value;
    }
}

事前：针对宕机情况，提交缓存服务的高可用：使用集群缓存，保证缓存服务的高可用，这种方案就是在发生雪崩前对缓存集群实现高可用，如果是使用 Redis，可以使用 主从+哨兵 ，Redis Cluster 来避免 Redis 全盘崩溃的情况。

事中Hystrix限流&降级,避免MySQL被打死：使用 Hystrix进行限流 & 降级 ，比如一秒来了5000个请求，我们可以设置假设只能有一秒 2000个请求能通过这个组件，那么其他剩余的 3000 请求就会走限流逻辑。然后去调用我们自己开发的降级组件（降级），比如设置的一些默认值呀之类的。以此来保护最后的 MySQL 不会被大量的请求给打死。

事后：开启Redis持久化机制，尽快恢复缓存集群：一旦重启，就能从磁盘上自动加载数据恢复内存中的数据。

缓存预热

缓存预热如字面意思，当系统上线时，缓存内还没有数据，如果直接提供给用户使用，每个请求都会穿过缓存去访问底层数据库，如果并发大的话，很有可能在上线当天就会宕机，因此我们需要在上线前先将数据库内的热点数据缓存至Redis内再提供出去使用，这种操作就成为"缓存预热"。

缓存预热的实现方式有很多，比较通用的方式是写个批任务，在启动项目时或定时去触发将底层数据库内的热点数据加载到缓存内。

缓存更新

缓存服务（Redis）和数据服务（底层数据库）是相互独立且异构的系统，在更新缓存或更新数据的时候无法做到原子性的同时更新两边的数据，因此在并发读写或第二步操作异常时会遇到各种数据不一致的问题。如何解决并发场景下更新操作的双写一致是缓存系统的一个重要知识点。

第二步操作异常：缓存和数据的操作顺序中，第二个动作报错。如数据库被更新， 此时失效缓存的时候出错，缓存内数据仍是旧版本；

缓存更新的设计模式有四种：

• Cache aside：查询：先查缓存，缓存没有就查数据库，然后加载至缓存内；更新：先更新数据库，然后让缓存失效；或者先失效缓存然后更新数据库； （更新的时候让缓存1s后失效，然后让缓存在查询的时候自己加载）
• Read through：在查询操作中更新缓存，即当缓存失效时，Cache Aside 模式是由调用方负责把数据加载入缓存，而 Read Through 则用缓存服务自己来加载；
• Write through：在更新数据时发生。当有数据更新的时候，如果没有命中缓存，直接更新数据库，然后返回。如果命中了缓存，则更新缓存，然后由缓存自己更新数据库；
• Write behind caching：俗称write back回写，在更新数据的时候，只更新缓存，不更新数据库，缓存会异步地定时批量更新数据库；

1. 为了避免在并发场景下，多个请求同时更新同一个缓存导致脏数据，因此不能直接更新缓存而是令缓存失效。
2. 先更新数据库后失效缓存：并发场景下，推荐使用延迟失效（写请求完成后给缓存设置1s过期时间），在读请求缓存数据时若redis内已有该数据（其他写请求还未结束）则不更新。当redis内没有该数据的时候（其他写请求已令该缓存失效），读请求才会更新redis内的数据。这里的读请求缓存数据可以加上失效时间，以防第二步操作异常导致的不一致情况。
3. 先失效缓存后更新数据库：并发场景下，推荐使用延迟失效（写请求开始前给缓存设置1s过期时间），在写请求失效缓存时设置一个1s延迟时间，然后再去更新数据库的数据，此时其他读请求仍然可以读到缓存内的数据，当数据库端更新完成后，缓存内的数据已失效，之后的读请求会将数据库端最新的数据加载至缓存内保证缓存和数据库端数据一致性；在这种方案下，第二步操作异常不会引起数据不一致，例如设置了缓存1s后失效，然后在更新数据库时报错，即使缓存失效，之后的读请求仍然会把更新前的数据重新加载到缓存内。

缓存降级

缓存降级是指当访问量剧增、服务出现问题（如响应时间慢或不响应）或非核心服务影响到核心流程的性能时，即使是有损部分其他服务，仍然需要保证主服务可用。可以将其他次要服务的数据进行缓存降级，从而提升主服务的稳定性。

降级的目的是保证核心服务可用，即使是有损的。如去年双十一的时候淘宝购物车无法修改地址只能使用默认地址，这个服务就是被降级了，这里阿里保证了订单可以正常提交和付款，但修改地址的服务可以在服务器压力降低，并发量相对减少的时候再恢复。

降级可以根据实时的监控数据进行自动降级也可以配置开关人工降级。是否需要降级，哪些服务需要降级，在什么情况下再降级，取决于大家对于系统功能的取舍。

Redis 锁

参考：https://zhuanlan.zhihu.com/p/111354065?from_voters_page=true

这个文章介绍 Redis 分布式锁，没把我笑死

setnx

# 这个命令，可以使设置key和expire 变成原子命令。可以看成setnx和expire的结合体，是原子性的。
set key value [EX seconds] [PX milliseconds] [NX|XX]
EX seconds：设置失效时长，单位秒
PX milliseconds：设置失效时长，单位毫秒
NX：key不存在时设置value，成功返回OK，失败返回(nil)
XX：key存在时设置value，成功返回OK，失败返回(nil)

案例：设置name=p7+，失效时长100s，不存在时设置
1.1.1.1:6379> set name p7+ ex 100 nx
OK
1.1.1.1:6379> get name
"p7+"
1.1.1.1:6379> ttl name
(integer) 94

说到redis锁的时候，可以先从setnx讲起，最后慢慢引出set命令的可以加参数，可以体现出自己的知识面。

早在2013年，也就是7年前，Redis就发布了2.6.12版本，并且官网(set命令页)，也早早就说明了“SETNX, SETEX, PSETEX可能在未来的版本中，会弃用并永久删除”。

其实目前通常所说的setnx命令，并非单指redis的setnx key value这条命令。一般代指redis中对set命令加上nx参数进行使用（版本2.6.12之前，set是不支持nx参数的）， set这个命令，目前已经支持这么多参数可选：

SET key value [EX seconds|PX milliseconds] [NX|XX] [KEEPTTL]

主要是因为这个命令，可以指定过期事件，构成原子操作，而且可以满足 setnx 命令的特性。

• set 指令扩展：是在2.8版本中才引入的，而且官方也声明了 setnx、setex、psetex 会弃用。

• 过期时间：避免A线程拿到之后，自己挂了，没法释放锁，然后导致其它线程无法获得锁。过期时间就可以自动释放锁。

• 超时问题 ：如果锁的时间到了，但是逻辑没执行完，其它线程获得该锁，就会有问题。为了避免这个问题， Redis 分布式锁不要用于较长时间的任务。如果真的偶尔出现了，数据出现的小波错乱可能需要人工介入解决。

• value赋值UUID、或者随机数唯一值（解决超时问题）：避免上边的情况，A挂了，但是它的锁自动到期了，此时B获得锁成功，然后A线程又活了（或者是逻辑执行时间比锁的时间长），回手就把B线程获得的锁给删除了。增加了唯一value，就可以在A线程删除的时候（也叫解锁的时候），先获取value判断是否是当前进程加的锁，再去删除。（注：此时还有问题，在finally代码块中，get和del并非原子操作，还是有进程安全问题。）

• 使用lua脚本，保证get、del的原子性（解决删除锁的get、del原子问题）：EVAL "redis.call('SET', KEYS[1], ARGV[1]);redis.call('EXPIRE', KEYS[1], ARGV[2]); return 1;" 1 userAge 10 60

• 应用场景：击穿、雪崩，用这个加锁，解决并发改的问题。

  锁冲突解决办法（就是获取锁失败的情况）：

  1. 直接抛出异常， 通知用户稍后重试； 告诉客户，让客户点击，起到人工延时的效果。
  2. sleep 一会再重试；
  3. 将请求转移至延时队列， 过一会再试；

RedLock 红锁

https://blog.csdn.net/weixin_37512224/article/details/105439524 (大佬对战)

https://blog.csdn.net/jiangxiulilinux/article/details/107015292

https://redis.io/topics/distlock （官网）

分布式锁的基础：

获取锁：多客户端，使用 setnx 命令方式，同时在 redis 上创建相同的一个 key，因为 setnx 命令 不允许 key 重复，因此，谁创建成功，就像相当于谁获取到锁。

释放锁：在执行操作完成的之后，删除对应的 key，就相当于释放锁了。每个对应的 key 也有自己的失效时间，目的是为了方式死锁现象。（可能是删除 key 失败）

Redlock (Redis Distributed Lock)并非是一个工具，而是redis官方提出的一种分布式锁的算法。

假设有N个redis的master节点，这些节点是相互独立的（不需要主从或者其他协调的系统，RedLock作者指出，之所以要用独立的，是避免了redis异步复制造成的锁丢失，比如：主节点没来的及把刚刚set进来这条数据给从节点，就挂了。）。N推荐为奇数，不然出现并发，同时获得半数的锁。

客户端在获取锁时，需要做以下操作：

• 获取当前时间戳，以微妙为单为。
• 使用相同的lockName和lockValue，尝试从N个节点获取锁。（在获取锁时，要求等待获取锁的时间远小于锁的释放时间，如锁的lease_time为10s，那么wait_time应该为5-50毫秒；避免因为redis实例挂掉，客户端需要等待更长的时间才能返回，即需要让客户端能够fast_fail；如果一个redis实例不可用，那么需要继续从下个redis实例获取锁）（建议多线程，同时去获取所有节点的锁）
• 当从N个节点获取锁结束后，如果客户端能够从多数节点(N/2 + 1)中成功获取锁，且获取锁的时间小于失效时间，那么可认为，客户端成功获得了锁。（获取锁的时间=当前时间戳 - 步骤1的时间戳）
• 客户端成功获得锁后，那么锁的实际有效时间 = 设置锁的有效时间 - 获取锁的时间。
• 客户端获取锁失败后，N个节点的redis实例都会释放锁，即使某些节点未能加锁成功。

简明流程：

1. 顺序向五个节点请求加锁
2. 根据一定的超时时间来推断是不是跳过该节点
3. 三个节点加锁成功并且花费时间小于锁的有效期
4. 认定加锁成功

为什么N推荐为奇数呢？

原因1：本着最大容错的情况下，占用服务资源最少的原则，2N+1和2N+2的容灾能力是一样的，所以采用2N+1；比如，5台服务器允许2台宕机，容错性为2，6台服务器也只能允许2台宕机，容错性也是2，因为要求超过半数节点存活才OK。

原因2：假设有6个redis节点，client1和client2同时向redis实例获取同一个锁资源，那么可能发生的结果是——client1获得了3把锁，client2获得了3把锁，由于都没有超过半数，那么client1和client2获取锁都失败，对于奇数节点是不会存在这个问题。

失败时重试

当客户端无法获取到锁时，应该随机延时后进行重试，防止多个客户端在同一时间抢夺同一资源的锁（会导致脑裂，最终都不能获取到锁）。客户端获得超过半数节点的锁花费的时间越短，那么脑裂的概率就越低。所以，理想的情况下，客户端最好能够同时（并发）向所有redis发出set命令。

当客户端从多数节点获取锁失败时，应该尽快释放已经成功获取的锁，这样其他客户端不需要等待锁过期后再获取。（如果存在网络分区（特指跨机房断网的情况），客户端已经无法和redis进行通信，那么此时只能等待锁过期后自动释放）

TODO如何防止Redis脑裂导致数据丢失？

https://my.oschina.net/lishangzhi/blog/4742868 （介绍的很清楚）

redis 集群脑裂的原因？

端口受阻。网络线路抖动，导致通电信号受阻。。导致健康信号包发送不了也接收不了。

原因总结：master 并非真正的故障，由于网络原因，sentinel (哨兵) 心跳不能感知到 master ，然后新选举了一个 master，旧的 master 原来连接的客户端，仍然给 旧master 消息，这些消息 在 旧master 变为从库的时候，同步 新master 数据的时候，会清库，就丢失了。

redis的集群脑裂是指因为网络问题，导致redis master节点跟redis slave节点和sentinel集群处于不同的网络分区，此时因为sentinel集群无法感知到master的存在，所以将slave节点提升为master节点。此时存在两个不同的master节点，就像一个大脑分裂成了两个。

集群脑裂问题中，如果客户端还在基于原来的master节点继续写入数据，那么新的master节点将无法同步这些数据，当网络问题解决之后，sentinel集群将原先的master节点降为slave节点，此时再从新的master中同步数据，将会造成大量的数据丢失。

如何应对脑裂问题？

解决原理总结：通过 ACK 超时，最少需要的同步的从库的数量，来限制 master 接收客户端消息。从而避免消息丢失。

Redis 已经提供了两个配置项来限制主库的请求处理，分别是 min-slaves-to-write 和 min-slaves-max-lag。

• min-slaves-to-write：这个配置项设置了主库能进行数据同步的最少从库数量；
• min-slaves-max-lag：这个配置项设置了主从库间进行数据复制时，从库给主库发送 ACK 消息的最大延迟（以秒为单位）。

注：新版参数是： min-replicas-to-write 3 min-replicas-max-lag 10

我们可以把 min-slaves-to-write 和 min-slaves-max-lag 这两个配置项搭配起来使用，分别给它们设置一定的阈值，假设为 N 和 T。这两个配置项组合后的要求是，主库连接的从库中至少有 N 个从库，和主库进行数据复制时的 ACK 消息延迟不能超过 T 秒，否则，主库就不会再接收客户端的请求了。就可以减少数据同步之后的数据丢失

即使原主库是假故障，它在假故障期间也无法响应哨兵心跳，也不能和从库进行同步，自然也就无法和从库进行 ACK 确认了。这样一来，min-slaves-to-write 和 min-slaves-max-lag 的组合要求就无法得到满足，原主库就会被限制接收客户端请求，客户端也就不能在原主库中写入新数据了。

redisson客户端

Redisson是java的redis客户端之一，提供了一些api方便操作redis。Redisson普通的锁实现源码主要是RedissonLock这个类，源码中加锁/释放锁操作都是用lua脚本完成的，封装的非常完善，开箱即用。加锁解锁的lua脚本考虑的非常全面，其中就包括锁的重入性，这点可以说是考虑非常周全。用起来像jdk的ReentrantLock一样丝滑。

消息模式

异步队列，延迟队列

Redis 的 list(列表) 数据结构常用来作为异步消息队列使用，使用 rpush/lpush操作入队列，使用 lpop 和 rpop 来出队列。

队列空了怎么办？

• 休眠：线程休眠 sleep 1s 。缺点：睡眠会导致消息的延迟增大。
• 阻塞：使用 blpop/brpop 命令，阻塞读。缺点：空闲连接自动断开，这时候 blpop/brpop 会抛出异常来，所以编写客户端消费者的时候要小心，注意捕获异常，还要重试。

发布订阅

Redis 发布订阅（pub/sub）也是一种消息通信模式：发布者（pub）发送消息，订阅者（sub）接收消息。类似设计模式中的「观察者模式」。

# 订阅
SUBSCRIBE channel  [channel1...]
PSUBSCRIBE pattern [pattern1...]
# 发布
PUBLISH channel message
# 退订
UNSUBSCRIBE [channel1 [channel1...]]
PUNSUBSCRIBE [pattern [pattern...]]
# 查询
查看所有的频道：PUBSUB CHANNELS
查询订阅者的数量：PUBSUB NUMSUB
查询服务器被订阅者的数量：PUBSUB NUMPAT

缺点：不支持消息多播。如果某个订阅客户端因网络延迟失联，再来也接收不到期间的消息。

应用场景

1. 构建实时消息系统 ，比如普通的即时聊天，群聊等功能
2. 在一个博客网站，100个粉丝订阅了你，当你发布新文章，就可以推送消息给粉丝们
3. 微博，每个用户的粉丝都是该用户的订阅者，当用户发完微博，所有粉丝都将收到该用户的动态
4. 新闻，咨询站点通常有多个频道，每个频道就是一个主题，用户可以通过主题来订阅，这样有新闻发布时，订阅者可以获得更新
5. 集中配置中心管理，当配置信息发生更改后，订阅配置信息的节点可以收到通知消息

TODOStream

Redis5.0 被作者 Antirez 突然放了出来，增加了很多新的特色功能。而 Redis5.0 最大的新特性就是多出了一个数据结构 Stream，它是一个新的强大的支持多播的可持久化的消息队列，作者坦言 Redis Stream 狠狠地借鉴了 Kafka 的设计。

全局ID生成

https://tech.meituan.com/2017/04/21/mt-leaf.html

分布式场景下，要求全局唯一ID。

唯一ID有哪些特性或者说要求呢？按照分析有以下特性：

• 全局唯一性：全局唯一性：不能出现重复的ID号，既然是唯一标识，这是最基本的要求。
• 趋势递增：在MySQL InnoDB引擎中使用的是聚集索引，由于多数RDBMS使用B-tree的数据结构来存储索引数据，在主键的选择上面我们应该尽量使用有序的主键保证写入性能。
• 单调递增：保证下一个ID一定大于上一个ID，例如事务版本号、IM增量消息、排序等特殊需求。
• 信息安全：如果ID是连续的，恶意用户的扒取工作就非常容易做了，直接按照顺序下载指定URL即可；如果是订单号就更危险了，竞对可以直接知道我们一天的单量。所以在一些应用场景下，会需要ID无规则、不规则。
• 高可用：

数据库自增长序列

• 数字ID天然排序，对分页或者需要排序的结果很有帮助。
• 针对主库单点，如果有多个Master库，则每个Master库设置的起始数字不一样，步长一样，可以是Master的个数。比如：Master1 生成的是 1，4，7，10，Master2生成的是2,5,8,11 Master3生成的是 3,6,9,12。这样就可以有效生成集群中的唯一ID，也可以大大降低ID生成数据库操作的负载。(一：多台机器不同初始值、同步长自增；二：批量缓存自增ID)

UUID

一般来说全球唯一。UUID是指在一台机器在同一时间中生成的数字在所有机器中都是唯一的。按照开放软件基金会(OSF)制定的标准计算，用到了以太网卡地址、纳秒级时间、芯片ID码和许多可能的数字。

标准的UUID格式为：

xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx (8-4-4-4-12)

以连字号分为五段形式的36个字符，示例：

550e8400-e29b-41d4-a716-446655440000

Java标准类库中已经提供了UUID的API。

UUID.randomUUID()

优点：

1. 无需网络，单机自行生成
2. 各语言均有相应实现库供直接使用
3. 速度快，QPS高（支持100ns级并发）
4. 全球唯一，在遇见数据迁移，系统数据合并，或者数据库变更等情况下，可以从容应对。

缺点：

1. 没有排序，无法保证趋势递增。
2. String存储，占空间，DB查询及索引效率低
3. 存储空间比较大，UUID太长，16字节128位，通常以36长度的字符串表示，很多场景不适用，如果是海量数据库，就需要考虑存储量的问题。
4. 信息不安全，基于MAC地址生成UUID的算法可能会造成MAC地址泄露，这个漏洞曾被用于寻找梅丽莎病毒的制作者位置。
5. 传输数据量大，并且不可读。

SnowFlake雪花算法

snowflake是Twitter开源的分布式ID生成算法，结果是一个long型的ID。生成的是一个64位的二进制正整数，然后转换成10进制的数。64位二进制数由如下部分组成：

其核心思想是：使用41bit作为毫秒数，10bit作为机器的ID（5个bit是数据中心，5个bit的机器ID），12bit作为毫秒内的流水号（意味着每个节点在每毫秒可以产生 4096 个 ID），最后还有一个符号位，永远是0。

时钟回拨问题：

• 将ID生成交给少量服务器，并关闭时钟同步。
• 直接报错，交给上层业务处理。
• 如果回拨时间较短，在耗时要求内，比如5ms，那么等待回拨时长后再进行生成。
• 如果回拨时间很长，那么无法等待，可以匀出少量位（1~2位）作为回拨位，一旦时钟回拨，将回拨位加1，可得到不一样的ID，2位回拨位允许标记三次时钟回拨，基本够使用。如果超出了，可以再选择抛出异常。

优点：

1. 时间戳在高位，自增序列在低位，整个ID是趋势递增的，按照时间有序递增。
2. 灵活度高，可以根据业务需求，调整bit位的划分，满足不同的需求。百度的UidGenerator、美团的Leaf等，都是基于雪花算法做一些适合自身业务的变化。

缺点：

1. 依赖机器的时钟，如果服务器时钟回拨，会导致重复ID生成。
2. 在分布式环境上，每个服务器的时钟不可能完全同步，有时会出现不是全局递增的情况。

package com.leon.distributed.algorithm;

/**
* Twitter_Snowflake
* SnowFlake的结构如下(每部分用-分开):
* 0 - 0000000000 0000000000 0000000000 0000000000 0 - 00000 - 00000 - 000000000000
* 1位标识，由于long基本类型在Java中是带符号的，最高位是符号位，正数是0，负数是1，所以id一般是正数，最高位是0
* 41位时间截(毫秒级)，注意，41位时间截不是存储当前时间的时间截，而是存储时间截的差值（当前时间截 - 开始时间截)
* 得到的值），这里的的开始时间截，一般是我们的id生成器开始使用的时间，由我们程序来指定的（如下下面程序IdWorker类的startTime属性）。
* 41位的时间截，可以使用69年，年T = (1L << 41) / (1000L * 60 * 60 * 24 * 365) = 69
* 10位的数据机器位，可以部署在1024个节点，包括5位datacenterId和5位workerId
* 12位序列，毫秒内的计数，12位的计数顺序号支持每个节点每毫秒(同一机器，同一时间截)产生4096个ID序号
* 加起来刚好64位，为一个Long型。
* SnowFlake的优点是，整体上按照时间自增排序，并且整个分布式系统内不会产生ID碰撞(由数据中心ID和机器ID作区分)，并且效率较高，
* 经测试，我的渣机器SnowFlake每秒能够产生11万ID左右。
*/
public class SnowflakeIdWorker {

// ========Fields====================
   /** 开始时间截 (2015-01-01) */
   private final long twepoch = 1420041600000L;

   /** 机器id所占的位数 */
   private final long workerIdBits = 5L;

   /** 数据标识id所占的位数 */
   private final long datacenterIdBits = 5L;

   /** 支持的最大机器id，结果是31 (这个移位算法可以很快的计算出几位二进制数所能表示的最大十进制数) */
   private final long maxWorkerId = -1L ^ (-1L << workerIdBits);

   /** 支持的最大数据标识id，结果是31 */
   private final long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);

   /** 序列在id中占的位数 */
   private final long sequenceBits = 12L;

   /** 机器ID向左移12位 */
   private final long workerIdShift = sequenceBits;

   /** 数据标识id向左移17位(12+5) */
   private final long datacenterIdShift = sequenceBits + workerIdBits;

   /** 时间截向左移22位(5+5+12) */
   private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;

   /** 生成序列的掩码，这里为4095 (0b111111111111=0xfff=4095) */
   private final long sequenceMask = -1L ^ (-1L << sequenceBits);

   /** 工作机器ID(0~31) */
   private long workerId;

   /** 数据中心ID(0~31) */
   private long datacenterId;

   /** 毫秒内序列(0~4095) */
   private long sequence = 0L;

   /** 上次生成ID的时间截 */
   private long lastTimestamp = -1L;

   //===========Constructors=========
   /**
    * 构造函数
    * @param workerId 工作ID (0~31)
    * @param datacenterId 数据中心ID (0~31)
    */
   public SnowflakeIdWorker(long workerId, long datacenterId) {
       if (workerId > maxWorkerId || workerId < 0) {
           throw new IllegalArgumentException(String.format("worker Id can't be greater than %d or less than 0", maxWorkerId));
       }
       if (datacenterId > maxDatacenterId || datacenterId < 0) {
           throw new IllegalArgumentException(String.format("datacenter Id can't be greater than %d or less than 0", maxDatacenterId));
       }
       this.workerId = workerId;
       this.datacenterId = datacenterId;
   }

   // ===============Methods=================
   /**
    * 获得下一个ID (该方法是线程安全的)
    * @return SnowflakeId
    */
   public synchronized long nextId() {
       long timestamp = timeGen();

       //如果当前时间小于上一次ID生成的时间戳，说明系统时钟回退过这个时候应当抛出异常
       if (timestamp < lastTimestamp) {
           throw new RuntimeException(
                   String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
       }

       //如果是同一时间生成的，则进行毫秒内序列
       if (lastTimestamp == timestamp) {
           sequence = (sequence + 1) & sequenceMask;
           //毫秒内序列溢出
           if (sequence == 0) {
               //阻塞到下一个毫秒,获得新的时间戳
               timestamp = tilNextMillis(lastTimestamp);
           }
       }
       //时间戳改变，毫秒内序列重置
       else {
           sequence = 0L;
       }

       //上次生成ID的时间截
       lastTimestamp = timestamp;

       //移位并通过或运算拼到一起组成64位的ID
       return ((timestamp - twepoch) << timestampLeftShift) //
               | (datacenterId << datacenterIdShift) //
               | (workerId << workerIdShift) //
               | sequence;
   }

   /**
    * 阻塞到下一个毫秒，直到获得新的时间戳
    * @param lastTimestamp 上次生成ID的时间截
    * @return 当前时间戳
    */
   protected long tilNextMillis(long lastTimestamp) {
       long timestamp = timeGen();
       while (timestamp <= lastTimestamp) {
           timestamp = timeGen();
       }
       return timestamp;
   }

   /**
    * 返回以毫秒为单位的当前时间
    * @return 当前时间(毫秒)
    */
   protected long timeGen() {
       return System.currentTimeMillis();
   }

   //==================Test===========
   /** 测试 */
   public static void main(String[] args) {
       SnowflakeIdWorker idWorker = new SnowflakeIdWorker(0, 0);

       long startTime=System.currentTimeMillis();
       for (int i = 0; i < 500000; i++) {
           long id = idWorker.nextId();
           System.out.println(Long.toBinaryString(id));
           System.out.println(id);
       }
       long endTime=System.currentTimeMillis();
       System.out.println("当前程序耗时："+(endTime-startTime)+"ms");
   }
}

snowflake算法可以根据自身项目的需要进行一定的修改。比如估算未来的数据中心个数，每个数据中心的机器数以及统一毫秒可以能的并发数来调整在算法中所需要的bit数。例如：百度的UidGenerator、美团的Leaf等，都是基于雪花算法做一些适合自身业务的变化。

Redis生成ID

当使用数据库来生成ID性能不够要求的时候，我们可以尝试使用Redis来生成ID。这主要依赖于Redis是单线程的，所以也可以用生成全局唯一的ID。可以用Redis的原子操作 INCR和INCRBY来实现。

可以使用Redis集群来获取更高的吞吐量。假如一个集群中有5台Redis。可以初始化每台Redis的值分别是1,2,3,4,5，然后步长都是5。各个Redis生成的ID为：

A：1,6,11,16,21
B：2,7,12,17,22
C：3,8,13,18,23
D：4,9,14,19,24
E：5,10,15,20,25

这个，随便负载到哪个机子确定好，未来很难做修改。但是3-5台服务器基本能够满足器上，都可以获得不同的ID。但是步长和初始值一定需要事先需要了。使用Redis集群也可以解决单点故障的问题。

另外，比较适合使用Redis来生成每天从0开始的流水号。比如订单号=日期+当日自增长号。可以每天在Redis中生成一个Key，使用INCR进行累加。

优点：

1. 不依赖于数据库，灵活方便，且性能优于数据库。
2. 数字ID天然排序，对分页或者需要排序的结果很有帮助。

缺点：

1. 如果系统中没有Redis，还需要引入新的组件，增加系统复杂度。
2. 需要编码和配置的工作量比较大。
3. Redis是单线程的，若造成阻塞，则会引发高并发问题，需要处理好集群与主从关系

比较

一致性 Hash

hash_tag

https://www.jianshu.com/p/528ce5cd7e8f

MurMurHash算法,性能高,碰撞率低

应用场景

在使用分布式对数据进行存储时，经常会碰到需要新增节点来满足业务快速增长的需求。然而在新增节点时，如果处理不善会导致所有的数据重新分片，这对于某些系统来说可能是灾难性的。

那么是否有可行的方法，在数据重分片时，只需要迁移与之关联的节点而不需要迁移整个数据呢？当然有，在这种情况下我们可以使用一致性Hash来处理。

• redis-cluster把所有的物理节点映射到[0-16383]slot（插槽）上,cluster 负责维护node<->slot<->value
• 一致性Hash算法将流量均匀分发到不同服务器

特性：

• 平衡性(Balance)平衡性是指哈希的结果能够尽可能分布到所有的缓冲中去，这样可以使得所有的缓冲空间都得到利用。很多哈希算法都能够满足这一条件。
• 单调性(Monotonicity)单调性是指如果已经有一些内容通过哈希分派到了相应的缓冲中，又有新的缓冲区加入到系统中，那么哈希的结果应能够保证原有已分配的内容可以被映射到新的缓冲区中去，而不会被映射到旧的缓冲集合中的其他缓冲区。简单的哈希算法往往不能满足单调性的要求，如最简单的线性哈希：x = (ax + b) mod (P)，在上式中，P表示全部缓冲的大小。不难看出，当缓冲大小发生变化时(从P1到P2)，原来所有的哈希结果均会发生变化，从而不满足单调性的要求。哈希结果的变化意味着当缓冲空间发生变化时，所有的映射关系需要在系统内全部更新。而在P2P系统内，缓冲的变化等价于Peer加入或退出系统，这一情况在P2P系统中会频繁发生，因此会带来极大计算和传输负荷。单调性就是要求哈希算法能够应对这种情况。
• 分散性(Spread)在分布式环境中，终端有可能看不到所有的缓冲，而是只能看到其中的一部分。当终端希望通过哈希过程将内容映射到缓冲上时，由于不同终端所见的缓冲范围有可能不同，从而导致哈希的结果不一致，最终的结果是相同的内容被不同的终端映射到不同的缓冲区中。这种情况显然是应该避免的，因为它导致相同内容被存储到不同缓冲中去，降低了系统存储的效率。分散性的定义就是上述情况发生的严重程度。好的哈希算法应能够尽量避免不一致的情况发生，也就是尽量降低分散性。
• 负载(Load)负载问题实际上是从另一个角度看待分散性问题。既然不同的终端可能将相同的内容映射到不同的缓冲区中，那么对于一个特定的缓冲区而言，也可能被不同的用户映射为不同的内容。与分散性一样，这种情况也是应当避免的，因此好的哈希算法应能够尽量降低缓冲的负荷。
• 平滑性(Smoothness)平滑性是指缓存服务器的数目平滑改变和缓存对象的平滑改变是一致的。

实现原理

一致性Hash算法也是使用取模的方法，不过，上述的取模方法是对服务器的数量进行取模，而一致性的Hash算法是对2的32方取模。即，一致性Hash算法将整个Hash空间组织成一个虚拟的圆环，Hash函数的值空间为0 ~ 2^32 - 1(一个32位无符号整型)，整个哈希环如下：

整个圆环以顺时针方向组织，圆环正上方的点代表0，0点右侧的第一个点代表1，以此类推。 第二步，我们将各个服务器使用Hash进行一个哈希，具体可以选择服务器的IP或主机名作为关键字进行哈希，这样每台服务器就确定在了哈希环的一个位置上，比如我们有三台机器，使用IP地址哈希后在环空间的位置如图1-4所示：

现在，我们使用以下算法定位数据访问到相应的服务器：

将数据Key使用相同的函数Hash计算出哈希值，并确定此数据在环上的位置，从此位置沿环顺时针查找，遇到的服务器就是其应该定位到的服务器。

例如，现在有ObjectA，ObjectB，ObjectC三个数据对象，经过哈希计算后，在环空间上的位置如下：

根据一致性算法，Object -> NodeA，ObjectB -> NodeB, ObjectC -> NodeC

一致性Hash算法的容错性和可扩展性

现在，假设我们的Node C宕机了，我们从图中可以看到，A、B不会受到影响，只有Object C对象被重新定位到Node A。所以我们发现，在一致性Hash算法中，如果一台服务器不可用，受影响的数据仅仅是此服务器到其环空间前一台服务器之间的数据（这里为Node C到Node B之间的数据），其他不会受到影响。如图1-6所示：

作者：oneape15 链接：https://www.jianshu.com/p/528ce5cd7e8f 来源：简书 著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

数据倾斜问题

在一致性Hash算法服务节点太少的情况下，容易因为节点分布不均匀面造成数据倾斜（被缓存的对象大部分缓存在某一台服务器上）问题，如图1-8特例：

这时我们发现有大量数据集中在节点A上，而节点B只有少量数据。为了解决数据倾斜问题，一致性Hash算法引入了虚拟节点机制，即对每一个服务器节点计算多个哈希，每个计算结果位置都放置一个此服务节点，称为虚拟节点。 具体操作可以为服务器IP或主机名后加入编号来实现，实现如图1-9所示：

数据定位算法不变，只需要增加一步：虚拟节点到实际点的映射。 所以加入虚拟节点之后，即使在服务节点很少的情况下，也能做到数据的均匀分布。

每台机器映射的虚拟节点越多，则分布的越均匀~~~

Redis 安全

指令安全

keys 指令会导致 Redis 卡顿， flushdb 和 flushall 会让 Redis 的所有数据全部清空。

# 将某些危险的指令修改成特别的名称，用来避免人为误操作。
rename-command keys abckeysabc
# 将指令 rename 成空串,这个指令就无法使用了
rename-command flushall ""

端口安全

默认会监听 *:6379，如果当前的服务器主机有外网地址， Redis 的服务将会直接暴露在公网上，任何一个初级黑客使用适当的工具对 IP 地址进行端口扫描就可以探测出来。

# 指定访问的服务器ip
bind 10.100.20.13
# 增加 Redis 的密码访问限制
requirepass yoursecurepasswordhereplease

客户端：必须使用 auth 指令传入正确的密码才可以访问 Redis，

从库复制：必须在配置文件里使用 masterauth 指令配置相应的密码才可以进行复制操作 masterauth yoursecurepasswordhereplease

Lua 脚本安全

禁止用户数据参数，不然容易脚本注入。

Redis 用普通用户启动，这样即使恶意代码也无法拿到root权限。

SSL代理

如果要跨机房，暴漏在公网中，使用官方推荐的spiped SSL 代理软件 ，

面试问题

为什么快？

为什么说Redis是单线程的以及Redis为什么这么快！

官方提供的数据是可以达到100000+的QPS（每秒内查询次数）。

1. 完全基于内存，绝大部分请求是纯粹的内存操作，非常快速。数据存在内存中，类似于HashMap，HashMap的优势就是查找和操作的时间复杂度都是O(1)；（基于内存）
2. 数据结构简单，对数据操作也简单，Redis中的数据结构是专门进行设计的；
3. 采用单线程，避免了不必要的上下文切换和竞争条件，也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗；（避免线程切换，竟态消耗）
4. 使用多路I/O复用模型，非阻塞IO；
5. 使用底层模型不同，它们之间底层实现方式以及与客户端之间通信的应用协议不一样，Redis直接自己构建了VM 机制 ，因为一般的系统调用系统函数的话，会浪费一定的时间去移动和请求；

Redis VM机制

拒绝躺平，Redis 选择实现了自己的 VM

Redis 之 VM 机制

Redis 的 VM (虚拟内存)机制就是暂时把不经常访问的数据(冷数据)从内存交换到磁盘中，从而腾出宝贵的内存空间用于其它需要访问的数据(热数据)。通过 VM 功能可以实现冷热数据分离，使热数据仍在内存中、冷数据保存到磁盘。这样就可以避免因为内存不足而造成访问速度下降的问题。Redis 提高数据库容量的办法有两种：一种是可以将数据分割到多个 Redis Server上；另一种是使用虚拟内存把那些不经常访问的数据交换到磁盘上。「需要特别注意的是 Redis 并没有使用 OS 提供的 Swap，而是自己实现。」

Redis 为了保证查找的速度，只会将 value 交换出去，而在内存中保留所有的 Key。所以它非常适合 Key 很小，Value 很大的存储结构。如果 Key 很大，value 很小，那么vm可能还是无法满足需求。

VM 相关配置

通过在 redis 的 redis.conf 文件里，设置 VM 的相关参数来实现数据在内存和磁盘之间 换入和 换出操作。相关配置如下：

#开启vm功能
vm-enabled yes
#交换出来的value保存的文件路径
vm-swap-file /tmp/redis.swap
#设置当内存消耗达到上限时开始将value交换出来
vm-max-memory 1000000
#设置单个页面的大小，单位是字节
vm-page-size 32
#设置最多能交换保存多少个页到磁盘
vm-pages 13417728
#设置完成交换动作的工作线程数，设置为表示0不使用工作线程而使用主线程,这会以阻塞(会阻塞所有客户端)的方式来运行。建议设置成CPU核个数
vm-max-threads 4

redis 规定同一个数据页面只能保存一个对象，但一个对象可以保存在多个数据页面中。在 redis 使用的内存没超过 vm-max-memory 时，是不会交换任何 value 到磁盘上的。当超过最大内存限制后，redis 会选择较老的对象(如果两个对象一样老会优先交换比较大的对象)将它从内存中移除，这样会更加节约内存。

对于 Redis 来说，一个数据页面只会保存一个对象，也就是一个 Value 值，所以应该将 vm-page-size 设置成大多数 value 可以保存进去。如果设置太小，一个 value 对象就会占用几个数据页面，如果设置太大，就会造成页面空闲空间浪费。

VM 的工作机制

redis 的 VM 的工作机制分为两种：一种是 vm-max-threads=0，一种是 vm-max-threads > 0。

「第一种：vm-max-threads = 0」

• 数据换出：主线程定期检查使用的内存大小，如果发现内存超出最大上限，会直接以阻塞的方式，将选中的对象 换出 到磁盘上(保存到文件中)，并释放对象占用的内存，此过程会一直重复直到下面条件满足任意一条才结束：

  1. 内存使用降到最大限制以下。
  2. 设置的交换文件数量达到上限。
  3. 几乎全部的对象都被交换到磁盘了。
• 数据换入：当有 client 请求 key 对应的 value 已被换出到磁盘中时，主线程会以阻塞的方式从换出文件中加载对应的 value 对象，加载时此时会阻塞所有 client，然后再处理 client 的请求。「这种方式会阻塞所有的 client。」

「第二种：vm-max-threads > 0」

• 数据换出：当主线程检测到使用内存超过最大上限，会将选中的要交换的数据放到一个队列中交由工作线程后台处理，主线程会继续处理 client 请求。
• 数据换入：当有 client 请求 key 的对应的 value 已被换出到磁盘中时，主线程先阻塞当前 client，然后将加载对象的信息放到一个队列中，让工作线程去加载，此时进主线程继续处理其他 client 请求。加载完毕后工作线程通知主线程，主线程再执行被阻塞的 client 的命令。「这种方式只阻塞单个 client。」

总结：

Redis 直接自己构建了 VM 机制 ，不会像一般的系统会调用系统函数处理，会浪费一定的时间去 移动 和 请求，而 Redis 不存在。这也是 Redis 能够那么快的一个原因之一了。

多路 I/O 复用模型

多路I/O复用模型是利用 select、poll、epoll 可以同时监察多个流的 I/O 事件的能力，在空闲的时候，会把当前线程阻塞掉，当有一个或多个流有 I/O 事件时，就从阻塞态中唤醒，于是程序就会轮询一遍所有的流（epoll 是只轮询那些真正发出了事件的流），并且只依次顺序的处理就绪的流，这种做法就避免了大量的无用操作。

这里“多路”指的是多个网络连接，“复用”指的是复用同一个线程。采用多路 I/O 复用技术可以让单个线程高效的处理多个连接请求（尽量减少网络 IO 的时间消耗），且 Redis 在内存中操作数据的速度非常快，也就是说内存内的操作不会成为影响Redis性能的瓶颈，主要由以上几点造就了 Redis 具有很高的吞吐量。

为什么是单线程

官方FAQ表示，因为Redis是基于内存的操作，CPU不是Redis的瓶颈，Redis的瓶颈最有可能是机器内存的大小或者网络带宽。既然单线程容易实现，而且CPU不会成为瓶颈，那就顺理成章地采用单线程的方案了（毕竟采用多线程会有很多麻烦！线程间切换，并发问题，锁竞争等）。

注意：单线程，只是在处理我们的网络请求的时候只有一个线程来处理。但是一个正式的Redis Server运行的时候肯定是不止一个线程的。例如Redis进行持久化的时候会以子线程的方式执行。

注意：因为是单线程，耗时的查询，会导致读、写性能下降。解决方法，耗时的查询可以放到slave进行，组建 master-slave 形式。

大key问题

key设计建议

• 可读性和可管理性：以业务名(或数据库名)为前缀(防止key冲突)，用冒号分隔，比如业务名:表名:id o2o:order:1
• 简洁性：保证语义的前提下，控制key的长度，当key较多时，内存占用也不容忽视，例如： user:{uid}:friends:messages:{mid} 简化为 u:{uid}🇫🇷m:{mid}
• 不要包含特殊字符：反例：包含空格、换行、单双引号以及其他转义字符

value设计

• 拒绝bigkey(防止网卡流量、慢查询)

• 选择适合的数据类型：例如：实体类型要用map，而不是多个string。

• # 不推荐
  set user:1:name tom
  set user:1:age 19
  set user:1:favor football
  # 推荐，用下边替换
  hmset user:1 name tom age 19 favor football

• 控制key的生命周期，redis不是垃圾桶：建议使用expire设置过期时间(条件允许可以打散过期时间，防止集中过期)

如何定义bigKey 呢?

• 字符串类型：它的big体现在单个value值很大，一般认为超过10KB就是bigkey。
• 非字符串类型：哈希、列表、集合、有序集合，它们的big体现在元素个数太多。

一般来说，string类型控制在10KB以内，hash、list、set、zset元素个数不要超过5000。当然了这不是绝对的，请依据场景，灵活处理。

一般来说，bigkey的产生都是由于程序设计不当，或者对于数据规模预料不清楚造成的，来看几个例子：

• 社交类：粉丝列表，如果某些明星的粉丝数据，如果不精心设计下，一个明星的粉丝 百万很少了吧，你都把这百万的粉丝数据放到一个key中存储，毫无疑问是bigkey
• 统计类：比如按天存储某项功能或者网站的用户集合，用户很少，倒是没多大问题，一旦用户多了起来，必是bigkey
• 缓存类：将数据从数据库加载出来以后序列化放到Redis里，这个方式非常常用，但有两个地方需要注意，第一，是不是有必要把所有字段都缓存；第二，有没有相关关联的数据，不要为了图方便把相关数据都存一个key下，产生bigkey。

bigkey的危害

• 导致redis阻塞：这个也很好理解： 我们知道Redis 官方号称10万QPS， 我们通常达不到这个值，但是大几万的QPS还是没问题的，这也就意味着 redis 的执行速度 1秒几万条， 速度相当的快的。 假设你有个bigKey , 操作一次耗时1秒，那Redis 单线程 在这1秒钟就只能处理你这个Key, 后面堵了一堆请求。。。。 并且你的应用 序列化和反序列化这种大key , 也消耗CPU 。
• 导致网络拥塞:假设我们的交换机，千兆网络（小b），那么 实际带宽 1024 / 8 = 128M . 假设你的这个key的大小 500KB， 客户端并发 1000获取这个key， 那么就意味着 1000 * 500KB = 500M ,那就是每秒产生500M的流量。先不说你的Redis能不能处理的过来这个并发下的bigKey，单说你的这个千兆网络， 你说你这个网络I/O能扛得住吗？ 一般服务器会采用单机多实例的方式来部署，也就是说一个bigkey可能会对其他实例也造成影响，其后果不堪设想。
• 内存空间不均匀：比如在 Redis cluster 或者 codis 中，会造成节点的内存使用不均匀。
• 过期时可能阻塞 ：如果 Bigkey 设置了过期时间，当过期后，这个 key 会被删除，假如没有使用 Redis 4.0 的过期异步删除，就会存在阻塞 Redis 的可能性，并且慢查询中查不到（因为这个删除是内部循环事件）。但是有了 lazy del 之后，这个问题就没那么严重了。

如何发现大key

1. redis-cli --bigkeys命令可以统计bigkey的分布情况。对于 String 类型来说，会输出最大 bigkey 的字节长度，对于集合类型来说，会输出最大 bigkey 的元素个数。redis-cli -h xxxxxx.redis.rds.aliyuncs.com -a shiliName:password --bigkeys
2. 使用debug object key命令，从命令结果中的serializedlength的值来判断当前key的字节数
3. 使用strlen命令来判断当前key的长度
4. 使用scan命令+debug object结合起来，scan命令可以渐进的扫描所有的key值，分别计算每个key的serializedlength值，找到对应的bigkey进行相应的处理和报警。这个操作尽量在从库进行，因为debug object命令执行速度可能会比较慢，造成Redis的阻塞。
5. redis-rdb-tools工具。redis实例上执行bgsave，然后对dump出来的rdb文件进行分析，找到其中的大KEY

解决

• 拆分：可以考虑对 Bigkey 进行拆分，具体方法如下：对于 string 类型的 Bigkey，可以考虑拆分成多个 key-value。对于 hash 或者 list 类型，可以考虑拆分成多个 hash 或者 list。 big list： list1、list2、…listN big hash：可以讲数据分段存储，比如一个大的key，假设存了1百万的用户数据，可以拆分成 200个key，每个key下面存放5000个用户数据（拆分成小key）
• 控制集合的操作个数：如果bigkey不可避免，也要思考一下要不要每次把所有元素都取出来(例如有时候仅仅需要hmget，而不是hgetall)，删除也是一样，尽量使用优雅的方式来处理
• 选择适合的数据类型，（减小内存占用）
• 控制key的生命周期（减少内存占用）
• 大文档不用Redis：Redis对于长文本不是最优的，可考虑文档型数据库如：MongoDB等

不能直接删除。4.0之后可以lazy del。之前是 scan 逐渐删除。

topn的问题

基本操作

• 加入成员: zadd key score member
• 查询成员: zrange key start stop [withscores] 注: start = 0 end = -1 注: start/end都包含
• 删除成员: zrem key member
• 查看某个成员的 score: zscore key member
• 增加某个成员的score: zincrby key increment member
• 查看集合的size: zcard key
• 查看topN/倒叙: zrevrange key start stop [withscores] 注: start/end都包含

实际使用例子

图书销量排行榜：

# 添加
zadd books 1 Python编程
# 查询所有books成员=>带分数
zrange books 0 -1 withscores
# 删除books中成员: 'Python编程'
zrem books 'Python编程'
# 单独查看books成员score
zscore books 'Python编程'
# 增加books中成员排名score
zincrby books 20 Python编程
# 查看books成员总数
zcard books
# 查看books榜单top5
zrevrange books 0 4 withscores

热榜新闻：

# 添加新闻
zadd rank:202008 1 恭喜！潘玮柏晒全家照宣布结婚 
# 查看新闻总条数
zcard rank:202008 
# 用户点击新闻(8月)
zincrby rank:202008 1 TikTok正式起诉美国政府
# 查看新闻热点前三(8月)
zrevrange rank:202008 0 2

#  查看8月和9月共同的新闻交集(相同成员score会sum加起来)
zinterstore rank:20200809 2 rank:202008 rank:202009
zrange rank:20200809 0 -1
zrange rank:20200809 0 -1 withscores

# 查看8月和9月新闻的并集(相同成员score会sum加起来)
zunionstore rank:20200809:union 2 rank:202008 rank:202009
zrevrange rank:20200809:union 0 -1 withscores

并发key修改

解决分布式高并发修改同一个Key的问题 https://www.cnblogs.com/yy3b2007com/p/9383713.html

• 乐观锁：解决并发下修改同一个值覆盖的问题。

找热点key

如果 qps 过高，可以考虑通过 monitor 指令快速观察一下究竟是哪些 key 访问比较频繁，从而在相应的业务上进行优化，以减少 IO 次数。 monitor 指令会瞬间吐出来巨量的指令文本，所以一般在执行monitor 后立即 ctrl+c 中断输出。(热点key)

Redis主从怎么配置？

1.编辑配置文件Redis.conf

redis默认只允许本机连接，所以需要找到“bind 127.0.0.1”并将这行注释掉：

redis在3.0版本以后增加了保护模式 ,如需保护，改成yes

将默认的“daemonize no”改为yes，设置redis以守护线程方式启动：

分别配置pid，log，db文件的保存地址

启动redis

设置开机启动

2.Redis主从配置

从节点配置

(1) 修改redis从配置文件，添加一行配置“slaveof 192.168.0.101 6379”映射到主节点

(2) 重启从节点的redis

3.查看并验证主从配置

（1）主节点与从节点均登录redis并执行info命令查看主从配置结果

找到“# Replication”模块，可以看到主节点提示存在一个从节点，并且会列出从节点的相关信息，同样，可以在从节点看到自己的主节点是哪个，列出主节点的相关信息

（2）验证主从

登录主节点redis，set age 24，到从节点直接get age，看到可以get到我们在主节点设置的值24，说明主从配置成功

为啥用Redis？

看你简历上写了你项目里面用到了Redis，你们为啥用Redis？

因为传统的关系型数据库如Mysql已经不能适用所有的场景了，比如秒杀的库存扣减，APP首页的访问流量高峰等等，都很容易把数据库打崩，所以引入了缓存中间件，目前市面上比较常用的缓存中间件有Redis 和 Memcached 不过综合考虑了他们的优缺点，最后选择了Redis。

应用场景不一样：Redis出来作为NoSQL数据库使用外，还能用做消息队列、数据堆栈和数据缓存等；Memcached适合于缓存SQL语句、数据集、用户临时性数据、延迟查询数据和session等。

灾难恢复–memcache挂掉后，数据不可恢复; redis数据丢失后可以通过aof恢复

存储数据安全–memcache挂掉后，数据没了；redis可以定期保存到磁盘（持久化）

Redis优点和 Memcached 比较：

• 持久化，主从复制
• 支持数据类型丰富
• 支持脚本LUA
• 乐观锁：命令块和脚本的原子执行
• 密码验证
• Redis 部署简单，不需要依赖项

数据量很大，怎么做

一、增加内存

　　redis存储于内存中，数据太多，占用太多内存，那么增加内存就是最直接的方法，但是这个方法一般不采用，因为内存满了就加内存，满了就加，那代价也太大，相当于用钱解决问题，不首先考虑，一般所有方面都做到最优化，才考虑此方法

二、搭建Redis集群

​

(1)所有的redis节点彼此互联(PING-PONG机制),内部使用二进制协议优化传输速度和带宽.

(2)节点的fail（失败）是通过集群中超过半数的节点检测失效时才生效.

(3)客户端与redis节点直连,不需要中间proxy层.客户端不需要连接集群所有节点,连接集群中任何一个可用节点即可

(4)redis-cluster把所有的物理节点映射到[0-16383]slot上,cluster 负责维护node<->slot<->value

Redis集群中内置了 16384 个哈希槽，当需要在 Redis 集群中放置一个 key-value 时，redis 先对key 使用 crc16 算法算出一个结果，然后把结果对 16384 求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽，redis 会根据节点数量大致均等的将哈希槽映射到不同的节点

RDB、AOF、混合持久，我应该用哪一个？

一般来说， 如果想尽量保证数据安全性， 你应该同时使用 RDB 和 AOF 持久化功能，同时可以开启混合持久化。

如果你非常关心你的数据， 但仍然可以承受数分钟以内的数据丢失， 那么你可以只使用 RDB 持久化。

如果你的数据是可以丢失的，则可以关闭持久化功能，在这种情况下，Redis 的性能是最高的。

redis缓存一致性问题解决方案

更新DB和操作缓存两个动作之间，明显缺乏原子性，有可能更新DB完成，但是操作（淘汰或者更新）缓存失败，反之亦然。所以两者之间必然是有断层的，那么先选择操作谁才是最佳的方案？

推荐先更新DB，然后再更新或者淘汰缓存，原因如下

• 如果先更新缓存，然后更新DB的动作失败了，下一个读请求过来，读取到的缓存数据就是未曾更新到DB的数据，这样的数据明显是业务错误的。因此建议先更新DB，再后续选择操作缓存。（把DB失败的数据，放入缓存，肯定是不行的）
• 如果采用先淘汰缓存，后更新DB操作，如果在DB更新完成之前，来了一个新的读请求，那么就会查询出数据库的旧数据，缓存到redis，导致DB更新完成之后，两者数据不一致。（并发读，触发的加载，此时还没有改完DB）
• 如果先更新DB，然后操作缓存失败，客户端读取到的是旧的数据，此时也是存在DB与缓存不一致，但是实际业务上，我们更多的还是以DB数据为准，这种读取到旧数据的业务影响可能比读取到为未更新到DB的数据影响要小。

缓存不直接失效，而是设置过期时间，延迟失效。等到查询的时候，过期更新。

Redis 限流

zset 滑动窗口

Redis-Cell

Redis 4.0 提供了一个限流 Redis 模块，它叫 redis-cell。该模块也使用了漏斗算法，并提供了原子的限流指令。有了这个模块，限流问题就非常简单了。

该模块只有 1 条指令 cl.throttle，

使用场景

1. 记录帖子的点赞数、 评论数和点击数 (hash) （redis原子操作计数）

2. 记录用户的帖子 ID 列表 (排序)， 便于快速显示用户的帖子列表 (zset)。

3. 记录帖子的标题、 摘要、 作者和封面信息， 用于列表页展示 (hash)。

4. 记录帖子的点赞用户 ID 列表， 评论 ID 列表， 用于显示和去重计数 (zset)。

5. 缓存近期热帖内容 (帖子内容空间占用比较大)， 减少数据库压力 (hash)。

6. 记录帖子的相关文章 ID， 根据内容推荐相关帖子 (list)。

7. 如果帖子 ID 是整数自增的， 可以使用 Redis 来分配帖子 ID(计数器)。

8. 收藏集和帖子之间的关系 (zset)。

9. 记录热榜帖子 ID 列表， 总热榜和分类热榜 (zset)。

10. 缓存用户行为历史， 进行恶意行为过滤 (zset,hash)

• 计数器：数据统计的需求非常普遍，通过redis原子操作计数。例如，点赞数、收藏数、分享数等。

• 热点排行榜：排行榜按照得分进行排序，例如，展示最近、最热、点击率最高、活跃度最高等各种类型的top list。

• 好友列表：例如，用户点赞列表、用户收藏列表、用户关注列表等。

• 缓存：缓存热点数据，这也是redis最典型的应用之一，根据实际情况，缓存用户信息，缓存session等。

[catcatbai]

fasd