Redis

思维导图

前言

文章源码

NoSql

NoSQL（Not Only SQL）数据库主要指在设计上不遵循传统关系型数据库的模式。它包括了一系列不同类型的数据存储方式，不仅仅局限于简单的 key-value 存储模型。NoSQL 数据库的类型包括文档型、列存储、图形数据库等。

• 不遵循 SQL 标准。

• 大多不支持 ACID事务。

• 特定领域远超于 SQL 的性能。

适用场景

• 高并发读写处理能力
• 海量数据处理
• 数据的高可扩展性

不适用场景

• 需要强事务支持的场景： 对于那些需要严格的事务一致性和复杂事务管理的应用，NoSQL数据库可能不是最佳选择。虽然一些NoSQL数据库提供了一定程度的事务支持，但它们无法与传统的关系型数据库（如PostgreSQL或MySQL）相比，后者提供更全面、更复杂的事务管理能力，如ACID事务。
• 复杂SQL查询和结构化数据的场景： NoSQL数据库通常不支持复杂的SQL查询和那些需要高度结构化和关联数据的应用场景。对于需要执行复杂的连接操作、子查询、存储过程和视图等SQL特性的场景，传统的关系型数据库是更合适的选择，因为它们原生支持这些功能，能多更好地处理复杂的数据关系和结构化查询。

Redis是什么

Redis（Remote Dictionary Server，远程字典服务器）是一个完全开源且免费的、用C语言编写的基于内存的高性能键值（Key/Value）存储系统。遵循BSD协议，Redis不仅作为分布式内存数据库获得广泛应用，同时也是支持持久化的NoSQL数据库。由于其灵活的数据结构和高效的性能，Redis被广泛认为是当前最受欢迎的NoSQL数据库之一，并常被称作数据结构服务器。

Redis与其他KV系统相比具有以下独特特点：

• 数据持久化：Redis支持将内存中的数据持久化到磁盘中，通过RDB（快照）或AOF（追加文件）两种方式实现。这使得Redis在重启后能够重新加载并使用之前的数据。
• 丰富的数据类型：不同于普通的键值存储，Redis提供了多种数据结构，如列表（list）、集合（set）、有序集合（zset）、散列（hash）等，增强了其存储和处理数据的灵活性。
• 数据备份和主从复制：Redis支持数据备份，以及通过主从（master-slave）模式实现数据的复制。这增强了数据的可用性和容错能力，使Redis在分布式系统中更为可靠。

Redis能做什么

• 缓存系统：使用Redis作为数据缓存，减少数据库的读取压力，提高系统响应速度。这是Redis最常见的用途之一。
• 序列生成器：用于构建分布式系统下全局唯一的序列号。
• 分布式锁：在多个进程或服务间同步资源访问时，Redis可以用作分布式锁的实现工具。
• 分布式任务队列：在多个服务间分配任务，当没任务时线程阻塞。
• 排行榜/计分板：使用Redis的有序集合（Sorted Sets），可以方便地实现排行榜或计分板等功能。
• 实时计数器：Redis的原子操作特性使其适合于实现如网站访问量、在线用户数等实时计数功能。
• 地理空间数据处理：Redis的Geo类型支持地理空间数据的存储和查询，适用于地理位置服务。

Redis为什么这么快

高性能服务器并非必须依赖多线程：经常存在一个误区，认为高性能服务器一定需要通过多线程来实现。其实并不一定，首先需要明确，对于CPU、内存和硬盘的速度和工作机制有基本的了解是非常重要的，服务器的性能不仅仅取决于是否采用多线程。

多线程不一定比单线程更高效：另一个误解是多线程在所有情况下都比单线程更有效率。实际上，多线程编程涉及复杂性和额外的性能开销，如线程管理和上下文切换。而单线程模型，通过避免这些问题，在特定情况下可以提供更优的性能。

1. 内存存储：Redis将所有数据保存在内存中，内存访问速度远快于磁盘。这种内存中数据处理机制使得Redis能够提供极快的数据读写速度。
2. 单线程架构：虽然Redis是单线程的，但这恰恰使其避免了常见的多线程编程问题，如线程间的上下文切换和锁竞争。单线程也意味着在处理请求时几乎没有任何内存锁定的开销。
3. I/O多路复用：Redis使用非阻塞I/O多路复用技术（线程模型）。这意味着Redis可以在单个线程中同时处理多个网络连接，提高I/O操作的效率。

Redis单实例理论QPS为 8W (来源：阿里云redis 参考值)

五大基础数据类型

文档来源

key

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

数据库

序号	命令
SELECT index	切换数据库
DBSIZE	返回当前数据库的 key 的数量
FLUSHDB	清空当前数据库中的所有 key
FLUSHALL	清空整个 Redis 服务器的数据(删除所有数据库的所有 key )
SORT	返回或保存给定列表、集合、有序集合 key 中经过排序的元素

String

Redis中的字符串（String）类型是最基本且广泛使用的数据结构。在内部，这种类型的数据实际上是以 字节数组（byte array）的形式存储的。这意味着字符串在Redis中不仅可以存储文本，还可以存储任何形式的二进制数据，比如图片或序列化的对象。

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

数据结构

struct SDS<T> {
  T capacity; // 数组容量
  T len; // 数组长度
  byte flags; // 特殊标识位
  byte[] content; // 数组内容
}

String 的数据结构为简单动态字符串(Simple Dynamic String,缩写 SDS)。是可以修改的字符串，内部结构实现上类似于 Java的ArrayList。创建字符串时 len 和 capacity 一样长，不会多分配冗余空间。这是因为绝大多数场景下我们不会使用 append 操作来修改字符串。

当字符串较短时，len 和 capacity 字段可能使用较小的数据类型（如8位或16位整数）来存储，以节省内存，对于更长的字符串，这些字段会使用更大的数据类型（如32位或64位整数），因此SDS多种结构体。

字符串扩容策略

• 小于1MB的字符串扩容每次是翻倍。
• 超过1MB的字符串，每次扩容将增加1MB空间。
• 字符串最大长度限制为512MB。

存储方式

Redis对象，是Redis内部用于表示所有键值数据的基础数据结构。这个对象不仅用于字符串，还用于列表、集合、哈希表等所有Redis支持的数据类型。

struct RedisObject {
    int4 type; // 4bits
    int4 encoding; // 4bits
    int24 lru; // 24bits
    int32 refcount; // 4bytes
    void *ptr; // 8bytes，64-bit system
} robj;

• Emb（嵌入式）
  • String小于 44 字节。
  • 字符串数据直接存储在Redis对象的内部，避免了额外的内存分配。
• Raw（原始式）
  • Redis对象包含元数据。
  • 字符串数据（SDS）存储在对象之外的单独内存区域，需要额外的内存分配。

内存分配最多可以给 RedisObject 分配到64字节，去掉RedisObject本身元数据占用剩下45个字节，因为字符串是以 ‘\0’ 结尾，所以剩下 44 字节。

简单分布式锁

List

Redis 列表是简单的字符串列表，按照插入顺序排序。可以添加一个元素到列表的头部(左边) 或者尾部(右边)。

主要特点

1. 自然顺序：List中的元素按照插入的顺序排列。
2. 双端：可以在列表的头部或尾部添加或删除元素。
3. 元素可重复：List中的元素可以重复，即同一个值可以出现多次。
4. 自动删除：当列表没有元素自动删除。
5. 可用于实现队列和栈

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

数据结构

在列表元素较少的情况下会使用一块连续的内存存储，这个结构是 zipList，也即是压缩列表，它将所有的元素紧挨着一起存储，分配的是一块连续的内存。当数据量比较多的时候才会改成 quickList (快速链表)。

zipList 转 quickList

注意：这两个条件是可以修改的，在 redis.conf 中

list-max-ziplist-value 64 
list-max-ziplist-entries 512 

• 试图往列表新添加一个字符串值，且这个字符串的长度超过 server.list_max_ziplist_value （默认值为 64 ）
• ziplist 包含的节点超过 server.list_max_ziplist_entries （默认值为 512 ）

zipList

压缩列表是一块连续的内存空间，元素之间紧挨着存储，没有任何冗余空隙。

struct ziplist<T> {
    int32 zlbytes; // 整个压缩列表占用字节数
    int32 zltail_offset; // 最后一个元素距离压缩列表起始位置的偏移量，用于快速定位到最后一个节点
    int16 zllength; // 元素个数
    T[] entries; // 元素内容列表，挨个挨个紧凑存储
    int8 zlend; // 标志压缩列表的结束，值恒为 0xFF
}

struct entry {
    int<var> prevlen; // 前一个 entry 的字节长度
    int<var> encoding; // 元素类型编码
    optional byte[] content; // 元素内容
}

压缩列表为了支持双向遍历，所以才会有 ztail_offset 这个字段，用来快速定位到最后一个元素，然后倒着遍历。

Redis的压缩链表是一种内存高效的数据结构，主要用于存储较小的元素集合。它的特殊之处在于它不使用传统的链表结构，即没有为每个元素维护前驱和后继指针（prev和next）。相反，它通过存储每个元素（entry）的长度来定位元素，这样可以减少所需的存储空间。

原理

• 访问下一个元素，程序简单地将当前元素的长度加到当前元素的指针上。
• 访问前一个元素，程序则将前一个元素的长度从当前元素的指针上减去。

在集合元素较少时，通过元素长度定位元素，避免存储前驱和后继指针带来的内存开销，是一种 时间换空间 的做法。

quickList

• 快速链表是由多个压缩列表（ziplists）组成的双向链表。
• 每个节点（ziplist）包含了列表的一部分元素。

为什么从压缩链表转向快速链表

• 提高性能：大的压缩链表在扩容时效率更低。

• 灵活性：拆分小的压缩列表在数据插入和删除时更具备灵活性。

• 减少了对大量连续内存的需求。

struct ziplist {
    ...
}
struct ziplist_compressed {
    int32 size;
    byte[] compressed_data;
}
struct quicklistNode {
    quicklistNode* prev;
    quicklistNode* next;
    ziplist* zl; // 指向压缩列表
    int32 size; // ziplist 的字节总数
    int16 count; // ziplist 中的元素数量
    int2 encoding; // 存储形式 2bit，原生字节数组还是 LZF 压缩存储
    ...
}
struct quicklist {
    quicklistNode* head;
    quicklistNode* tail;
    long count; // 元素总数
    int nodes; // ziplist 节点的个数
    int compressDepth; // LZF 算法压缩深度
    ...
}

快速链表插入、查找操作

为了进一步节约空间，Redis 还会对 ziplist 进行压缩存储，使用 LZF 算法压缩，可以选择压缩深度。

任务队列实现

Set

Set 对外提供的功能与 List 类似列表的功能，特殊之处在于 Set 是可以 自动排重 的，当需要存储一个列表数据，又不希望出现重复数据时，Set 是一个很好的选择，并且 Set 提供了判断某个成员是否在一个 Set 集合内的重要接口，这个也是 List 所不能提供的。

• 随机获取
• 去重
• 无序
• 交集
• 并集
• 差集

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

数据结构

Redis的 set 底层使用了 intset 和 hashtable 两种数据结构存储。

Intset（整数集合）

使用条件：

• 当所有元素都是 整数 且集合元素少于 512，Redis使用intset。
• intset本质上就是一个数组，用于高效地存储整数值。

特点：

• 内存效率：对于小整数集合，intset非常节省内存。
• 性能：操作intset通常比操作哈希表快，尤其是在集合较小的情况下。
• 插入元素时，通过二分查找法确保元素唯一。

intset

intset是一个由 整数 组成的 有序 集合，从而便于在上面进行 二分查找，用于快速地判断一个元素是否属于这个集合。主要就是针对整数型的小集合进行性能优化。

typedef struct intset {
	uint32_t encoding; // 编码类型 int_16t、int_32t、int_64t
	uint32_t length;   // 元素数量
	int8_t contents[]; // 元素数组
}intset;

#define INTSET_ENC_INT16 (sizeof(int16_t))   //16位，2个字节，表示范围-32,768~32,767
#define INTSET_ENC_INT32 (sizeof(int32_t))   //32位，4个字节，表示范围-2,147,483,648~2,147,483,647
#define INTSET_ENC_INT64 (sizeof(int64_t))   //64位，8个字节，表示范围-9,223,372,036,854,775,808~9,223,372,036,854,775,807

intset 升级与降级
比如一开始set存储的是int16_t类型的数据, 但是当我们添加了一个int32_t类项的数据时，就需要操作升级。

根据新元素的类型, 扩展底层元素的空间, 并为新元素分配空间将现有的元素都转为新的元素类型, 并存储在正确的空间上面将新元素添加进数组内，不支持降级。

为什么不使用 zipList

Set需要对自身元素进行去重，HashTable和数组（二分查找）能提供更高效的排重性能。zipList的优势是少量集合内存使用效率高。

Dict 字典

typedef struct dict {
	// 指向 dictType 结构的指针（dictType 结构保存的是操作特定类型键值对的函数）
    dictType *type;
    // 保存上述所说函数的参数
    void *privdata;
    // 哈希表，2个hash table，一个新的与一个旧的
    dictht ht[2];
    // rehash 索引，rehash 不进行时值为 -1 
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    // 正在运行的迭代器数量
    int iterators; /* number of iterators currently running */
} dict;

类似 Java 的HashMap，存储使用到Key，Value为空。

特点：

• 灵活性：可以存储任意类型的元素，包括字符串和复杂对象。
• 扩展性：适用于较大的集合，哈希表的性能优于 intset。

文章关注共同好友实现(取交集)

## 文章关注人
sadd article zhangsan lisi wangwu luliu chenqi

## 我的好友
sadd friend zhangsan wangwu

## 文章共同关注好友
sinter article friend

抽奖系统

## 初始化奖品  1 一等奖，2 二等奖，3 三等奖，<3 谢谢惠顾
sadd lottery 1 2 3 4 5 6 7 8 9 10

## 开始抽奖
srandmember lottery

Hash

特点：Hash 是一个string类型的field和value的映射表，hash特别适合用于存储对象。

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

数据结构

• zpiList

  • 元素个数小于hash-max-ziplist-entries 配置（默认 512 个）
  • 所有值都小于hash-max-ziplist-value 配置（默认 64 字节）

• hashTable

  元素过多或者value太长影响zipList性能时切换数据结构

zipList

• 存储格式：在 ziplist 中，Hash 被存储为字段和值的连续序列。首先是字段名，紧接着是对应的值，然后是下一个字段名，以此类推。
• 优化：通过这种方式，ziplist 能够在保持良好读写性能的同时，显著减少内存使用。
• 自动转换：当 ziplist 中的数据超过配置的阈值时，Redis 会自动将其转换为一个更标准的哈希表结构，以维持性能。

Dict 字典

typedef struct dict {
	// 指向 dictType 结构的指针（dictType 结构保存的是操作特定类型键值对的函数）
    dictType *type;
    // 保存上述所说函数的参数
    void *privdata;
    // 哈希表，2个hash表，一个新的与一个旧的
    dictht ht[2];
    // rehash 索引，rehash 不进行时值为 -1 
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    // 正在运行的迭代器数量
    int iterators; /* number of iterators currently running */
} dict;

/*
 * 哈希表结构
 * 每个字典都使用两个哈希表，从而实现渐进式 rehash 。
 */
typedef struct dictht {
	// 哈希表数组
    dictEntry **table;
    // 哈希表的大小
    unsigned long size;
    // 哈希表的大小掩码（总是 size-1）
    unsigned long sizemask;
	// 哈希表当前的已有结点数量
	unsigned long used;
} dictht;

typedef struct dictEntry {
	// 键
    void *key;
    // 值
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    // 指向下一个哈希表结点的指针
    struct dictEntry *next;
} dictEntry;

当 哈希类型 无法满足 ziplist 的条件时，Redis 会使用 hashtable 作为 哈希 的 内部实现，因为此时 ziplist 的 读写效率 会下降。

维护用户信息

hmset wgf name wgf age 18 address shenzhen

# 获取用户信息
hget wgf

# 维护年龄
hincrby wgf age 1

# 维护地址
hset wgf address guangzhou

Sorted Set（Z set）

• 成员唯一：有序集合也是由字符串类型的元素组成的集合，其中每个成员都是唯一的，不允许重复。
• 关联双精度分数：有序集合中的每个元素都会关联一个 double 类型的分数，作为排序的依据。
• 按分数排序：Redis 通过这个分数来对集合中的成员进行从小到大的排序。
• 分数可重复，成员不重复：尽管有序集合的成员是唯一的，但是不同的成员可以有相同的分数。

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

数据结构

ziplist

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

• 有序集合保存的元素数量小于128个
• 有序集合保存的所有元素的长度小于64字节

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

skiplist + dict

• 双重数据结构的优势：
  • 跳跃表：用于维护元素的有序性。跳跃表优势在于它提供了快速的顺序访问，特别适用于执行范围查询和迭代操作。
  • 字典：用于高效的元素查找。字典以成员作为键，分数作为值，使得对特定元素的访问、更新或删除操作变得极为迅速。
• 空间效率和内存优化：
  • 成员（Member）：成员由 redisObject 结构表示的字符串，dict 和 skipList 通过指针引用同一个成员。
  • 分数（Score）：在跳跃表中，每个元素的分数是直接存储在每个节点中的；在字典中，元素的成员作为键，其对应的分数作为值。

skiplist详解

skiplist

排行榜实现

## 初始化日销售排行榜
zadd rank  98789 wgf  23476 zs  48739 ls 28374 ww

## 更新日销量
zadd rank 99746 wgf

## 获取日销量第一名
zrevrange rank 0 0

## 获取销量榜前三名及销售额
 zrevrange rank 0 2 withscores

Redis 内部数据结构

redis 内部数据结构详解

三大特殊类型

参考文档

GEO

特点：GEO 是 Redis 3.2 版本中新增的功能，专门用于存储和操作地理位置信息。这一功能允许用户高效地处理和查询地理空间数据，适用于多种地理位置相关的应用场景。

序号	描述及命令
1	GEOADD key longitude latitude member [longitude latitude member ...] geoadd 用于存储指定的地理空间位置，可以将一个或多个经度(longitude)、纬度(latitude)、位置名称(member)添加到指定的 key 中
2	GEOPOS key member [member ...] 用于从给定的 key 里返回所有指定名称(member)的位置（经度和纬度），不存在的返回 nil
3	GEODIST key member1 member2 [m|km|ft|mi] 用于返回两个给定位置之间的距离 m: 米 km:千米 mi:英里 ft:英尺
4	GEORADIUS key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key] 以给定的经纬度为中心， 返回键包含的位置元素当中， 与中心的距离不超过给定最大距离的所有位置元素
5	GEORADIUSBYMEMBER key member radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key] 和 GEORADIUS 命令一样， 都可以找出位于指定范围内的元素， 但是 georadiusbymember 的中心点是由给定的位置元素决定的， 而不是使用经度和纬度来决定中心点
6	GEOHASH key member [member ...] GEO 使用 geohash 来保存地理位置的坐标

数据结构

底层采用 Zset 数据结构，Menber存储值，Score存储经纬度

使用

## 初始化城市
geoadd city 114.100 22.200 xianggang
geoadd city 113.233 23.166 guangzhou
geoadd city 113.516 22.300 zhuhai
geoadd city 114.066 22.616 shenzhen

## 获取深圳的经纬度
geopos city shenzhen

## 获取深圳和广州的直线距离
geodist city shenzhen guangzhou km

## 以给定的经纬度为中心，获取指定范围内的城市
georadius city 114.100 22.200 150 km

## 获取深圳100KM内的城市
GEORADIUSBYMEMBER city shenzhen 100 km

SpringBoot

@Slf4j
@SpringBootTest
public class GeoTest {
    @Autowired
    private RedisTemplate redisTemplate;

    private static final String KEY = "city";

    @Test
    public void geoAddTest() {
        redisTemplate.opsForGeo().add(KEY, new Point(114.100d, 22.200d), "xianggang");
        redisTemplate.opsForGeo().add(KEY, new Point(113.233, 23.166d), "guangzhou");
        redisTemplate.opsForGeo().add(KEY, new Point(113.516d, 22.300d), "zhuhai");
        redisTemplate.opsForGeo().add(KEY, new Point(114.066d, 22.616d), "shenzhen");
    }


    @Test
    public void geoPosTest() {
        List<Point> shenzhen = redisTemplate.opsForGeo().position(KEY, "shenzhen");
        log.info("x: {}, y: {}", shenzhen.get(0).getX(), shenzhen.get(0).getY());
    }


    @Test
    public void geoDistTest() {
        Distance distance = redisTemplate.opsForGeo().distance(KEY, "shenzhen", "guangzhou", Metrics.KILOMETERS);
        log.info("value: {}", distance.getValue());
    }


    @Test
    public void geoRadiusTest() {
        Point point = new Point(114.100d, 22.200d);
        Metrics kilometers = Metrics.KILOMETERS;
        Distance distance = new Distance(150, kilometers);
        Circle circle = new Circle(point, distance);

        GeoResults<RedisGeoCommands.GeoLocation> radius = redisTemplate.opsForGeo().radius(KEY, circle);
        radius.forEach(geo -> {
            log.info("name: {}", geo.getContent().getName());
        });
    }


    @Test
    public void geoRadiusByMemberTest() {
        GeoResults<RedisGeoCommands.GeoLocation> results = redisTemplate.opsForGeo().radius(KEY, "shenzhen", new Distance(100, Metrics.KILOMETERS));
        results.forEach(geo -> {
            log.info("name: {}", geo.getContent().getName());
        });
    }
}

使用场景

• 位置服务

  如地点搜索、附近的人/商家等功能

• 地理围栏

  判断是否在某个经纬度的半径范围内

• 距离计算

  计算两个地点之间的直线距离

HyperLogLog

HyperLogLog 是一种用于高效进行基数统计的算法，特别适合处理大量数据元素。它的主要特点和应用场景如下：

1. 处理重复元素

   ​ HyperLogLog 算法在统计基数时，同一元素的多次出现只被计算一次。这意味着重复的元素对基数的贡献只会被累加一次，确保统计的是独一无二的元素数量。

2. 基数估计而非值存储

   ​ HyperLogLog 专注于统计不同元素的数量（即基数），而不存储这些元素本身。因此，它无法像集合那样返回输入的各个元素。

3. 空间效率

   ​ HyperLogLog 的一个显著优点是其内存效率。即使输入元素的数量或体积非常大，HyperLogLog 所需的存储空间总是固定且相对较小。每个 HyperLogLog 键仅需大约 12 KB 的内存，就可以计算接近 2^64 个不同元素的基数。

4. 误差率

   ​ 在进行基数估计时，HyperLogLog 存在一定的误差。标准误差率约为 0.81%，这对于大多数应用场景来说是一个可接受的误差范围。

5. 应用场景

   ​ HyperLogLog 的主要应用场景包括统计大量数据的独特元素数。例如，在网站分析中，它被用于估算页面浏览量（PV）或独立访客数（UV）。

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

数据结构

TyperLogLog 详解

1. 存储结构

   HyperLogLog 在 Redis 中确实采用 String 类型进行存储。它通过一种特殊的编码方式，高效地维护统计信息。

2. 桶（Bucket）的使用

   HyperLogLog 内部维护了 16384（即 2^14）个桶。这些桶并不直接记录各自桶的元素数量，而是记录与元素散列值相关的特定统计信息。

3. 元素的处理

   当一个元素被添加到 HyperLogLog 中时，首先计算其散列值。这个散列值决定了元素被分配到哪个桶，并影响该桶的统计值。具体来说，散列值的一部分用于确定桶的编号，另一部分则用于计算并更新桶的统计信息。

4. 概率算法

   HyperLogLog 利用概率算法来估计基数。由于是基于概率的方法，每个桶的统计值并不代表精确计数，而是一个概率估计值。

5. 基数的估算

   通过将所有桶的计数值进行数学处理（如调和均值计算），HyperLogLog 能够估算出接近真实的总基数。虽然单个桶的统计值可能不够精确，但综合所有桶的信息后，得到的基数估算通常非常接近真实值。

计算页面UV

## 模拟用户访问
pfadd pv 192.168.1.1
pfadd pv 192.168.1.2
pfadd pv 192.168.1.1
pfadd pv 192.168.1.3
pfadd pv 192.168.1.3
pfadd pv 192.168.1.4

## 获取当前UV
pfcount pv

返回 4

SpringBoot

@Slf4j
@SpringBootTest
public class HyperLogLogTest {
    @Autowired
    RedisTemplate redisTemplate;

    private static String KEY = "pv";

    @Test
    public void pfAddTest() {
        for (int i = 0; i < 1000; i++) {
            redisTemplate.opsForHyperLogLog().add(KEY, i);
        }
    }

    @Test
    public void pfCountTest() {
        log.info("count: {}", redisTemplate.opsForHyperLogLog().size(KEY));
    }
}

使用场景

• 独立访客计数

  计算网站的独立访客量，比如通过ip维度

• 流量统计

  估算一定时间内网页或API接口的独立请求次数

• 去重统计

  处理大量数据时，移除或统计重复项

BitMap

BitMap 是以二进制位（bit）为单位的数组，其中每个位可以是 0 或 1。在 Redis 中，BitMap 实际上是存储在字符串值中的位序列。

序号	命令及描述
1	SETBIT key offset value 添加元素到BitMap
2	GETBIT key offset 根据偏移量获取BitMap上的元素
3	BITCOUNT key [start] [end] 计算给定字符串中，被设置为 1 的比特位的数量
4	BITPOS key bit [start] [end] 返回位图中第一个值为 bit 的二进制位的位置
5	BITOP operation destkey key [key …] 对一个或多个保存二进制位的字符串 key 进行位元操作，并将结果保存到 destkey 上
6	BITFIELD key [GET type offset] [SET type offset value] [INCRBY type offset increment] [OVERFLOW WRAP|SAT|FAIL]

数据结构

底层使用 String 数据类型，采用 offset 和 bit 存储数据，bit映射被限制在512MB之内，所以最大是2^32。

注意：假设初始的offset非常大，BitMap的初始化会非常慢，假设offset值为 2^32-1=4294967295，由于 redis 没有采用压缩实现，就会直接申请到 512MB 内存空间来存储 2^32-1 bit 位置的值 1，中间的 bit 也会全填上 0（内存暴涨），在偏移量较大的场景可以参考 Kafka 相对偏移量的做法。

网站记录一周用户签到情况

## 用户一周签到情况, 1代表签到
setbit wgf 0 1
setbit wgf 3 1
setbit wgf 5 1
setbit wgf 6 1

## 获取一周用户签到次数
bitcount wgf

## 获取周二用户是否签到
getbit wgf 1

## 获取用户最早没有签到的日期
bitpos wgf 0

SpringBoot

@Slf4j
@SpringBootTest
public class BitMapTest {
    @Autowired
    private RedisTemplate redisTemplate;


    @Test
    public void setBitTest() {
        redisTemplate.opsForValue().setBit("wgf", 0 ,true);
        redisTemplate.opsForValue().setBit("wgf", 3 ,true);
        redisTemplate.opsForValue().setBit("wgf", 5 ,true);
        redisTemplate.opsForValue().setBit("wgf",6 ,true);
    }


    @Test
    public void getBitTest() {
        log.info("value: {}", redisTemplate.opsForValue().getBit("wgf", 0));
    }


    @Test
    public void bitCountTest() {
        Long num = (Long) redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.bitCount("wgf".getBytes());
            }
        });

        log.info("num: {}", num);
    }

    
    @Test
    public void bitPosTest() {
        Long value = (Long) redisTemplate.execute((RedisCallback<Long>) connection -> connection.bitPos("wgf".getBytes(), false));
        log.info("value: {}", value);
    }
}

使用场景

• 用户签到功能

• 状态跟踪

  通过唯一id跟踪用户上下线状态

• 数据去重

• MQ消息队列幂等

• 分布式布隆过滤器

事务

• 本质：Redis事务通过将一系列命令放入一个队列中，并在EXEC命令发出时一次性、按顺序执行这些命令。这个过程确保了事务内的命令执行时不会被其他命令打断。

• Redis事务的原子性：在Redis中，事务提供了一种有限的原子性。如果事务中的命令无法执行（如因为语法错误），则整个事务将失败。然而，如果事务中的命令在运行时出错（例如，因数据类型不匹配），之前的命令仍然会被执行。

• 隔离级别：Redis事务没有传统意义上的隔离级别概念。Redis的事务是通过单线程的特性来实现隔离的，即在执行事务中的命令序列时，不会有其他命令插队执行。

Redis 事务可以一次执行多个命令， 并且带有以下三个重要的保证：

• 事务操作被放入一个队列中，并在事务提交时（使用EXEC命令）一起执行。
• 如果某个命令执行失败（语法性错误），其他命令仍然会继续执行。
• 在事务执行过程，其他客户端提交的命令请求不会插入到事务执行命令序列中。

事务的三个阶段

一个事务从开始到执行会经历以下三个阶段：

• 开始事务
• 命令入队
• 执行 / 取消事务

命令

序号	命令及描述
1	DISCARD 取消事务，放弃执行事务块内的所有命令。
2	EXEC 执行所有事务块内的命令。
3	MULTI 标记一个事务块的开始。
4	UNWATCH 取消 WATCH 命令对所有 key 的监视。
5	WATCH key [key ...] 监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

事务使用

正常提交事务

## 设置库存
SET stock 100

## 设置订单数量
SET order 0

## 开启事务模拟下单
MULTI
DECRBY stock 10
INCRBY order 10
EXEC

## 提交事务返回结果
90
10

取消事务

## 设置库存
SET stock 100

## 设置订单数量
SET order 0

## 开启事务模拟下单
MULTI
DECRBY stock 10
INCRBY order 10

## 取消事务
DISCARD

## 查看数据
GET stock
100

GET order
0

命令性错误

不存在命令 命令参数不正确

若在事务队列中存在命令性错误（类似于java编译性错误），则执行EXEC命令时，所有命令都不会执行

## 设置库存
SET stock 100

## 开启事务
MULTI
DECRBY stock 10
GETSET stock 	// 命令性错误（缺少参数，或执行非Redis命令）
ERR wrong number of arguments for 'getset' command
EXEC

## 再次获取库存
GET stock
100

语法性错误

命令使用正确，但是语法使用不正确

若在事务队列中存在语法性错误（类似于java的1/0的运行时异常），则执行EXEC命令时，其他正确命令会被执行，错误命令抛出异常

## 设置库存
SET stock 100
SET flag test

## 开启事务
MULTI
DECRBY stock 10
INCRBY flag 10	// 语法性错误，字符串不能参与数字运算（命令没有错）
DECRBY stock 20
EXEC

GET stock
70				// 事务中语法正常的命令被执行

GET flag		
test			// 事务中正确 命令会被执行，错误命令抛出异常

WATCH 监控key

1. 乐观锁定：
   • WATCH命令用于监视一个或多个键，以检测这些键在事务执行之前是否被其他命令修改。
   • 如果在执行事务的EXEC命令之前，任何被WATCH监视的键被修改（包括更新、删除或过期），那么事务不会执行。
2. 使用场景：
   • WATCH命令通常用于复杂的事务场景，其中需要根据被监视键的值来决定是否执行事务。
   • 它特别适用于需要避免竞态条件的场景，例如，在多个客户端同时修改同一个键的值时。
3. 事务流程：
   • 一个典型的使用WATCH命令的事务流程包括：首先用WATCH命令监视一个或多个键，然后读取这些键的值，接着开始一个事务（使用MULTI命令），执行一系列命令，最后提交事务（使用EXEC命令）。
   • 如果监视的键在执行EXEC之前被修改，事务将被取消，EXEC命令返回一个空回复以指示事务未执行。
4. 取消监视：
   • 使用UNWATCH命令可以取消所有被WATCH命令设置的监视。
   • 另外，执行EXEC命令后，无论事务是否成功执行，所有的监视都会自动被取消。

使用

############# 窗口1 ##############
## 设置库存
SET stock 100

## 开启库存监控
WATCH stock

## 开启事务
MULTI

## 下单
DECRBY stock 1




############# 窗口2 ##############
## 获取库存
GET stock
"100"

## 下单
DECRBY stock 10




############# 窗口1 ##############
## 提交事务
EXEC
null	## 放弃事务执行

GET stock
90

## 重试
WATCH stock
MULTI
DECRBY stock 1
EXEC
89		## 事务执行成功，库存扣减

秒杀实现

StockController

这里的秒杀只是用于实践 事务 和 监控 的用法，实际上秒杀使用 Lua 为最优解，因为简单使用 decrby 会存在库存遗留问题

@Slf4j
@RestController
public class StockController {
 @Autowired
 private RedisTemplate redisTemplate;
 
 private ExecutorService executorService = Executors.newFixedThreadPool(100);
 
 @GetMapping("decr2")
 public void decr2() {
     Runnable task = () -> {
         Boolean flag = decrStock("stock", getNum(), 100);
         if (flag) {
             log.info("下单成功");
         } else {
             log.info("下单失败");
         }
     };

     for (int i = 0; i < 3000; i++) {
         executorService.execute(task);
     }
 }
 
 
 /**
     * 秒杀扣减库存实现
     * @param key 库存key
     * @param num 购买数量
     * @param tryCount 重试扣减次数
     * @return
     */
    private Boolean decrStock(String key, int num, int tryCount) {
        Boolean result = (Boolean) this.redisTemplate.execute(new SessionCallback<Boolean>() {
            @Override
            public Boolean execute(RedisOperations operations) throws DataAccessException {
                for (int i = 0; i < tryCount; i++) {
                    operations.watch(key);
                    Integer stockNum = (Integer) operations.opsForValue().get(key);
                    if (stockNum < num) {
                        // log.warn("库存不足");
                        operations.unwatch();
                        try {
                            TimeUnit.MILLISECONDS.sleep(30);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        continue;
                    }

                    operations.multi();
                    operations.opsForValue().decrement(key, num);

                    List<Object> result = null;
                    result = operations.exec();

                    if (Objects.nonNull(result) && result.size() > 0) {
                        //log.info("剩余库存：{}", result.get(0));
                        return true;
                    }
                }
                return false;
            }
        });

        return result;
    }
}

发布订阅

Redis的发布/订阅（Pub/Sub）是一种消息通信模式，允许客户端之间通过频道进行消息传递。这个模式包含发布者（发送消息的客户端）和订阅者（接收消息的客户端）。在Redis中，发布/订阅功能被广泛应用于构建消息队列、聊天系统、实时通知等场景。以下是其主要特点和工作方式：

1. 频道（Channels）：
   • 在Redis的Pub/Sub系统中，消息是通过频道发送的。频道可以被视为消息传递的媒介。
   • 客户端可以订阅任何频道。当消息发送到频道时，这些消息会被转发给所有订阅了该频道的客户端。
2. 发布（Publish）：
   • 发布者使用PUBLISH命令发送消息到指定的频道。
   • 当一个消息被发布到频道，所有订阅该频道的客户端都会收到这个消息。
3. 订阅（Subscribe）：
   • 客户端使用SUBSCRIBE命令订阅一个或多个频道。
   • 一旦订阅了频道，客户端会进入一个订阅状态，等待接收任何发送到这些频道的消息。
4. 模式订阅（Pattern Subscribe）：
   • Redis还提供了模式订阅功能，允许客户端订阅符合特定模式的频道。这通过PSUBSCRIBE命令实现。
   • 比如，使用PSUBSCRIBE news.*可以订阅所有以news.开头的频道。
5. 取消订阅：
   • 客户端可以使用UNSUBSCRIBE命令来取消订阅特定的频道。
   • 类似地，PUNSUBSCRIBE命令用于取消模式订阅。
6. 应用场景：
   • Redis的Pub/Sub系统非常适合构建实时消息系统，例如实时聊天应用、实时广播更新、实时在线监控系统等。
7. 限制：
   • Redis的发布/订阅模型不保证消息的持久化或可靠性。如果在消息发布时没有订阅者在线，这些消息将会丢失。
   • 它也不提供复杂的消息队列功能，如消息确认或持久订阅。

Redis的发布/订阅功能由于其简单性和高效性，在需要快速通信和实时更新的应用中非常有用。但是，在需要消息持久化和高可靠性的场景中，可能需要考虑其他消息队列解决方案。

订阅/发布消息图：

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

命令

############ 窗口1 ##############
## 订阅频道
SUBSCRIBE mychannel
1) "message"			# 类型
2) "mychannel"			# 频道
3) "hello!				# 窗口2发送消息这里会实时监控到

############ 窗口2 ##############
## 发送消息
PUBLISH mychannel "hello!"
1							# 返回当前接收消息客户端数量

############ 窗口1 ##############
## 取消订阅频道
UNSUBSCRIBE mychannel

Lua脚本

Lua 详解

Lua是一种轻量级的编程语言，被设计为嵌入到应用程序中，Redis从2.6版本开始内嵌了Lua解释器，允许执行Lua脚本。

优点

• 原子性执行：Lua脚本在Redis中以原子方式执行。这意味着在脚本执行期间，不会有其他Redis命令被执行，保证了操作的一致性和完整性。
• 减少网络开销：通过在服务器端执行脚本，可以减少在客户端和Redis服务器之间的往返通信，降低网络延迟。
• 操作封装：Lua脚本可以封装一系列复杂的Redis命令，简化客户端代码，使其更加整洁和易于维护。

缺点

• 性能影响：虽然Lua脚本在Redis中执行效率很高，但是过于复杂的脚本可能会长时间占用CPU，影响Redis服务器的性能。
• 安全性考虑：需要确保编写的Lua脚本安全可靠，特别是在处理外部输入数据时，避免执行恶意代码或操作。

EVAL 命令

EVAL script numkeys [key [key ...]] [arg [arg ...]]

• EVAL：这是Redis执行Lua脚本的命令。

• script：这是要执行的Lua脚本的文本。它是一段Lua代码，可以包含任何有效的Lua命令和Redis命令。

• numkeys：这个参数指定了随后在命令中列出的键(key)的数量。这个数字告诉Redis，接下来的多少参数应该被视为键名。

• [key [key ...]]：这部分是可选的，包含了将要被脚本处理的键名。这些键名放在Lua脚本中的KEYS数组里。numkeys参数指定了这里有多少个键。

• [arg [arg ...]]：这部分也是可选的，包含了传递给脚本的其他参数，这些参数放在Lua脚本中的ARGV数组里。

EVAL "return redis.call('set', KEYS[1], ARGV[1])" 1 mykey myvalue
OK

这段脚本说明有一个KEYS参数 mykey，有一个ARGV参与 myvalue。然后Lua脚本使用 redis.call 调用 redis 命令，最终将执行结果返回。

EVALSHA 脚本缓存语义

EVALSHA 是 Redis 中用于执行 Lua 脚本的命令，它与 EVAL 命令类似，但有一个关键的区别：它使用 Lua 脚本的 SHA1 校验码来引用被Redis加载的脚本，而不是直接提供脚本代码。这样做的优点是，一旦脚本被加载到 Redis 中，就可以通过其 SHA1 校验码高效地多次调用，而无需重新发送整个脚本代码。

SCRIPT LOAD script

用于将 Lua 脚本加载到 Redis 内存缓存并返回一个 sha1 码，Redis重启缓存脚本失效。

EVALSHA sha1 numkeys key [key ...] arg [arg ...] 

• sha1：这是预先加载到 Redis 中的 Lua 脚本的 SHA1 校验码。这个校验码是使用 SCRIPT LOAD 命令加载脚本时得到的。

• numkeys：这个参数指定了随后在命令中列出的 KEY 的数量。这个数字告诉 Redis，接下来的多少参数应该被视为键名。

• [key [key ...]]：这部分是可选的，包含了将要被脚本处理的键名。这些键名放在 Lua 脚本中的 KEYS 数组里。numkeys 参数指定了这里有多少个键。

• [arg [arg ...]]：这部分也是可选的，包含了传递给脚本的其他参数，这些参数放在 Lua 脚本中的 ARGV 数组里。这些参数可以是任何值，包括字符串、数字等，由脚本的逻辑来决定它们的用途。

使用 EVALSHA 命令执行 Lua 脚本

> SCRIPT LOAD "return 'Hello World'"			# 让Redis缓存一段脚本，但不执行，无限期缓存
470877a599ac74fbfda41caa908de682c5fc7d4b		# Redis返回的SHA1码

> EVALSHA "470877a599ac74fbfda41caa908de682c5fc7d4b" 0	# 执行缓存脚本

好处：相对于EVAL，它的第一个参数是一段脚本。EVALSHA 在执行Lua脚本可以使用缓存脚本，减少额外的网络开销。

Lua调用Redis命令

• redis.call()

  如果执行的命令出现错误，redis.call() 会停止脚本的执行，并将错误返回给脚本的调用者，也就是说错误要在脚本之外处理。

• redis.pcall()

  与 redis.call() 不同的是，如果命令执行失败，redis.pcall() 不会停止脚本执行或抛出错误，而是返回一个包含错误信息的Lua表，可以在脚本内处理错误。

set k v
eval "return redis.call('incr', KEYS[1])" 1 k
(error) ERR Error running script (call to f_2bab3b661081db58bd2341920e0ba7cf5dc77b25): @user_script:1: ERR value is not an integer or out of range


-- 实际脚本要合并为一行才能执行，脚本内部错误处理
eval "
  local result = redis.pcall('incr', KEYS[1]) 
  if result.err then
    return 0
  else
    return result
  end    
" 1 k
(integer) 0

Redis 和 Lua 的数据关系

在Lua 5.2及之前的版本中，Lua只有一种数字类型，即双精度浮点数。当需要将数字存储到Redis并期望它们作为整数处理时（例如用于INCR、DECR或其他需要整数的Redis命令），需要确保这些数字在转换前是整数，可以使用math.floor保证。

在Lua 5.3及以后的版本中，Lua支持整数和浮点数。在这些版本中，Lua会根据数字的使用方式自动选择合适的类型。

Redis Lua 版本默认为5.1

Lua 数据类型	描述	转换为 Redis 数据类型	描述	示例（Lua -> Redis）
string	字符串	String	字符串，可以包含任何数据类型	redis.call('set', 'key', stringValue)
number	数字	String	数字存储为字符串	redis.call('set', 'key', tostring(numberValue))
table	数组型表	List/Set/Sorted Set	根据使用的命令转换为列表、集合或有序集合	redis.call('rpush', 'key', unpack(arrayTable))
	键值对表	Hash	字符串字段和字符串值的映射	redis.call('hmset', 'key', unpack(hashTable))
boolean	布尔值	String	布尔值通常转换为"0"或"1"的字符串	redis.call('set', 'key', booleanValue and '1' or '0')
nil	空值	删除键	在Redis中删除对应的键	redis.call('del', 'key')

Redis 数据类型	描述	转换为 Lua 数据类型	描述	示例（Redis -> Lua）
String	字符串，可以包含任何数据类型	string/number	字符串或数字	local value = redis.call('get', 'key')
List	字符串列表，按插入顺序排序	table	数组型表	local list = redis.call('lrange', 'key', 0, -1)
Set	无序集合，不重复的字符串集合	table	数组型表	local set = redis.call('smembers', 'key')
Sorted Set	有序集合，字符串和分数的映射	table	键值对表	local zset = redis.call('zrange', 'key', 0, -1, 'WITHSCORES')
Hash	字符串字段和字符串值的映射	table	键值对表	local hash = redis.call('hgetall', 'key')
Bitmap	由位组成的数组	string	字符串	local bitmap = redis.call('get', 'key')
HyperLogLog	用于基数统计的概率数据结构	nil	不直接转换	N/A
Stream	消息流	table	键值对表	local stream = redis.call('xrange', 'key', start, end)
Geo	地理位置信息	table	键值对表	local geopos = redis.call('geopos', 'key', 'member')

Lua脚本实现秒杀

   @Autowired
   private RedisTemplate redisTemplate;

   @Autowired
   private ExecutorService executorService;

// 定义Lua脚本，库存足够执行INCRBY命令后返回0，库存不够返回0
   private static RedisScript<Long> STOCK_SCRIPT = new DefaultRedisScript<Long>(
           "local stock = redis.call('get', KEYS[1])\n" +
                   "if (stock >= ARGV[1]) then\n" +
                   "  redis.call('decrby', KEYS[1], ARGV[1])\n" +
                   "  return 1\n" +
                   "else\n" +
                   "  return 0\n" +
                   "end", Long.class);

   @ApiOperation(value = "使用Lua实现秒杀")
   @GetMapping("decr3")
   public void decr3() {
       Runnable task = () -> {
           int num = getNum();
           Long result = (Long) this.redisTemplate.execute(STOCK_SCRIPT, Arrays.asList("stock"), num);
           if (result > 0) {
               log.info("下单成功");
           } else {
               log.error("下单失败");
           }
       };

       for (int i = 0; i < 3000; i++) {
           executorService.execute(task);
       }

       // 测试最终结果： 执行效率高，解决库存遗留问题
   }


   public static int getNum() {
       int min = 1;
       int max = 20;

       Random random = new Random();
       int value = random.nextInt(max + min) + min;

       return value;
   }

Lua库存扣减脚本

-- Lua 脚本实现多个产品库存的扣减
-- 检查所有产品库存是否满足扣减条件
-- 如果所有产品库存满足，执行 DECRBY
-- 如果任何一个产品库存不满足，返回该产品的 KEY

-- 获取键的数量
local numKeys = #KEYS

-- 确保 KEYS 和 ARGV 长度一致
assert(#ARGV == numKeys, "每个键必须有对应的扣减值")

-- 首先检查所有库存是否满足扣减条件
for i = 1, numKeys do
    local key = KEYS[i]
    local decrement = tonumber(ARGV[i])
    local currentStock = tonumber(redis.call('get', key) or 0)

    if currentStock < decrement then
        -- 如果任何一个产品的库存不满足条件，返回该产品的 KEY
        return key
    end
end

-- 执行库存扣减
for i = 1, numKeys do
    local key = KEYS[i]
    local decrement = tonumber(ARGV[i])
    redis.call('decrby', key, decrement)
end

-- 所有操作成功，返回 "OK"
return "OK"

Docker 安装 Redis

docker pull redis:latest


docker run -p 6379:6379 --name redis1 -v /data/redis/config/redis_1.conf:/etc/redis/redis.conf -v /data/redis/data/redis1:/data -d redis redis-server /etc/redis/redis.conf --appendonly yes

说明
-v /data/redis/config/redis_1.conf:/etc/redis/redis.conf：把宿主机配置好的redis.conf放到容器内的这个位置中
-v /data/redis/data/redis1:/data： 把redis持久化的数据在宿主机内显示，做数据备份
redis-server /etc/redis/redis.conf：这个是关键配置，让redis不是无配置启动，而是按照这个redis.conf的配置启动
–appendonly yes：redis启动后数据持久化

配置文件 redis.conf

# 允许外部IP访问
bind 0.0.0.0
daemonize no
requirepass wgf123		# 密码
appendonly yes			# 持久化

Redis 配置

参考 掘金文档

Redis 配置

Redis的持久化

Redis是一个高性能的内存键值存储数据库，它主要使用内存作为数据操作的主要媒介，这意味着数据主要在内存中进行读取和写入操作，确保了极高的处理速度。然而，内存的易失性意味着在服务器进程崩溃或由于其他原因意外停止时，存储在内存中的数据可能会丢失。为了克服这个挑战，Redis提供了两种主要的持久化机制：RDB 和 AOF。RDB持久化会在指定的时间间隔内创建数据集的时间点快照，而AOF持久化则记录每个写操作命令。

RDB (Redis DataBase 默认)

性能比AOF优秀 数据丢失风险大 适合海量数据恢复 适合对数据完整性要求不高的场景

什么是RDB

RDB（Redis Database）是 Redis 的一种持久化机制，它通过创建内存数据集的快照来实现持久化。在配置的时间间隔或满足特定条件时，Redis 会自动执行快照操作，将当前内存中的所有数据保存到一个 RDB 文件中。

优点和缺点

优点

1. 数据恢复速度快：在需要从磁盘恢复数据时，RDB 可以快速地加载快照文件，尤其适合大数据集的恢复。
2. 节省磁盘空间：RDB 文件会通过压缩来减少所需的磁盘空间。
3. 较小的性能开销：快照操作在子进程中执行，主进程可以继续处理客户端请求，对数据库性能的影响相对较小。

缺点

1. 数据丢失的风险：RDB 通过定时创建数据的快照。如果 Redis 服务器在两次快照之间发生故障，那么自上次快照以来的所有数据更改都将丢失。
2. 不适合实时持久化：由于 RDB 是周期性创建的，它不适合需要实时持久化的应用。
3. 资源占用：在执行快照时，Redis 需要 fork 一个子进程，这个过程中会消耗额外系统资源。

备份如何执行

1. 使用子进程进行持久化：Redis 通过 fork 创建一个子进程来执行 RDB 持久化操作。这样做的目的确实是为了减少主进程的 I/O 阻塞，允许主进程在持久化过程中继续处理客户端请求。子进程利用操作系统的写时复制（Copy-On-Write, COW）机制来访问数据集的快照，这意味着当主进程修改内存中的数据时（修改数据的副本），子进程所访问的数据仍然保持不变。
2. 使用临时文件写入：子进程不是直接写入 dump.rdb 文件，而是先将快照数据写入一个临时文件中。这个过程中，子进程会创建整个数据集的一个完整、一致性的副本。
3. 安全地同步文件：一旦子进程完成数据的写入，它会将临时文件重命名为 dump.rdb。这种方式确保了即使在持久化过程中出现故障，原有的 dump.rdb 文件也不会受到影响，从而提供了一定程度的数据安全性。
4. 恢复效率的考虑：对于需要进行大规模数据恢复的场景，RDB 模式通常比 AOF 模式更高效，因为 RDB 文件是数据集的压缩表示，可以更快地被载入。同时，如果数据的完整性不是最重要的考虑因素，RDB 可能是更好的选择，因为 AOF 在恢复大量数据时会比较慢。

Fork工作原理

• 初始共享：Redis 通过 fork 创建一个子进程，他们实际上共享相同的物理内存页。操作系统只保留一份数据的副本，而多个进程可以同时访问这份副本。

• 写时复制：Redis 主线程在持久化期间接收客户端请求，修改这些共享的数据时，操作系统首先会为该进程创建这些数据的一个私有副本。

• 数据快照：子进程快照的是原始的共享数据，不会快照主线程在持久化期间创建的副本数据，这些副本数据会在下次快照时被持久化。

dump文件

配置位置及SNAPSHOTTING解析

默认配置文件

参数	说明
dir	设置快照文件的存放路径，这个配置项一定是个目录
stop-writes-on-bgsave-error	默认值为yes。当启用了RDB磁盘满了，Redis是否停止接收数据
rdbcompression	默认值是yes。对于存储到磁盘中的快照，可以设置是否进行压 缩存储
rdbchecksum	默认值是yes。在存储快照后，我们还可以让redis使用CRC64算 法来进行数据校验
dbfilename	设置快照的文件名，默认是 dump.rdb
save	这里是用来配置触发 Redis的持久化条件，也就是什么时候将内存 中的数据保存到硬盘

• 15分钟有一次修改触发保存
• 5分钟有十次修改触发保存
• 1分钟有一万次修改触发保存

如果想禁用RDB持久化的策略，只要不设置任何save指令，或者给save传入一个空字符串参数也可以。若要修改完毕需要立马生效，可以手动使用 save 命令，立马生效。

命令修改RDB频率

save 120 10 # 120秒内修改10次则触发RDB

如果想禁用RDB持久化的策略，只要不设置任何save指令，或者给save传入一个空字符串参数也可以。
若要修改完毕需要立马生效，可以手动使用 save 命令！立马生效，需要删除dump.rdb文件

如何触发RDB快照

• 配置文件配置了 save 策略，服务定期备份
• 命令save或者是bgsave
  • save 时只管保存，其他不管，全部阻塞
  • bgsave，Redis 会在后台异步进行快照操作，快照同时还可以响应客户端请求。可以通过lastsave 命令获取最后一次成功执行快照的时间。
• 执行 flushall 命令，也会产生 dump.rdb 文件，但里面是空的，无意义 !
• 退出 的时候也会产生 dump.rdb 文件！

RDB恢复

1. 命令获取Redis目录

   > config get dir
   dir
   /data

2. 将备份文件（dump.rdb）移动到redis安装目录并启动服务即可。

AOF（Append Only File）

数据完整性要求高 内存数据量不大 影响写入性能

什么是AOF

Redis的AOF（Append-Only File）持久化机制是一种用于确保数据持久性的方法，它通过记录并保存Redis服务器执行的所有写操作到磁盘上的一个文件中。这些记录包括了所有写命令及其参数，并以追加的方式保存在日志文件中。AOF机制的关键在于它记录的是实际执行的命令，而不仅仅是数据变更，这使得通过回放这些命令能够准确重建数据集。

AOF保存的是 appendonly.aof 文件

优点和缺点

优点

• 安全性：AOF 持久化可以配置为每次写操作后都同步到磁盘，或者每秒同步一次，这降低了数据丢失的风险。
• 可读的日志格式：AOF 文件是一个只追加的日志文件，其内容是 Redis 命令的纯文本，这使得文件可以用于故障排查和数据恢复。
• 灵活的恢复策略：可以通过编辑 AOF 文件来手动修复数据或删除错误的写操作。

缺点

• 恢复速度慢：AOF 比 RDB 持久化有更高的磁盘 I/O 开销，尤其是在高写入负载的情况下。
• 对性能的影响： AOF 需要记录每个写操作，因此在高写入负载下会对服务性能产生影响。
• 数据冗余：随着时间的推移，AOF 文件可能包含一些过时或重复的命令，尽管 Redis 提供了重写机制来减少这种情况。

AOF配置

参数	说明
appendfilename	appendfilename AOF 文件名称
appendfsync	appendfsync aof持久化策略的配置 no 写入系统内核缓冲区，由系统自动刷盘，速度最快 always 每次执行写命令后都进行磁盘同步 everysec 每秒进行一次磁盘同步
no-appendfsync-on-rewrite	重写时是否可以运用Appendfsync，用默认no即可，保证数据安全性
auto-aof-rewrite-min-size	64mb。设置允许重写的最小aof文件大小
auto-aof-rewrite-percentage	默认值为100。aof自动重写配置，当目前aof文件大小超过上一次重 写的aof文件大小的百分之多少进行重写

appendonly no # 是否以append only模式作为持久化方式，默认使用的是rdb方式持久化，这
种方式在许多应用中已经足够用了

appendfsync everysec # appendfsync aof持久化策略的配置
# no表示不执行fsync，由操作系统保证数据同步到磁盘，速度最快。
# always表示每次写入都执行fsync，以保证数据同步到磁盘。
# everysec表示每秒执行一次fsync，可能会导致丢失这1s数据。

正常恢复

• 启动：修改默认的appendonly no，改为yes
• 将有数据的aof文件复制一份保存到对应目录（config get dir）
• 恢复：重启redis然后重新加载

异常恢复

• 启动：修改默认的appendonly no，改为yes

• 故意破坏 appendonly.aof 文件

• 修复： redis-check-aof --fix appendonly.aof 进行修复

• 恢复：重启 redis 然后重新加载

  ## 开启aof,设置数据
  > set k1 v1
  OK
  > set k2 v2
  OK
  > set k3 v3
  OK
                            
  停止Redis
                            
  破坏aof文件结构
  *2
  $6
  SELECT
  $1
  0
  *3
  $3
  set
  $2
  k1
  $2
  v1
  *2
  $6
  SELECT
  $1
  0
  *3
  $3
  set
  $2 // 删除
  k2 // 删除
  $2
  v2
  *3
  $3
  set
  $2
  k3
  $2
  v3
                            
  执行命令
  /data/redis/data/redis1# redis-check-aof --fix appendonly.aof 
  0x              49: Expected prefix '$', got: '*'
  AOF analyzed: size=102, ok_up_to=52, diff=50
  This will shrink the AOF from 102 bytes, with 50 bytes, to 52 bytes
  Continue? [y/N]: y
  Successfully truncated AOF
                            
  启动Redis
  > keys *
  k1

AOF重写

压缩AOF文件

auto-aof-rewrite-min-size 64MB   // 当文件小于64M时不进行重写
auto-aof-rewrite-min-percenrage 100  // 当文件比上次重写后的文件大100%时进行重写

AOF采用文件追加方式，记录了所有写入命令，因此AOF文件会随着时间的推移变得越来越大。为了避免出现过大的AOF文件，Redis引入了AOF重写机制。当AOF文件的大小超过预设的阈值（默认为64MB），或者比上次重写后的文件大小增长了100%时，Redis会触发AOF重写。

AOF重写是一种内容压缩的过程，它会根据内存中的数据重新创建一个新的AOF文件，其中只包含能够完全恢复数据的最小指令集。重写过程是在后台进行的，不会阻塞主线程的正常操作。可以使用命令 bgrewriteaof 手动触发AOF重写过程。

通过AOF重写，可以将过大的AOF文件压缩为较小的文件，从而减少磁盘空间的占用和提高AOF文件的读写效率。

重写原理

当Redis的AOF文件持续增长过大时，Redis会进行AOF重写操作以优化文件大小。AOF重写是通过fork出一个新的进程来实现的，首先该进程会将AOF文件的数据以二进制形式重新生成一个新的AOF文件，这个过程是在一个临时文件中进行的。当新的AOF文件生成完毕后，Redis会将旧的AOF文件进行覆盖，即先将临时文件重命名为新的AOF文件的名字，从而完成AOF重写操作。

参考 AOF在重写时如何保证继续服务

触发机制

• 当AOF文件超过64MB（可配置）或AOF文件超过上次重写大小的100%时。
• 调用 BGREWRITEAOF 手动触发

AOF和RDB同时开启

RDB 和 AOF 不是互斥的，可以同时开启，如果AOF和RDB同时存在的时候，Redis会优先使用从AOF文件来还原数据库状态，如果AOF关闭状态时，则从RDB中恢复。

集群数据恢复

在Redis 集群中，如果整个集群的数据量非常庞大，主节点可以开启 AOF 保证数据完整性，从节点开启 RDB 策略。

当节点宕机恢复时，先关闭 AOF 功能，并且先从从节点中同步 dump.rdb 文件使用 RDB 方式进行数据恢复，待 RDB 恢复完毕后再使用命令动态启动 AOF 恢复，保证数据完整性

config get *
config set config set appendonly yes

AOF在重写时如何保证继续服务

AOF重写过程依旧提供服务，由子进程重写

1. 触发重写：AOF 重写可以手动触发，也可以配置 Redis 自动在 AOF 文件达到特定大小后触发。手动触发是通过 BGREWRITEAOF 命令。
2. 创建子进程：Redis 使用子进程来执行重写操作。这样做的好处是避免阻塞主 Redis 进程，保证AOF重写期间主进程继续提供服务。
3. 重建 AOF 文件：子进程通过读取当前数据库的状态，重新生成一份写指令。
4. 记录新命令：在重写过程中，主进程继续处理新的写命令，将它们同时追加到旧的 AOF 文件和一个缓冲区中。这确保了在重写期间对数据库所做的更改不会丢失。
5. 切换 AOF 文件：一旦子进程完成了重建过程，它会通知主进程。然后，主进程将缓冲区中的所有新命令追加到新的 AOF 文件中，以确保数据的完整性。
6. 文件替换：通过 rename 修改文件名完成新旧文件替换。

如何选择

结合使用

为了平衡数据安全性和性能，可以同时启用 RDB 和 AOF。例如，可以使用 RDB 进行定时备份（例如每天一次），同时使用 AOF 来保证更高级别的数据安全性。

集群上

​ 如果做Redis集群，那么主节点选择 AOF 持久化方式保证数据完整性，从节点选择 RDB 保证速度。

业务场景

• 数据集大且读写频繁不太关注数据完整性场景如DB缓存层，可以使用 RDB 提高Redis性能。

• 数据集不大但对数据完整性要求高场景如商品库存数据，可以使用 AOF 保证数据完整性。

高可用

主从复制

AP模型 异步同步 最终一致性保证 主从同步延迟，数据可能不一致

CAP理论

主从复制是一种将一个Redis服务器的数据复制到其他Redis服务器的机制。在主从复制中，源服务器被称为主节点（master/leader），而目标服务器被称为从节点（slave/follower）。数据复制是单向的，只能从主节点复制到从节点。主节点负责写操作，而从节点主要用于读操作。

在默认情况下，每台Redis服务器都是主节点，也就是既能读取又能写入数据。每个主节点可以拥有多个从节点或者没有从节点。然而，每个从节点只能有一个主节点，它会从该主节点复制数据。

• 主节点：负责读写操作。
• 从节点：只能执行读操作，不能写入数据。

主从复制的作用

• 数据冗余：主节点将数据复制到多个从服务器，实现数据冗余保证数据安全。

• 服务冗余：主节点宕机，可以从从节点中重新指定一个主节点实现故障转移。

• 负载均衡：从节点分担主节点读取压力，实现读取能力的水平扩展。

主从架构的缺点

• 数据同步延迟：在主从架构中，主节点负责接收写操作并将数据同步到从节点，存在一定时间的数据延迟。
• 写入性能未提升：主从复制主要用于提高读取性能，因为所有写入操作仍然只能在主节点上进行。
• 单点故障：尽管主从架构可以提供高可用性，但主节点本身仍然是一个单点故障，故障转移需要时间恢复。

全量同步

slave初始化阶段

当一个从节点刚刚加入主从复制架构或由手动执行 SLAVEOF 命令，它需要从主节点获取所有数据的副本，以确保与主节点的数据一致性。这个过程称为全量同步。

全量同步过程如下：

1. 从节点发送 SYNC 命令给主节点，请求进行全量同步。
2. 主节点在接收到 SYNC 命令后，会开始执行 BGSAVE 命令，将当前内存中的数据快照保存到磁盘上的 RDB 文件中。
3. 主节点在完成 BGSAVE 后，将 RDB 文件发送给从节点，并将这段时间内的所有写命令发送给从节点。
4. 从节点接收到 RDB 文件后，会将其加载到内存中，恢复主节点的数据。
5. RDB 文件数据恢复后，从节点继续执行主节点发送的写命令，使自己的数据与主节点保持一致。

增量同步

增量同步是指在主从复制过程中，主节点将自己的写操作日志（也称为命令传播）发送给从节点，使得从节点可以按照相同的顺序来执行这些写操作，从而保持与主节点的数据一致性。

增量同步过程如下：

1. 前提是完成全量同步。
2. 从节点发送 SYNC 命令：附带一个偏移量参数，表示从节点最后一次成功接收到的写操作日志的位置。
3. 主节点处理：
   • 有相应日志： 将偏移量起始位置的所有操作日志发送给从节点，从节点安顺序执行。
   • 没有相应日志：重新全量同步。

部分同步

当从节点断线后重新连接时，可以使用部分同步来只同步断线期间的增量数据，而不是重新进行全量同步。

Redis 6.2 版本引入了复制积压缓冲区（Replication backlog）的功能。通过配置主节点的复制积压缓冲区大小，可以使主节点将写操作日志保存在缓冲区中，并在从节点重新连接时将缓冲区中的日志发送给从节点，从而实现部分同步（无磁盘复制）。

无磁盘复制

在传统的主从复制中，主节点会将数据写入磁盘，并将写操作的命令发送给从节点进行同步。而在无磁盘复制中，主节点在执行写操作后，将写操作的命令直接发送给从节点进行同步，而无需将数据写入磁盘。这种方式能够提高同步的速度和效率，减少了磁盘的IO开销。

无磁盘复制的实现主要依赖于Redis的复制功能。在Redis的复制模式下，主节点将写操作的命令发送给从节点进行同步。从节点会接收到命令并执行，从而达到数据同步的目的。主节点和从节点之间通过网络进行通信，数据的传输速度取决于网络的带宽和延迟。

一主二从配置

Master配置

bind 0.0.0.0
daemonize no
requirepass wgf123

# 关闭rdb
# save ""
save 900 1
save 300 10
save 60 10000

# 关闭aof
appendonly no

Slave1配置

bind 0.0.0.0
daemonize no
requirepass wgf123

# 关闭rdb
save 900 1
save 300 10
save 60 10000

# 关闭aof
appendonly no

#主从复制
# 修改端口
port 6380

# 配置master的ip和端口，如：192.168.1.2 6379
slaveof 192.168.0.181 6379

# 如果当前服务器为slave，那么这里配置的就是master的访问密码
masterauth wgf123

Slave2配置

bind 0.0.0.0
daemonize no
requirepass wgf123

# 关闭rdb
save 900 1
save 300 10
save 60 10000

# 关闭aof
appendonly no

#主从复制
# 修改端口
port 6381

# 配置master的ip和端口，如：192.168.1.2 6379
slaveof 192.168.0.181 6379

# 如果当前服务器为slave，那么这里配置的就是master的访问密码
masterauth wgf123

docker 部署

主节点参考docker-安装redis

docker 添加2个slave实例

需要保证 
	/data/redis/config/redis_slave_1.conf
	/data/redis/config/redis_slave_2.conf
这两个配置文件存在
创建两个目录
	/data/redis/data/redis_slave_1
	/data/redis/data/redis_slave_2

docker run -p 6380:6380 --name redis_slave_1 -v /data/redis/config/redis_slave_1.conf:/etc/redis/redis.conf -v /data/redis/data/redis_slave_1:/data -d redis redis-server /etc/redis/redis.conf

docker run -p 6381:6381 --name redis_slave_2 -v /data/redis/config/redis_slave_2.conf:/etc/redis/redis.conf -v /data/redis/data/redis_slave_2:/data -d redis redis-server /etc/redis/redis.conf

动态指定

还有一种方式可以不通过配置文件实现一主多从，使用slaveof 命令指定 master

slaveof 主机ip 主机端口

测试

查看节点信息

info replication

#### master ####

set k1 v1

set k2 v2

#### Slave1 ####

keys *
k1
k2

#### Slave2 ####

keys *
k1
k2

故障模拟

Docker 停止 Slave2, 模拟Slave 故障

docker stop redis_slave_2

#### master ####

set k3 v3

#### Slave1 ####

keys *
k1
k3
k2

Docker 启动 Slave2

docker start redis_slave_2

#### Slave2 ####

keys *
k1
k3
k2

故障回复，数据重新同步

薪火相传

在Redis的主从架构中，一个从节点也可以充当其他从节点的主节点，实现多级主从复制，也被称为链式复制或级联复制。

在这种情况下，原本是主节点的节点称为顶层主节点或根节点。其他节点既是上一级节点的从节点，同时也是下一级节点的主节点。这样的架构可以形成一个主从节点的链条，实现数据的级联传递和复制。

当一个从节点成为其他从节点的主节点时，它会接收并复制来自上一级节点的数据和操作，并将这些数据和操作传递给下一级节点，以实现层层传递。

需要注意的是，在级联复制中，数据同步的延迟可能会增加，因为每个从节点都需要等待上一级节点同步数据后才能进行同步，链式复制的好处是降低根节点的数据同步压力。

slaveof 主机ip 主机端口

使用场景

1. 减轻主节点同步压力：多个从节点场景，主节点将数据同步给一个从节点X，其他从节点从从节点X处同步数据。
2. 带宽优化：不同地理位置的数据中心仅需要一个节点进行异地复制，其他节点可以是内网复制。

反客为主

当主节点发生故障时，可以将一个从节点提升为新的主节点，接管写操作的处理。这个过程称为故障转移。一旦新的主节点选举完成并开始接收写操作，其他从节点会继续向新的主节点复制数据，以保持数据的一致性。

从节点切换为主节点的命令：

slaveof no one

哨兵模式

Redis哨兵是独立于Redis服务外的独立进程 Master故障转移

Redis哨兵模式通过使用独立的哨兵进程来提供一种高可用性的解决方案。每个哨兵进程都是独立运行的，它们通过发送命令给Redis服务器并等待其响应来监控一个或多个Redis实例的健康状况。当主Redis宕机后，哨兵自行将从Redis提升为主节点，实现故障转移。

哨兵的作用

• 监控：通过心跳机制检测Redis的健康状态。

• 故障转移：如果主服务器宕机。哨兵会自动将其中一个从服务器提升为新的主服务器，并让其余的从服务器更新配置，指向新的主服务器。

• 服务发现：在主服务器发生故障转移后，客户端可以重新连接到新的主服务器，而无需人工修改配置。

多哨兵

Redis 哨兵实例间的相互发现是通过Redis的发布-订阅功能实现的。每个Sentinel定期在特定频道发布自己的信息，并同时订阅这个频道以接收来自其他哨兵的信息。这样，当一个新的哨兵加入网络时，它通过发布信息到这个频道，被其他哨兵发现并加入到监控网络中。因此，多个哨兵监控同一个Redis主节点，它们就能够发现彼此。

在一个哨兵系统中，多个哨兵实例不仅监控 Redis 主服务器和从服务器，而且互相监控以确保哨兵系统本身的健康和可靠性。

哨兵Leader：哨兵们会选举出一个领导者来负责执行故障转移过程。

共识决策：避免单个哨兵因网络问题或其他原因错误地判定主服务器故障，多哨兵间会达成共识。

哨兵配置

#主节点信息，格式：sentinel <master-name> <ip> <redis-port> <quorum>
#<master-name> 为主节点设定的一个自定义名称
#<ip> <redis-port> 主节点的ip和端口
#<quorum> 多少个 Sentinel 同意主节点不可达时，才会发起故障转移（共识决策）

主观下线：主观下线是单个哨兵实例认为主节点线下。

客观下线：quorum 个哨兵认为主节点下线，此时需要进行故障转移。

哨兵模式的缺点

• 数据丢失：主服务器在故障前未能将所有数据同步到从服务器，故障转移可能导致数据丢失。
• 数据不一致：出现网络分区时，Sentinel 实例不能互相通信，可能导致多个从服务器被提升为主服务器，产生 脑裂 现象，导致数据不一致。

避免脑裂：quorum（故障转移达成共识的最少哨兵数）要大于等于哨兵总数的半数 + 1，满足过半原则。

哨兵配置说明

# Example sentinel.conf

# 哨兵sentinel实例运行的端口 默认26379
port 26379

# 哨兵sentinel的工作目录
dir /tmp

# 哨兵sentinel监控的redis主节点的 ip port 
# master-name  可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。
# quorum 当这些quorum个数sentinel哨兵认为master主节点失联 那么这时 客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 2

# 当在Redis实例中开启了requirepass foobared 授权密码 这样所有连接Redis实例的客户端都要提供密码
# 设置哨兵sentinel 连接主从的密码 注意必须为主从设置一样的验证密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster MySUPER--secret-0123passw0rd


# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认30秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000

# 这个配置项指定了在发生failover主备切换时最多可以有多少个slave同时对新的master进行 同步，
#这个数字越小，完成failover所需的时间就越长，
#但是如果这个数字越大，就意味着越 多的slave因为replication而不可用。
#可以通过将这个值设为 1 来保证每次只有一个slave 处于不能处理命令请求的状态。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1



# 故障转移的超时时间 failover-timeout 可以用在以下这些方面： 
#1. 同一个sentinel对同一个master两次failover之间的间隔时间。
#2. 当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的master那里同步数据时。
#3.当想要取消一个正在进行的failover所需要的时间。  
#4.当进行failover时，配置所有slaves指向新的master所需的最大时间。不过，即使过了这个超时，slaves依然会被正确配置为指向master，但是就不按parallel-syncs所配置的规则来了
# 默认三分钟
# sentinel failover-timeout <master-name> <milliseconds>
sentinel failover-timeout mymaster 180000

# SCRIPTS EXECUTION

#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮件通知相关人员。
#对于脚本的运行结果有以下规则：
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执行。

#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等等），将会去调用这个脚本，
#这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常运行的信息。调用该脚本时，将传给脚本两个参数，
#一个是事件的类型，
#一个是事件的描述。
#如果sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执行的，否则sentinel无法正常启动成功。
#通知脚本
# sentinel notification-script <master-name> <script-path>
sentinel notification-script mymaster /var/redis/notify.sh

# 客户端重新配置主节点参数脚本
# 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master地址已经发生改变的信息。
# 以下参数将会在调用脚本时传给脚本:
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# 目前<state>总是“failover”,
# <role>是“leader”或者“observer”中的一个。 
# 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的slave)通信的
# 这个脚本应该是通用的，能被多次调用，不是针对性的。
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh

配置哨兵

data/redis/config 目录下添加配置

sentinel.conf

port 26379
logfile "26379.log"
dir /data
sentinel monitor mymaster 192.168.0.181 6379 1
sentinel auth-pass mymaster wgf123
sentinel down-after-milliseconds mymaster 30000
sentinel parallel-syncs mymaster 2
# 故障转移
sentinel failover-timeout mymaster 60000

docker启动哨兵服务

docker run -p 26379:26379 --name redis_sentinel \
-v /data/redis/config/sentinel.conf:/etc/redis/redis-sentinel.conf \
-d redis redis-sentinel /etc/redis/redis-sentinel.conf

查看节点信息

> info replication
# Replication
role:slave						# 当前节点身份
master_host:192.168.0.181		# 主节点ip
master_port:6379				# 主节点port
master_link_status:up
master_last_io_seconds_ago:1
master_sync_in_progress:0
slave_read_repl_offset:2568
slave_repl_offset:2568
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:c5b383caf344b7bfda01b3891a7cf149983341b7
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:2568
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:2568

暂停主节点

docker stop redis1
等待1分钟

查看节点信息

> info replication
# Replication
role:slave
master_host:172.17.0.1
master_port:6380			# 6380 节点被哨兵选择为主节点
master_link_status:up
master_last_io_seconds_ago:1
master_sync_in_progress:0
slave_read_repl_offset:37895
slave_repl_offset:37895
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:556e8cc700a7e93dc23b8e3bd5556302c15157cf
master_replid2:c5b383caf344b7bfda01b3891a7cf149983341b7
master_repl_offset:37895
second_repl_offset:35671
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:173
repl_backlog_histlen:37723

启动原主节点docker

docker start redis1

查看节点信息

info replication
# Replication
role:slave
master_host:172.17.0.1
master_port:6380		# 6380 还是主节点，原主节点下线再上线就变成从节点
master_link_status:down
master_last_io_seconds_ago:-1
master_sync_in_progress:0
slave_read_repl_offset:1
slave_repl_offset:1
master_link_down_since_seconds:-1
slave_priority:100
slave_read_only:1
replica_announced:1
connected_slaves:0
master_failover_state:no-failover
master_replid:bcbe300ea51d12f5a158f4967582598328e87864
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

SpringBoot 集成哨兵模式

项目连接哨兵服务

添加依赖

<dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
  <groupId>org.apache.commons</groupId>
  <artifactId>commons-pool2</artifactId>
</dependency>
<dependency>
 <groupId>org.projectlombok</groupId>
 <artifactId>lombok</artifactId>
 <version>1.18.22</version>
 <scope>provided</scope>
</dependency>

配置

spring:
redis:
 database: 0
 password: wgf123
 connect-timeout: 3000   # 连接超时时间（毫秒）
 sentinel:
   master: mymaster
   nodes: 127.0.0.1:26379  # 多个哨兵逗号隔开
 lettuce: # Redis的Java驱动包,使用lettuce连接池
   pool:
     max-active: 200 # 连接池最大连接数（使用负值表示没有限制）
     max-wait: -1    # 连接池最大阻塞等待时间（使用负值表示没有限制）
     max-idle: 10    # 连接池中的最大空闲连接 (默认为8)
     min-idle: 0     # 连接池中的最小空闲连接

测试

@Slf4j
@SpringBootTest
public class SentinelTest {
 @Autowired
 RedisTemplate redisTemplate;

 @Test
 public void setTest() {
     redisTemplate.opsForValue().set("test", "test");
 }

 @Test
 public void getTest() {
   log.info("test = {}", redisTemplate.opsForValue().get("test"));
 }
}

Redis 集群

多主节点 哈希槽数据分片

部分参考：https://www.cnblogs.com/yufeng218/p/13688582.html

Redis集群是一个由 多个主从节点群 组成的分布式服务集群，Redis集群不需要哨兵也能完成节点移除和 故障转移 的功能。需要将每个节点设置成集群模式，这种集群模式没有中心节点，可水平扩展。以下是 Redis 集群的一些关键特点和功能：

1. 数据分片：Redis 集群通过自动将数据分布在不同的节点（称为分片）上来提供数据的水平扩展能力。这种分片是基于哈希槽（Hash Slot）的，总共有 16384 个哈希槽。
2. 高可用性：集群模式支持主从复制（Master-Slave Replication），每个主节点都可以有一个或多个从节点。如果主节点宕机，集群会自动执行故障转移，将一个从节点提升为新的主节点。
3. 无中心架构：Redis 集群没有中心节点，每个节点都保存着部分哈希槽的数据。这样的设计提高了系统的可靠性和可伸缩性。
4. 故障转移：在节点出现故障时，集群能够自动进行故障转移，确保数据的持续可用性。
5. 读写分离：可以通过从节点来分担读取负载，从而实现读写分离，提高性能。
6. 请求重定向：客户端与集群中的任何一个节点通信，如果请求的数据不在该节点，客户端将被重定向到正确的节点。
7. 容错性：虽然 Redis 是基于内存的，但它也支持持久化。在集群模式下，即使部分节点宕机，其他节点仍然可以继续提供服务。
8. 水平扩展：Redis 集群支持通过增加更多节点来线性扩展系统的处理能力和存储容量。

集群搭建

部署规划

​ 三组Redis主从复制，三个master，三个slave共6个节点

• 第一组 6400(主) 6401(从)
• 第二组 7400(主) 7401(从)
• 第三组 8400(主) 8401(从)

创建文件夹

在/data/redis/config/cluster目录下分别按照端口创建文件夹，用于存放配置文件

在/data/redis/data/cluster目录下分别按照端口创建文件夹，用于存放持久化数据

配置文件

# 记得修改端口号
port 6400
daemonize no

#指定数据文件存放位置，必须要指定不同的目录位置，不然会丢失数据(记得修改路径)
dir /data

#启动集群模式
cluster-enabled yes

#集群节点信息文件，这里640x最好和port对应上
cluster-config-file nodes-6400.conf

# 节点离线的超时时间
cluster-node-timeout 5000

#去掉bind绑定访问ip信息
bind 0.0.0.0

#关闭保护模式
protected-mode no 

#启动AOF文件
appendonly yes

#如果要设置密码需要增加如下配置：
#设置redis访问密码
requirepass wgf123

#设置集群节点间访问密码，跟上面一致
masterauth wgf123

分别复制6份到/data/redis/config/cluster目录对应的端口文件夹下，记得修改端口以及路径

运行docker

docker run -p 6400:6400 -p 16400:16400 --name redis_6400 -v /data/redis/config/cluster/6400/redis.conf:/etc/redis/redis.conf -v /data/redis/data/cluster/6400:/data -d redis redis-server /etc/redis/redis.conf

docker run -p 6401:6401 -p 16401:16401 --name redis_6401 -v /data/redis/config/cluster/6401/redis.conf:/etc/redis/redis.conf -v /data/redis/data/cluster/6401:/data -d redis redis-server /etc/redis/redis.conf

docker run -p 7400:7400 -p 17400:17400 --name redis_7400 -v /data/redis/config/cluster/7400/redis.conf:/etc/redis/redis.conf -v /data/redis/data/cluster/7400:/data -d redis redis-server /etc/redis/redis.conf

docker run -p 7401:7401 -p 17401:17401 --name redis_7401 -v /data/redis/config/cluster/7401/redis.conf:/etc/redis/redis.conf -v /data/redis/data/cluster/7401:/data -d redis redis-server /etc/redis/redis.conf

docker run -p 8400:8400 -p 18400:18400 --name redis_8400 -v /data/redis/config/cluster/8400/redis.conf:/etc/redis/redis.conf -v /data/redis/data/cluster/8400:/data -d redis redis-server /etc/redis/redis.conf

docker run -p 8401:8401 -p 18401:18401 --name redis_8401 -v /data/redis/config/cluster/8401/redis.conf:/etc/redis/redis.conf -v /data/redis/data/cluster/8401:/data -d redis redis-server /etc/redis/redis.conf

参数说明：-p 16400:16400 我除了要开放6400端口，还要在服务器控制台开放16400端口，才能建立起集群(原端口 +10000)

创建集群

# IP排在前面的是master节点
# 登录任意一台redis
redis-cli -a wgf123 --cluster create 192.168.0.181:6400 192.168.0.181:7400 192.168.0.181:8400 192.168.0.181:6401 192.168.0.181:7401 192.168.0.181:8401 --cluster-replicas 1


>>> Performing Cluster Check (using node 192.168.0.181:6400)
M: 8974118e3a7f1a599c744b0b7b1c7ec12db541a3 192.168.0.181:6400
   slots:[0-5460] (5461 slots) master								# 第一台master节点的hash槽分配
   1 additional replica(s)
S: 9bc6fc61a175c167c02b9c9817a17547098feb14 172.17.0.1:6401
   slots: (0 slots) slave
   replicates c9c36b72db1a20a2436d640ba5668caf611736f9
S: bafbf2b199305aacfe60ccdb5e049ba881a3d281 172.17.0.1:8401
   slots: (0 slots) slave
   replicates 8974118e3a7f1a599c744b0b7b1c7ec12db541a3
M: b4fae0480f54069e46b1ac3fd56b57426e37dc2d 172.17.0.1:8400			
   slots:[10923-16383] (5461 slots) master							# 第三台master节点的hash槽分配
   1 additional replica(s)
M: c9c36b72db1a20a2436d640ba5668caf611736f9 172.17.0.1:7400
   slots:[5461-10922] (5462 slots) master							# 第二台master节点的hash槽分配
   1 additional replica(s)
S: 52a7982c543f1131f7bea9a5886d3ef6fbf7b40f 172.17.0.1:7401
   slots: (0 slots) slave
   replicates b4fae0480f54069e46b1ac3fd56b57426e37dc2d
[OK] All nodes agree about slots configuration.
>>> Check for open slots...                                                                                                                                                                                                                                                       
>>> Check slots coverage...
[OK] All 16384 slots covered.


# 参数说明
-a ：密码；
--cluster-replicas 1：表示1个master下挂1个slave； --cluster-replicas 2：表示1个master下挂2个slave

集群命令

redis-cli --cluster help 

create：创建一个集群环境host1:port1 ... hostN:portN
call：可以执行redis命令
add-node：将一个节点添加到集群里，第一个参数为新节点的ip:port，第二个参数为集群中任意一个已经存在的节点的ip:port
del-node：移除一个节点
reshard：重新分片
check：检查集群状态

查看集群的信息cluster info，登录任意一台node

> cluster info
cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:6
cluster_size:3
cluster_current_epoch:6
cluster_my_epoch:5
cluster_stats_messages_ping_sent:4634
cluster_stats_messages_pong_sent:4667
cluster_stats_messages_meet_sent:1
cluster_stats_messages_sent:9302
cluster_stats_messages_ping_received:4667
cluster_stats_messages_pong_received:4635
cluster_stats_messages_received:9302

查看节点列表cluster nodes

> cluster nodes
b4fae0480f54069e46b1ac3fd56b57426e37dc2d 172.17.0.6:8400@18400 myself,master - 0 1644832798000 5 connected 10923-16383
52a7982c543f1131f7bea9a5886d3ef6fbf7b40f 172.17.0.1:7401@17401 slave b4fae0480f54069e46b1ac3fd56b57426e37dc2d 0 1644832800077 5 connected
bafbf2b199305aacfe60ccdb5e049ba881a3d281 172.17.0.1:8401@18401 slave 8974118e3a7f1a599c744b0b7b1c7ec12db541a3 0 1644832799575 1 connected
8974118e3a7f1a599c744b0b7b1c7ec12db541a3 172.17.0.1:6400@16400 master - 0 1644832799073 1 connected 0-5460
9bc6fc61a175c167c02b9c9817a17547098feb14 172.17.0.1:6401@16401 slave c9c36b72db1a20a2436d640ba5668caf611736f9 0 1644832799575 3 connected
c9c36b72db1a20a2436d640ba5668caf611736f9 172.17.0.1:7400@17400 master - 0 1644832799000 3 connected 5461-10922

查询集群中的值

• CLUSTER KEYSLOT <key>: 此命令用于查询某个特定键（key）被映射到的哈希槽的编号。

• CLUSTER COUNTKEYSINSLOT <slot>: 用于查询指定的哈希槽中包含的键的数量。

• CLUSTER GETKEYSINSLOT <slot> <count>: 此命令返回指定哈希槽中的一定数量的键。

Hash Tag

Redis 集群中的哈希标签（Hash Tag）是一种特殊的机制，用于确保特定的键被分配到相同的哈希槽（Hash Slot），从而允许在集群模式下完成对多个键的操作。

哈希标签通过在键名中加入 {} 来使用。在 {} 中的内容被用于计算哈希槽，而 {} 外的部分被忽略。例如，在键 user{12345}:followers 中，12345 是哈希标签。

Hash Tag 使用

在 Redis 集群模式下，MSET 命令默认情况下不可用，因为它涉及到一次性对多个键进行设置，而这些键可能属于不同的哈希槽。

使用哈希标签解决 MSET 问题:

例如，使用 MSET key1{tag} value1 key2{tag} value2 可以确保 key1{tag} 和 key2{tag} 被分配到同一个哈希槽，从而能使用 MSET 命令。

集群原理分析

• 槽位分配：Redis Cluster 将所有数据划分为 16384 个槽位（slots）。每个槽位负责存储一部分数据，从而实现数据的分布式存储。
• 主从节点与槽位：在集群中，槽位只直接分配给主节点（master nodes）。每个主节点负责一部分槽位。虽然从节点（slave nodes）不直接分配槽位，但它们通过复制各自的主节点来存储相应槽位的数据。
• 客户端的槽位缓存：当 Redis 集群的客户端连接到集群时，它会接收到集群的槽位配置信息，并在本地缓存这些信息。客户端可以直接定位到包含特定键的目标节点，提高查询效率。

槽位定位算法

Cluster 默认会对 key 值使用 crc16 算法进行 hash 得到一个整数值，然后用这个整数值对 16384 进行取模来得到具体槽位。

HASH_SLOT = CRC16(key) % 16384

跳转重定位

类似ES的协调节点 纠正槽位映射

1. 客户端指令处理：当客户端向 Redis 集群的某个节点发送指令时，这个节点首先根据指令中的 key 计算出对应的槽位。
2. 槽位归属检查：接着，该节点检查该槽位是否属于自己管理。如果是，它会直接执行该指令；如果不是，进入下一步。
3. 重定向操作：当槽位不属于当前节点时，该节点会向客户端发送一个重定向响应，其中包含负责该槽位的目标节点地址。
4. 响应客户端：客户端收到重定向响应后，会根据提供的信息连接到正确的节点，并在那里重新执行原指令。
5. 更新槽位映射表：客户端也会更新其本地槽位映射表，以反映最新的槽位和节点对应关系。这样，后续对相同或相关 key 的操作可以直接定位到正确的节点，减少重定向的发生。

Redis集群节点发现

gossip协议：

• Redis 集群使用 Gossip 协议进行节点间的通信和信息交换。这个协议允许节点定期随机交换信息，如其他节点的状态、已知的地址等。
• Gossip 通信是增量的和不频繁的，帮助减少网络带宽消耗，同时确保了集群状态信息的最新性和一致性。

节点发现：

• 新节点加入集群时，它会通过已知节点的信息来发现其他节点，并通过 Gossip 协议与它们建立通信。
• 节点使用 TCP 连接在集群内部彼此通信，默认端口为 Redis 服务端口的加10000，如 Redis 端口为6379，则集群通信端口为16379。

Redis集群选举原理

过半机制 由其他Master节点选取新的Master

• 客观下线：某个主节点宕机，够多的主节点（根据集群配置的 quorum）同意某个主节点失效时，该节点被标记为客观下线（ODOWN）。
• 发起投票：从节点开始发起投投票，只有主节点能够投票。
• 投票限制：每个主节点在一个选举周期内只能投票一次。
• 赢得选举：某个节点获得超过半数主节点的投票，当选新的主节点。
• 通知和同步：通过 Gossip 协议通知集群中的所有节点当选信息，从节点开始同步数据。

故障恢复

如果主节点下线？从节点能否自动升为主节点？注意：15 秒超时。

• 当 6400 挂掉后，8401 成为新的主机。

主节点恢复后，主从关系会如何？主节点回来变成从机。

• 当 6400 重启后，6400 成为 8401 的从机。

如果所有某一段插槽的主从节点都宕掉，redis 服务是否还能继续?

• 如果某一段插槽的主从都挂掉，而 cluster-require-full-coverage=yes，那么 ，整个集群都挂掉。
• 如果某一段插槽的主从都挂掉，而 cluster-require-full-coverage=no，那么，该插槽数据全都不能使用，也无法存储。

redis.conf 中的参数 cluster-require-full-coverage

Redis集群为什么至少需要三个（奇数）master节点

1. 故障转移：在一个主节点故障时，至少需要另外两个主节点来达成共识（基于过半原则），以进行故障转移和选举新的主节点。如果只有两个主节点，当一个节点失效时，另一个节点无法单独做出决策。
2. 避免脑裂：在网络分区的情况下，集群可能被分成两部分。根据选举过半原则，三个节点被分成两部分，其中一部分满足选举过半机制，另一部分不满足。

集群的优缺点

优点

• 水平扩展：通过HashSlot对数据分片，实现水平扩展。
• 高可用性：提供主从复制和故障转移能力。
• 负载均衡：数据请求分摊到多个节点，避免单个节点过载。
• 去中心化架构：减少单点故障的风险，提升了系统的稳定性。

缺点

• 多键操作限制：不允许跨节点的 MSET，LUA，事务 操作，这些操作只能在同一个节点。
• 数据倾斜：数据可能在节点间分布不均，导致部分节点压力较大。

SpringBoot 使用Redis集群

添加maven依赖

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring‐boot‐starter‐data‐redis</artifactId>
</dependency>

<dependency>
  <groupId>org.apache.commons</groupId>
  <artifactId>commons‐pool2</artifactId>
</dependency>

配置文件

spring:
redis:
 database: 0
 password: wgf123
 connect-timeout: 3000   # 连接超时时间（毫秒）
 cluster:
   nodes: 127.0.0.1:6400,127.0.0.1:6401,127.0.0.1:7400,127.0.0.1:7401,127.0.0.1:8400,127.0.0.1:8401
 lettuce: # Redis的Java驱动包,使用lettuce连接池
   pool:
     max-active: 10 # 连接池最大连接数（使用负值表示没有限制）
     max-wait: -1    # 连接池最大阻塞等待时间（使用负值表示没有限制）
     max-idle: 8    # 连接池中的最大空闲连接 (默认为8)
     min-idle: 0     # 连接池中的最小空闲连接

配置类

@Configuration
public class RedisConfig {
 @Bean
 public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
     RedisTemplate redisTemplate = new RedisTemplate<>();
     redisTemplate.setConnectionFactory(redisConnectionFactory);
     // 使用Jackson2JsonRedisSerialize替换默认序列化
     Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
     ObjectMapper                objectMapper                = new ObjectMapper();
     objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
     objectMapper.activateDefaultTyping(new LaissezFaireSubTypeValidator(),
             ObjectMapper.DefaultTyping.EVERYTHING);
     jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
     // 设置value的序列化规则和key的序列化规则
     redisTemplate.setKeySerializer(new StringRedisSerializer());
     redisTemplate.setHashKeySerializer(new StringRedisSerializer());
     redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
     redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
     redisTemplate.afterPropertiesSet();
     return redisTemplate;
 }
}

单元测试

@Test
public void test() {
  for (int i = 1; i <= 100; i++) {
  String key = "k_" + i;
  String value = "v_" + i;
  redisTemplate.opsForValue().set(key, value);
  }
}

最终这100个键值对会hash散落到3个master实例中

分布式锁实现

跨JVM的互斥机制来控制共享资源的访问

spring-integration-redis

申请锁LUA脚本

local lockClientId = redis.call('GET', KEYS[1])					// 根据锁的key获取clientId
if lockClientId == ARGV[1] then
redis.call('PEXPIRE', KEYS[1], ARGV[2])						// 如果当前key的value和clientId一致，则重新设置锁时间（可重入）
return true
elseif not lockClientId then				
redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])			// 如果clientId不存在，则尝试获取锁
return true
end
return false

释放锁LUA脚本(非阻塞)

Reids unlink

if (redis.call('unlink', KEYS[1]) == 1) then					// 将键与键空间断开连接，异步删除
redis.call('publish', ARGV[1], KEYS[1])
return true 
end
return false

释放锁LUA脚本(阻塞)

if (redis.call('del', KEYS[1]) == 1) then 
redis.call('publish', ARGV[1], KEYS[1])  
return true 
end
return false	

防止锁被误删

当线程A获取分布式锁后，如果不加控制，线程B调用释放锁方法可能会把线程A获得的锁进行释放，为了防止这种情况发生，Spring在实现 RedisLockRegistry 子类 RedisLock 的加锁方式时，RedisLock类内部会有一个获得锁的私有变量，当获得分布式锁时设置标志，释放锁时先判断 RedisLock 是否有获得锁的标志，有才执行锁释放方法防止误删

还有另一种做法是使用UUID，在 Redis set时，value拼接上UUID，删除时判断锁对象的UUID和Redis value的 UUID是否相等

整合

添加maven依赖

<!-- 分布式锁支持 -->
<dependency>
 <groupId>org.springframework.integration</groupId>
 <artifactId>spring-integration-redis</artifactId>
</dependency>

添加分布式锁配置

@Configuration
public class RedisLockConfiguration {

 /**
     * 锁前缀
     */
    private final static String LOCK_NAME = "redis-lock";

    /**
     * 锁过期时间
     */
    private final static Long EXPIRE = 20000L;

    @Bean
    public RedisLockRegistry redisLockRegistry(RedisConnectionFactory redisConnectionFactory) {
        return new RedisLockRegistry(redisConnectionFactory, LOCK_NAME, EXPIRE);
    }
}

使用

@Api(tags = "分布式锁API")
@RestController
public class LockController {
    public static int NUM = 100000;

    @Autowired
    private ExecutorService executorService;

    @Autowired
    private RedisLockRegistry redisLockRegistry;

    @ApiOperation(value = "分布式锁")
    @GetMapping("lock")
    public int lock() {
        CountDownLatch countDownLatch = new CountDownLatch(1000);

        Runnable task = new Task(countDownLatch);

        for (int i = 0; i < 1000; i++) {
            executorService.execute(task);
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return NUM;
    }

    class Task implements Runnable {
        private CountDownLatch latch;

        public Task(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            // 获取锁对象
            Lock    lock    = redisLockRegistry.obtain("lock");
            boolean lockRes = false;

            try {
                // 尝试获取锁，10秒超时
                //lockRes = lock.tryLock(10, TimeUnit.SECONDS);

                // 死等
                lock.lock();
                log.info("获取锁成功");
                NUM--;
                latch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 释放锁
                lock.unlock();
            }
        }
    }
}

总结

​ spring-integration-redis 设计优秀之处在于引入本地锁概念，通过本地锁防止高并发场景频繁向Redis发起访问，降低Redis压力。其次是实现了可重入锁的功能，在锁里重新加锁则更新锁的持有时间（TTL）

自定义实现极简版

分布式锁

SET key value [PX milliseconds] [NX] 命令是实现分布式锁的核心

@Component
public class MyRedisLock {
 private final String PREFIX = "redis-lock:";

 @Autowired
 private RedisTemplate redisTemplate;

 /**
     * 获取锁
     *
     * @param lockName    锁名称
     * @param lockTimeOut 锁超时时间 毫秒
     * @return
     */
    public void lock(String lockName, long lockTimeOut) {
        lockName = getLockName(lockName);
        while (true) {
            boolean result = redisTemplate.opsForValue().setIfAbsent(lockName, null, lockTimeOut, TimeUnit.MILLISECONDS);
            if (result) {
                return;
            }

            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 尝试获取锁
     *
     * @param lockName    锁名称
     * @param lockTimeOut 锁超时时间 毫秒
     * @param waitTimeOut 获取锁等待超时时间 毫秒
     * @return
     */
    public boolean tryLock(String lockName, long lockTimeOut, long waitTimeOut) {
        lockName = getLockName(lockName);
        long startTime = System.currentTimeMillis();

        while ((System.currentTimeMillis() - startTime) < waitTimeOut) {
            boolean result = redisTemplate.opsForValue().setIfAbsent(lockName, null, lockTimeOut, TimeUnit.MILLISECONDS);

            if (result) {
                return result;
            }

            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return false;
    }


    /**
     * 释放锁
     *
     * @param lockName 锁名称
     * @return
     */
    public boolean unlock(String lockName) {
        lockName = getLockName(lockName);
        return redisTemplate.delete(lockName);
    }

    private String getLockName(String lockName) {
        return String.format("%s%s", PREFIX, lockName);
    }
}

使用

@RestController
public class LockController {
    public static int NUM = 100000;

    @Autowired
    private ExecutorService executorService;

    @Autowired
    private MyRedisLock myRedisLock;

 
    @ApiOperation(value = "自定义分布式锁")
    @GetMapping("my_lock")
    public int myLock() {
        CountDownLatch countDownLatch = new CountDownLatch(1000);
        Runnable       task           = new MyLockTask(countDownLatch);

        for (int i = 0; i < 1000; i++) {
            executorService.execute(task);
        }

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return NUM;
    }

    class MyLockTask implements Runnable {
        private CountDownLatch latch;

        public MyLockTask(CountDownLatch latch) {
            this.latch = latch;
        }

        @Override
        public void run() {
            try {
                myRedisLock.lock("test", 10 * 1000);
                log.info("获取锁成功");
                NUM--;
                latch.countDown();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                myRedisLock.unlock("test");
            }
        }
    }
}

总结

不推荐这个例子分布式锁的实现方式，死循环获取锁实现过于粗暴，当有强烈的锁竞争场景对Redis会造成较大压力

高可用问题

缓存穿透

访问了根本不存在的数据 请求全部落在DB上

概念

缓存穿透 是指当请求查询一个在缓存和数据库中都不存在的数据时，请求会穿过缓存层直接查询数据库。这种情况如果被恶意利用或频繁发生，会给数据库带来很大压力，甚至导致数据库宕机。

解决方案

布隆过滤器

会有误判率 元素不能删除 布谷鸟过滤器

基本原理：

• 数据结构：布隆过滤器本质上是一个很大的位数组（bit array）和几个哈希函数。
• 插入操作：当插入一个元素时，使用多个哈希函数对元素进行哈希，并将得到的哈希值所对应的位数组中的位置置为 1。
• 查询操作：当查询一个元素时，同样使用这些哈希函数进行哈希，并检查位数组中相应的位置是否都是 1。如果都是 1，那么元素可能存在于集合中；如果任何一个位置是 0，则元素绝对不在集合中。

特点：

• 不确定性：布隆过滤器可能会有误判。
• 不支持删除：标准的布隆过滤器不支持从集合中删除元素，因为将位数组的位从 1 改回 0 会影响其他元素。

应用场景：

• 网络应用：常被用于网络应用中，例如用于网页爬虫的 URL 检查。
• 数据库：数据库系统使用布隆过滤器来快速判断数据是否存，从而避免不必要的访问。

缓存空对象

基本原理：

• 当缓存和数据库都未命中某个查询请求时，在缓存层添加一个 空对象 缓存，并设置 过期时间。
• 后续对同一不存在数据的查询可以直接从缓存中获取这个空对象，避免请求落库。

存在的问题：

• 存储大量空对象会占用缓存空间，被恶意攻击可能导致缓存空间迅速被耗尽。
• 即使设置了过期时间，也可能导致短暂时间的数据不一致（数据库后续添加了空对象对应的业务数据）。

缓存击穿

承载大并发的热点Key过期失效 海量请求落库查询

缓存击穿是指当某个热点数据（即高频访问数据）在缓存中失效（过期）的那一刻，同时有大量并发请求这个数据时，这些请求会直接击穿缓存，全部落到数据库上。在重新缓存数据这一期间，大量并发请求直接访问数据库，可能导致数据库崩溃。

解决方案

设置热点数据永不过期

• 不设置过期时间：对于热点数据，缓存层不设置过期时间。
• 逻辑过期：在缓存的 value 设置过期时间，使用单独线程定期扫描，发现逻辑过期后重新构建缓存。

使用互斥锁

当缓存失效时，不是每个请求都去数据库加载数据，而是使用互斥锁机制确保只有一个请求去数据库查询数据并重新加载到缓存中，其他请求等待缓存恢复后再访问缓存。

缓存雪崩

缓存集中失效 缓存服务节点宕机、断网

缓存雪崩是指在一个缓存系统中，由于缓存系统宕机或者缓存数据大面积失效，导致后续的请求都落到了数据库上，从而引发数据库压力剧增，甚至可能导致数据库服务崩溃。场景如缓存服务宕机恢复后或大量缓存数据集中在同一时间过期。

解决方案

分散Key的过期时间

• 随机过期时间：为缓存数据设置随机的过期时间，而不是统一的过期时间，以避免同时大量缓存失效。
• 预设不同过期时间：根据数据的不同特性和重要性，设置不同的过期时间。

部署高可用缓存系统

• 集群部署：Redis Cluster 缓存集群，提高缓存系统的高可用性。
• 哨兵模式：部署一主多从模式，故障自动转移。

限流降级

当缓存大面积失效后触发熔断器，对缓存服务进行降级处理，返回降级内容。

• 在系统入口处实施限流措施，合理控制访问频率。
• 在缓存失效或不可用时，可以启用服务降级策略，如返回默认数据或简化的服务内容。

使用布隆过滤器

布隆过滤器介绍

布隆过滤器 动画演示

单机 guava 版

添加maven依赖

<dependency>
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  <version>31.0.1-jre</version>
</dependency>

使用源码

@Api(tags = "单机版本布隆过滤器")
@Slf4j
@RestController
public class GuavaBloomFilterController {

 /**
     * 插入多少数据
     */
    private static final int insertions = 1000000;

    /**
     * 期望的误判率
     */
    private static double fpp = 0.02;

    /**
     * 单机版本布隆过滤器
     */
    private BloomFilter<String> bf;

    /**
     * 用于校验bloomFilter 准确性
     */
    private Set<String> keySet = new HashSet<>();

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 数据预热
     * 初始化布隆过滤器
     */
    @PostConstruct
    public void init() {
        bf = BloomFilter.create(Funnels.stringFunnel(Charsets.UTF_8), insertions, fpp);

        // 添加缓存数据
        for (int i = 0; i < 5000; i++) {
            String key   = generate("k", i);
            String value = generate("v", i);

            // 数据添加布隆过滤器
            bf.put(key);
            keySet.add(key);
            redisTemplate.opsForValue().set(key, value);
        }
    }

    @ApiImplicitParams({
            @ApiImplicitParam(name = "start", value = "起始检索点 取值[0, 5000]", dataType = "int"),
            @ApiImplicitParam(name = "end", value = "截止检索点 取值[0, 5000] 需要比start大", dataType = "int")
    })
    @ApiOperation(value = "测试单机版布隆过滤器")
    @GetMapping("test")
    public Map<String, Object> test(@RequestParam("start") Integer start, @RequestParam("end") Integer end) {

        // 根据参数生成key
        List<String> keyList = IntStream.range(start, end)
                .mapToObj(line -> generate("k", line))
                .collect(Collectors.toList());

        // 布隆过滤器拦截
        int          count = 0;
        List<String> keys  = new ArrayList<>();
        for (String key : keyList) {

            // 布隆过滤器过滤
            if (bf.mightContain(key)) {
                keys.add(key);
            }

            // 判断布隆过滤器准确性
            if (keySet.contains(key)) {
                count++;
            }
        }

        double accuracy = keys.size() / count * 100.0d;

        List                result = redisTemplate.opsForValue().multiGet(keys);
        Map<String, Object> map    = new HashMap<>();
        map.put("bloomFilter 准确率", accuracy);
        map.put("缓存数据命中个数", result.size());

        return map;
    }

    private String generate(Object obj1, Object obj2) {
        return String.format("%s_%s", obj1, obj2);
    }

}

测试

启动项目访问 swagger

Redis BitMap 分布式版

核心是使用redis的BitMap 位图代替本地内存存储缓存标识，实现多实例分布式可用的BloomFilter；

实现上依赖 com.google.guava

源码实现

BitMapBloomFilter

public class BitMapBloomFilter<T> {
 /**
     * 布隆过滤器Key
     */
    private String key;

    /**
     * 需要执行hash函数次数
     */
    private int numHashFunctions;

    /**
     * 过滤器大小
     */
    private int bitSize;

    private Funnel<T>     funnel;
    private RedisTemplate redisTemplate;

    /**
     * @param redisTemplate
     * @param expectedInsertions 预计插入的元素数量
     */
    public BitMapBloomFilter(RedisTemplate redisTemplate, String key, int expectedInsertions) {
        this.key           = key;
        this.redisTemplate = redisTemplate;
        this.funnel        = (Funnel<T>) Funnels.stringFunnel(Charset.defaultCharset());
        bitSize            = optimalNumOfBits(expectedInsertions, 0.02);
        numHashFunctions   = optimalNumOfHashFunctions(expectedInsertions, bitSize);
    }

    /**
     * @param redisTemplate
     * @param expectedInsertions 预计插入的数量
     * @param fpp                期望的误判率
     */
    public BitMapBloomFilter(RedisTemplate redisTemplate, String key, int expectedInsertions, double fpp) {
        this.key           = key;
        this.redisTemplate = redisTemplate;
        this.funnel        = (Funnel<T>) Funnels.stringFunnel(Charset.defaultCharset());
        bitSize            = optimalNumOfBits(expectedInsertions, fpp);
        numHashFunctions   = optimalNumOfHashFunctions(expectedInsertions, bitSize);
    }

    /**
     * @param redisTemplate
     * @param funnel
     * @param expectedInsertions 预计插入的数量
     * @param fpp                期望的误判率
     */
    public BitMapBloomFilter(RedisTemplate redisTemplate, String key, Funnel<T> funnel, int expectedInsertions, double fpp) {
        this.key           = key;
        this.redisTemplate = redisTemplate;
        this.funnel        = funnel;
        bitSize            = optimalNumOfBits(expectedInsertions, fpp);
        numHashFunctions   = optimalNumOfHashFunctions(expectedInsertions, bitSize);
    }

    /**
     * 使用murmur hash算法计算值的下标
     *
     * @param value
     * @return
     */
    public int[] murmurHashOffset(T value) {
        int[] offset = new int[numHashFunctions];

        long hash64 = Hashing.murmur3_128().hashObject(value, funnel).asLong();
        int  hash1  = (int) hash64;
        int  hash2  = (int) (hash64 >>> 32);
        for (int i = 1; i <= numHashFunctions; i++) {
            int nextHash = hash1 + i * hash2;
            if (nextHash < 0) {
                nextHash = ~nextHash;
            }
            offset[i - 1] = nextHash % bitSize;
        }

        return offset;
    }

    /**
     * 计算bit数组长度
     */
    private int optimalNumOfBits(long n, double p) {
        if (p == 0) {
            p = Double.MIN_VALUE;
        }
        return (int) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
    }

    /**
     * 计算hash方法执行次数
     */
    private int optimalNumOfHashFunctions(long n, long m) {
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }

    /**
     * 删除布隆过滤器
     *
     * @param key KEY
     */
    public void delete(String key) {
        redisTemplate.delete(key);
    }

    /**
     * 根据给定的布隆过滤器添加值，在添加一个元素的时候使用，批量添加的性能差
     *
     * @param bloomFilterHelper 布隆过滤器对象
     * @param key               KEY
     * @param value             值
     * @param <T>               泛型，可以传入任何类型的value
     */
    public void add(T value) {
        int[] offset = this.murmurHashOffset(value);
        for (int i : offset) {
            redisTemplate.opsForValue().setBit(key, i, true);
        }
    }

    /**
     * 根据给定的布隆过滤器添加值，在添加一批元素的时候使用，批量添加的性能好，使用pipeline方式(如果是集群下，请使用优化后RedisPipeline的操作)
     *
     * @param bloomFilterHelper 布隆过滤器对象
     * @param key               KEY
     * @param valueList         值，列表
     * @param <T>               泛型，可以传入任何类型的value
     */
    public void addList(List<? extends T> valueList) {
        redisTemplate.executePipelined(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                for (T value : valueList) {
                    int[] offset = murmurHashOffset(value);
                    for (int i : offset) {
                        connection.setBit(key.getBytes(), i, true);
                    }
                }
                return null;
            }
        });
    }

    /**
     * 根据给定的布隆过滤器判断值是否存在
     *
     * @param bloomFilterHelper 布隆过滤器对象
     * @param key               KEY
     * @param value             值
     * @return 是否存在
     */
    public boolean contains(T value) {
        int[] offset = this.murmurHashOffset(value);
        for (int i : offset) {
            if (!redisTemplate.opsForValue().getBit(key, i)) {
                return false;
            }
        }
        return true;
    }
}

BitMapBloomFilterRegistry

public class BitMapBloomFilterRegistry {

    public BitMapBloomFilterRegistry(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 默认需要插入的元素
     */
    private static final int DEFAULT_EXPECTED_INSERTIONS = 2000;

    private RedisTemplate redisTemplate;

    /**
     * 过滤器前缀
     */
    private final String prefix = "redis:bloomfilter:";

    /**
     * 布隆过滤器容器
     */
    private final Map<String, BitMapBloomFilter<CharSequence>> filters = new ConcurrentHashMap<>(16);


    /**
     * 获取布隆过滤器
     * 如果获取不到则根据默认配置创建布隆过滤器
     *
     * @param filterName
     * @return
     */

    public BitMapBloomFilter<CharSequence> obtain(String filterName) {
        filterName = this.getFilterName(filterName);
        BitMapBloomFilter<CharSequence> bitMapBloomFilter = filters.computeIfAbsent(filterName, 
                key -> new BitMapBloomFilter<CharSequence>(redisTemplate, key, DEFAULT_EXPECTED_INSERTIONS));
        return bitMapBloomFilter;
    }

    /**
     * 获取布隆过滤器，获取不到则根据配置创建
     *
     * @param filterName         布隆过滤器名称
     * @param expectedInsertions 预计插入的数量
     * @param fpp                期望的误判率
     * @return
     */
    public BitMapBloomFilter<CharSequence> obtain(String filterName, int expectedInsertions, double fpp) {
        filterName = this.getFilterName(filterName);
        Funnel<CharSequence>            funnel            = (Funnel<CharSequence>) Funnels.stringFunnel(Charset.defaultCharset());
        BitMapBloomFilter<CharSequence> bitMapBloomFilter = filters.computeIfAbsent(filterName, 
                key -> new BitMapBloomFilter<CharSequence>(redisTemplate, key, funnel, expectedInsertions, fpp));
        return bitMapBloomFilter;
    }

    private String getFilterName(String name) {
        return String.format("%s%s", prefix, name);
    }
}

添加注册器配置

@Configuration
public class RedisConfig {
    @Bean
    public BitMapBloomFilterRegistry bitMapBloomFilterRegistry(RedisTemplate redisTemplate) {
        return new BitMapBloomFilterRegistry(redisTemplate);
    }

    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        // 使用Jackson2JsonRedisSerialize替换默认序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper                objectMapper                = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(new LaissezFaireSubTypeValidator(),
                ObjectMapper.DefaultTyping.EVERYTHING);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        // 设置value的序列化规则和key的序列化规则
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
}

使用

@Api(tags = "Redis BitMap 版布隆过滤器")
@Slf4j
@RestController
@RequestMapping("bitmap")
public class BitMapBloomFilterController {

    @Autowired
    private BitMapBloomFilterRegistry bitMapBloomFilterRegistry;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 数据预热
     */
    @PostConstruct
    public void init() {
        BitMapBloomFilter<CharSequence> bloomFilter = bitMapBloomFilterRegistry.obtain("test");
        List<String> valueList = IntStream.range(1, 2000)
                .mapToObj(line -> String.format("%s_%s", "k", line))
                .collect(Collectors.toList());

        // 初始化布隆过滤器
        bloomFilter.addList(valueList);

        // 缓存数据初始化
        for (String s : valueList) {
            redisTemplate.opsForValue().set(s, s);
        }
    }


    @ApiImplicitParams({
            @ApiImplicitParam(name = "start", value = "起始检索点 取值[1, 2000]", dataType = "int"),
            @ApiImplicitParam(name = "end", value = "截止检索点 取值[1, 2000] 需要比start大", dataType = "int")
    })
    @ApiOperation(value = "Redis BitMap版布隆过滤器")
    @GetMapping("test")
    public Map<String, Object> test(@RequestParam("start") Integer start, @RequestParam("end") Integer end) {
        // 获取布隆过滤器
        BitMapBloomFilter<CharSequence> bloomFilter = bitMapBloomFilterRegistry.obtain("test");

        // 根据参数生成数据
        List<String> dataList = IntStream.range(start, end)
                .mapToObj(line -> String.format("%s_%s", "k", line))
                .collect(Collectors.toList());

        int          errNum        = 0;
        List<Object> cacheDataList = new ArrayList<>();
        for (String data : dataList) {
            boolean result = bloomFilter.contains(data);

            // 根据BloomFilter 过滤器判断，如果数据存在，则取缓存
            if (result) {
                Object cache = redisTemplate.opsForValue().get(data);
                cacheDataList.add(cache);
            } else {
                // 错误统计
                errNum++;
            }
        }

        Map<String, Object> map = new HashMap(8);
        map.put("命中缓存数据个数", cacheDataList.size());
        map.put("误判个数", errNum);

        return map;
    }
}

Redis 插件版本

Redis官方提供module(插件)redisbloom，这个插件集成了Bloom Filter、Cuckoo Filter、Count-Min-Sketch、Top-K

命令	说明
BF.ADD	将 item 添加到布隆过滤器
BF.EXISTS	检查过滤器中是否存在 item
BF.INFO	返回有关布隆过滤器的信息
BF.INSERT	将多个 item 添加到过滤器。如果过滤器尚不存在，则可以选择设置容量
BF.LOADCHUNK	恢复以前使用 BF.SCANDUMP 保存的布隆过滤器
BF.MADD	将多个 item 添加到过滤器
BF.MEXISTS	对于多个 item，检查每个项目是否存在于过滤器中
BF.RESERVE	创建一个布隆过滤器。设置误判率和容量
BF.SCANDUMP	对Bloom进行增量持久化操作

布隆过滤器

git源码下载

Jedis 提供客户端支持

官方docker整合镜像

docker run -p 6379:6379 --name redis-redisbloom redislabs/rebloom:latest

docker 集成

源码下载 RedisBloomV2.2.12.tar

mkidr /data/redis/module
cd /data/redis/module
wget https://codeload.github.com/RedisBloom/RedisBloom/tar.gz/refs/tags/v2.2.12

#解压
tar -zvxf RedisBloom-2.2.12.tar.gz

#执行安装
cd RedisBloom-2.2.12
make

# 安装成功在目录下会生成 redisbloom.so 文件

修改配置

bind 0.0.0.0
daemonize no
requirepass wgf123

# 关闭rdb
# save ""
save 900 1
save 300 10
save 60 10000

# 关闭aof
appendonly no

# module 配置
# 布隆过滤器 mudule
loadmodule /RedisBloom/redisbloom.so

启动容器

docker run -p 6379:6379 --name redis1\
-v /data/redis/config/redis_1.conf:/etc/redis/redis.conf\
-v/data/redis/data/redis1:/data\
-v/data/redis/module/RedisBloom-2.2.12/:/RedisBloom\
-d redis redis-server /etc/redis/redis.conf --appendonly yes

集成SpringBoot LUA脚本

LUA脚本对象

public class BloomCommand {

 /**
     * 创建布隆过滤器脚本，支持指定过期时间
     */
    private final static RedisScript<Boolean> RESERVE =
            new DefaultRedisScript<Boolean>(
                    "local exists = redis.call('EXISTS', KEYS[1])\n" +
                            "if (exists == 0) then\n" +
                            "  redis.call('BF.RESERVE', KEYS[1], ARGV[1], ARGV[2])\n" +
                            "  if (tonumber(ARGV[3]) > 0) then\n" +
                            "    redis.call('PEXPIRE', KEYS[1], ARGV[3])\n" +
                            "  end\n" +
                            "  return true\n" +
                            "  else\n" +
                            "  return false\n" +
                            "end", Boolean.class);

    /**
     * 将一个item添加到布隆过滤器，如果过滤器不存在则创建过滤器
     */
    private final static RedisScript<Boolean> ADD =
            new DefaultRedisScript<Boolean>("return redis.call('BF.ADD', KEYS[1], ARGV[1])", Boolean.class);

    /**
     * 将多个item添加到布隆过滤器
     */
    private final static String M_ADD = "return redis.call('BF.MADD', KEYS[1], %s)";

    /**
     * 确定一个item是否可能存在于布隆过滤器中
     */
    private final static RedisScript<Boolean> EXISTS =
            new DefaultRedisScript<>("return redis.call('BF.EXISTS', KEYS[1], ARGV[1])", Boolean.class);

    /**
     * 确定过滤器中是否存在一个或多个item
     */
    private final static String M_EXISTS = "return redis.call('BF.MEXISTS', KEYS[1], %s)";


    /**
     * 获取创建布隆过滤器脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getReserveScript() {
        return RESERVE;
    }

    /**
     * 获取添加脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getAddScript() {
        return ADD;
    }

    /**
     * 获取批量添加脚本
     *
     * @param values
     * @return
     */
    public static RedisScript<List> getMultiAdd(Collection<Object> values) {
        return CommandHelper.generateScript(M_ADD, values);
    }

    /**
     * 获取判断item是否存在脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getExistsScript() {
        return EXISTS;
    }

    /**
     * 获取批量判断存在脚本
     *
     * @return
     */
    public static RedisScript<List> getMultiExists(Collection<?> values) {
        return CommandHelper.generateScript(M_EXISTS, values);
    }
}

脚本助手

public abstract class CommandHelper {
    /**
     * 生成脚本
     *
     * @param script
     * @param values
     * @return
     */
    public static RedisScript<List> generateScript(String script, Collection<?> values) {
        StringBuilder sb = new StringBuilder();
        for (int i = 1; i <= values.size(); i++) {
            if (i != 1) {
                sb.append(",");
            }
            sb.append("ARGV[").append(i).append("]");
        }
        return new DefaultRedisScript<List>(String.format(script, sb.toString()), List.class);
    }

    public static List<Object> extract(List<Long> result, List<Object> values) {
        List<Object> list = new ArrayList<>();

        for (int i = 0; i < result.size(); i++) {
            if (1 == result.get(i)) {
                list.add(values.get(i));
            }
        }

        return list;
    }
}

布隆过滤器

public class RedisBloomFilter {

    private final RedisTemplate redisTemplate;
    private final List<String>  key;

    public RedisBloomFilter(RedisTemplate redisTemplate, String key) {
        this.redisTemplate = redisTemplate;
        this.key           = Collections.singletonList(key);
    }

    /**
     * 创建布隆过滤器
     *
     * @param errorRate    误报的期望概率 比如 "0.0001"
     * @param capacity     容量
     * @param milliseconds 过期时间，单位毫秒。-1为用不过期
     * @return
     */
    public boolean reserve(double errorRate, int capacity, int milliseconds) {
        return (boolean) redisTemplate.execute(BloomCommand.getReserveScript(), key, errorRate, capacity, milliseconds);
    }


    /**
     * 添加item到布隆过滤器，不存在则创建布隆过滤器
     *
     * @param item
     * @return
     */
    public boolean add(Object item) {
        return (boolean) redisTemplate.execute(BloomCommand.getAddScript(), key, item);
    }

    /**
     * 批量添加item到布隆过滤器，不存在则创建布隆过滤器
     *
     * @param items
     * @return 返回添加成功的item
     */
    public List<Object> multiAdd(List<Object> items) {
        List<Long> result = (List<Long>) redisTemplate.execute(BloomCommand.getMultiAdd(items), key, items.toArray());
        return CommandHelper.extract(result, items);
    }

    /**
     * 判断item是否存在
     * @param item
     * @return
     */
    public boolean exists(Object item) {
        return (boolean) redisTemplate.execute(BloomCommand.getExistsScript(), key, item);
    }

    /**
     * 批量判断items是否存在
     * @param items
     * @return 返回存在的item
     */
    public List<Object> multiExists(List<Object> items) {
        List<Long> result = (List<Long>) redisTemplate.execute(BloomCommand.getMultiExists(items), key, items.toArray());
        return CommandHelper.extract(result, items);
    }
}

布隆过滤器注册器

@Configuration
public class BloomFilterRegistry {
    private final static String                        PREFIX    = "bloom:";
    private final        Map<String, RedisBloomFilter> container = new ConcurrentHashMap<>(8);

    private RedisTemplate redisTemplate;

    public BloomFilterRegistry(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public RedisBloomFilter getFilter(String key) {
        String name = getName(key);
        return container.computeIfAbsent(name, line -> new RedisBloomFilter(redisTemplate, line));
    }

    private String getName(String key) {
        return String.format("%s%s", PREFIX, key);
    }
}

配置注册器

public class RedisConfig {
    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        // 使用Jackson2JsonRedisSerialize替换默认序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper                objectMapper                = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(new LaissezFaireSubTypeValidator(),
                ObjectMapper.DefaultTyping.EVERYTHING);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        // 设置value的序列化规则和key的序列化规则
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }
    
    // 布隆注册器
    @Bean
    public BloomFilterRegistry bloomFilterRegistry(RedisTemplate redisTemplate) {
        return new BloomFilterRegistry(redisTemplate);
    }
}

使用

@Api(tags = "Redis 插件 布隆过滤器")
@RestController
@RequestMapping("bloom")
public class RedisBloomFilterController {
    @Autowired
    private BloomFilterRegistry bloomFilterRegistry;

    private static final String filterName = "bloom-filter";


    @ApiOperation(value = "创建布隆过滤器")
    @GetMapping("reserve")
    public boolean reserve() {
        RedisBloomFilter redisBloomFilter = bloomFilterRegistry.getFilter(filterName);
        return redisBloomFilter.reserve(0.001, 10000, -1);
    }


    @ApiOperation(value = "添加数据到布隆过滤器")
    @ApiImplicitParam(name = "item", value = "添加数据", dataType = "obj")
    @GetMapping("add")
    public boolean add(@RequestParam("item") Object item) {
        RedisBloomFilter redisBloomFilter = bloomFilterRegistry.getFilter(filterName);
        return redisBloomFilter.add(item);
    }


    @ApiOperation(value = "批量添加数据到布隆过滤器")
    @ApiImplicitParam(name = "values", value = "添加数据，json数组 [1,2,3]")
    @PostMapping("multi_add")
    public List<Object> multiAdd(@RequestBody List<Object> values) {
        RedisBloomFilter redisBloomFilter = bloomFilterRegistry.getFilter(filterName);
        return redisBloomFilter.multiAdd(values);
    }


    @ApiOperation(value = "判断数据是否存在")
    @ApiImplicitParam(name = "item", value = "判断数据", dataType = "obj")
    @GetMapping("exists")
    public boolean exists(@RequestParam("item") Object item) {
        RedisBloomFilter redisBloomFilter = bloomFilterRegistry.getFilter(filterName);
        return redisBloomFilter.exists("test");
    }


    @ApiOperation(value = "批量判断数据是否存在")
    @ApiImplicitParam(name = "values", value = "判断数据,json数组 [1,2,3]")
    @PostMapping("multi_exists")
    public List<Object> multiExists(@RequestBody List<Object> values) {
        RedisBloomFilter redisBloomFilter = bloomFilterRegistry.getFilter(filterName);
        return redisBloomFilter.multiExists(values);
    }
}

测试：项目启动访问swagger

布谷鸟过滤器 可删除元素

命令	描述
CF.ADD	将 item 添加到过滤器
CF.ADDNX	仅当 item 不存在时才将项目添加到过滤器
CF.COUNT	返回 item 在过滤器中出现的可能次数
CF.DEL	从过滤器中删除 item
CF.EXISTS	检查过滤器中是否存在item
CF.INFO	返回过滤器信息
CF.INSERT	将多个 item 添加到过滤器。如果过滤器尚不存在，则可以选择设置容量
CF.INSERTNX	如果多个 item 尚不存在，则将它们添加到过滤器。如果过滤器尚不存在，则可以选择设置容量
CF.LOADCHUNK	恢复以前用CF.SCANDUMP 保存的布谷鸟过滤器
CF.MEXISTS	对于多个 item，检查每个 item 是否存在于过滤器中
CF.RESERVE	创建布谷鸟过滤器并设置其容量
CF.SCANDUMP	启动布谷鸟过滤器的增量保存

布谷鸟过滤器LUA脚本

public class CuckooCommand {

 /**
     * 创建布谷鸟过滤器脚本，支持指定过期时间
     */
    private final static RedisScript<Boolean> RESERVE =
            new DefaultRedisScript<Boolean>(
                    "local exists = redis.call('EXISTS', KEYS[1])\n" +
                            "if (exists == 0) then\n" +
                            "  redis.call('CF.RESERVE', KEYS[1], ARGV[1])\n" +
                            "  if (tonumber(ARGV[2]) > 0) then\n" +
                            "    redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" +
                            "  end\n" +
                            "  return true\n" +
                            "  else\n" +
                            "  return false\n" +
                            "end", Boolean.class);

    /**
     * 将一个item添加到布谷鸟过滤器，如果过滤器不存在则创建过滤器
     */
    private final static RedisScript<Boolean> ADD =
            new DefaultRedisScript<Boolean>("return redis.call('CF.ADD', KEYS[1], ARGV[1])", Boolean.class);


    /**
     * 如果item不存在，将一个item添加到布谷鸟过滤器
     */
    private final static RedisScript<Boolean> ADD_NX =
            new DefaultRedisScript<Boolean>("return redis.call('CF.ADDNX', KEYS[1], ARGV[1])", Boolean.class);

    /**
     * 确定一个item是否可能存在于布谷鸟过滤器中
     */
    private final static RedisScript<Boolean> EXISTS =
            new DefaultRedisScript<>("return redis.call('CF.EXISTS', KEYS[1], ARGV[1])", Boolean.class);

    /**
     * 删除一个item
     */
    private final static RedisScript<Boolean> DEL =
            new DefaultRedisScript<>("return redis.call('CF.DEL', KEYS[1], ARGV[1])", Boolean.class);


    /**
     * 获取创建布谷鸟过滤器脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getReserveScript() {
        return RESERVE;
    }

    /**
     * 获取添加脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getAddScript() {
        return ADD;
    }

    /**
     * 获取添加脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getAddNxScript() {
        return ADD_NX;
    }

    /**
     * 获取判断item是否存在脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getExistsScript() {
        return EXISTS;
    }

    /**
     * 获取删除item脚本
     *
     * @return
     */
    public static RedisScript<Boolean> getDelScript() {
        return DEL;
    }
}

布谷鸟过滤器

public class RedisCuckooFilter {
    private final RedisTemplate redisTemplate;
    private final List<String>  key;

    public RedisCuckooFilter(RedisTemplate redisTemplate, String key) {
        this.redisTemplate = redisTemplate;
        this.key           = Collections.singletonList(key);
    }

    /**
     * 创建布谷鸟过滤器
     *
     * @param capacity     容量
     * @param milliseconds 过期时间，单位毫秒。-1为用不过期
     * @return
     */
    public boolean reserve(int capacity, int milliseconds) {
        return (boolean) redisTemplate.execute(CuckooCommand.getReserveScript(), key, capacity, milliseconds);
    }

    /**
     * 添加item到布谷鸟过滤器，不存在则创建布隆过滤器
     *
     * @param item
     * @return
     */
    public boolean add(Object item) {
        return (boolean) redisTemplate.execute(CuckooCommand.getAddScript(), key, item);
    }

    /**
     * 如果item不存在，添加item到布谷鸟过滤器
     *
     * @param item
     * @return
     */
    public boolean addNx(Object item) {
        return (boolean) redisTemplate.execute(CuckooCommand.getAddNxScript(), key, item);
    }

    /**
     * 判断item是否存在
     *
     * @param item
     * @return
     */
    public boolean exists(Object item) {
        return (boolean) redisTemplate.execute(CuckooCommand.getExistsScript(), key, item);
    }

    /**
     * 删除item
     *
     * @param item
     * @return
     */
    public boolean del(Object item) {
        return (boolean) redisTemplate.execute(CuckooCommand.getDelScript(), key, item);
    }
}

布谷鸟过滤器注册器

public class CuckooFilterRegistry {
    private final static String                         PREFIX    = "cuckoo:";
    private final        Map<String, RedisCuckooFilter> container = new ConcurrentHashMap<>(8);
    private              RedisTemplate                  redisTemplate;

    public CuckooFilterRegistry(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public RedisCuckooFilter getFilter(String key) {
        String name = getName(key);
        return container.computeIfAbsent(name, line -> new RedisCuckooFilter(redisTemplate, line));
    }

    private String getName(String key) {
        return String.format("%s%s", PREFIX, key);
    }
}

配置注册器

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate redisTemplate(RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        // 使用Jackson2JsonRedisSerialize替换默认序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper                objectMapper                = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.activateDefaultTyping(new LaissezFaireSubTypeValidator(),
                ObjectMapper.DefaultTyping.EVERYTHING);
        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
        // 设置value的序列化规则和key的序列化规则
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    // 布谷鸟注册器
    @Bean
    public CuckooFilterRegistry cuckooFilterRegistry(RedisTemplate redisTemplate) {
        return new CuckooFilterRegistry(redisTemplate);
    }
}

使用

@Api(tags = "Redis 插件 布谷鸟过滤器")
@RestController
@RequestMapping("cuckoo")
public class RedisCuckooFilterController {
    @Autowired
    private CuckooFilterRegistry cuckooFilterRegistry;

    private static final String filterName = "cuckoo-filter";


    @ApiOperation(value = "创建布谷鸟过滤器")
    @GetMapping("reserve")
    public boolean reserve() {
        RedisCuckooFilter cuckooFilter = cuckooFilterRegistry.getFilter(filterName);
        return cuckooFilter.reserve(10000, -1);
    }


    @ApiOperation(value = "添加数据到布谷鸟过滤器")
    @ApiImplicitParam(name = "item", value = "添加数据", dataType = "obj")
    @GetMapping("add")
    public boolean add(@RequestParam("item") Object item) {
        RedisCuckooFilter cuckooFilter = cuckooFilterRegistry.getFilter(filterName);
        return cuckooFilter.add(item);
    }


    @ApiOperation(value = "数据不存在，添加数据到布谷鸟过滤器")
    @ApiImplicitParam(name = "item", value = "添加数据", dataType = "obj")
    @GetMapping("add_nx")
    public boolean addNx(@RequestParam("item") Object item) {
        RedisCuckooFilter cuckooFilter = cuckooFilterRegistry.getFilter(filterName);
        return cuckooFilter.addNx(item);
    }


    @ApiOperation(value = "判断数据是否存在")
    @ApiImplicitParam(name = "item", value = "判断数据", dataType = "obj")
    @GetMapping("exists")
    public boolean exists(@RequestParam("item") Object item) {
        RedisCuckooFilter cuckooFilter = cuckooFilterRegistry.getFilter(filterName);
        return cuckooFilter.exists("test");
    }


    @ApiOperation(value = "删除数据")
    @ApiImplicitParam(name = "item", value = "判断数据", dataType = "obj")
    @GetMapping("del")
    public boolean del(@RequestParam("item") Object item) {
        RedisCuckooFilter cuckooFilter = cuckooFilterRegistry.getFilter(filterName);
        return cuckooFilter.del("test");
    }
}

测试：项目启动访问swagger

总结

​ 如果是非分布式系统，不存在多服务使用同一布隆过滤器场景，推荐使用 单机-guava-版 性能最佳

​ 如果是分布式系统，推荐使用 redis-插件版本的布谷鸟过滤器, 性能和功能上比自定义的 redis-bitmap-分布式版 强， redis-bitmap-分布式版 更多的意义是用于了解布隆过滤器的实现方式

内存淘汰

键的过期删除策

Redis的数据已经设置了TTL，不是过期就已经删除了吗？为什么还存在所谓的淘汰策略呢？这个原因我们需要从redis的过期策略聊起

主动删除

过期字典 随机扫描

• 过期字典：Redis 维护一个专门的过期字典，其中存放了所有设置了过期时间的键。这个字典用于跟踪各个键的过期时间。

• 定期扫描： 默认每秒会进行十次过期扫描，大约每隔 100 毫秒一次。这个频率是为了平衡内存占用和性能开销。

• 扫描过程：

  每次扫描采用一种随机和贪心的策略（近视LRU）， 并不遍历整个过期字典。

  1. 从过期字典中随机选择一定数量的键（默认为 20 个）。
  2. 检查并删除其中已过期的键。
  3. 如果在这批键中，超过一定比例（如 1/4）的键已过期，则进行另一轮随机扫描。

被动删除

和定期删除互补

• 惰性删除：与定期删除策略相辅相成。懒惰删除的主要特点是，它并不主动去查找和删除过期的键，而是在键被访问时才检查其是否已过期。

• 删除过程：

  1. 当一个键被访问时（例如通过 GET 命令），Redis 首先检查该键是否设置了过期时间，如果设置了，Redis 接下来会判断该键是否已经过期。

  2. 如果键已经过期，Redis 会在返回结果之前从数据库中删除该键。因此，过期的键不会被返回给客户端。

如果一个过期的键长时间没有被访问，那么它将一直留在数据库中。这种键只有在被访问或者通过定期删除策略被检查到时才会被删除。

内存淘汰策略

内存不足时触发

有了以上过期策略的说明后，就很容易理解为什么需要淘汰策略了，因为不管是定期采样删除还是惰性删除都不是一种完全精准的删除，就还是会存在key没有被删除掉的场景，当内存不足时，所以就需要内存淘汰策略进行补充。

策略	说明	适用场景
noeviction	当内存使用超过配置时会返回错误，不会驱逐任何键（·默认策略）。	适用于内存资源充足，或不允许丢失任何数据的场景。
allkeys-lru	如果内存限制被超出，首先通过 LRU 算法驱逐最久未使用的键。	适用于普通缓存场景，特别是当存储空间有限但希望保留最活跃数据时。
volatile-lru	如果内存限制被超出，首先从设置了过期时间的键中驱逐最久未使用的键。	适用于优先驱逐那些已设置过期时间且最少使用的键的场景。
allkeys-random	如果内存限制被超出，从所有键中随机删除。	当其他方法的开销过大，或没有明确的访问模式时使用。
volatile-random	如果内存限制被超出，从设置了过期时间的键中随机删除。	适用于随机驱逐已设置过期时间的键的场景。
volatile-ttl	从配置了过期时间的键中驱逐即将过期的键。	适用于当内存紧张时优先删除即将过期的键。
volatile-lfu	从所有配置了过期时间的键中驱逐使用频率最低的键。	适用于频繁访问的数据变化不大，但希望清理不经常使用的过期数据的场景。
allkeys-lfu	从所有键中驱逐使用频率最低的键。	当数据的访问频率是驱逐决策的主要因素时使用。

这八种大体上可以分为4种，lru、lfu、random、ttl

内存淘汰流程

Redis 的内存淘汰流程是在内存达到限制时被触发的一系列操作，用于释放内存空间以适应新的写入操作。以下是这个流程的基本步骤：

1. 检测内存使用：Redis 持续监测内存使用情况。当内存使用接近配置的 maxmemory 限制时（默认没有设置），触发内存淘汰机制。
2. 淘汰样本数量：
   1. maxmemory-samples 配置决定了在每次淘汰检查中考虑的键的数量，其默认值通常为 5。
   2. 增加 maxmemory-samples 的值可以提高接近真实 LRU 或 LFU 行为的准确性，但相应地会增加 CPU 使用率。
3. 选择淘汰候选键：在淘汰过程中，Redis 从数据集中随机抽取 maxmemory-samples 个键，并根据配置的淘汰策略（LRU，LFU等）评估这些键。
4. 填充 eviction pool：在每次淘汰检查中，根据评估结果，Redis 会更新 eviction pool。eviction pool 是一个固定大小的结构（通常为 16 条目），用于跟踪当前最有可能被淘汰的键。
5. 执行淘汰：当需要释放内存时，Redis 会从 eviction pool 中选择并淘汰键。通常，从该池中淘汰的是评估为“最应该被淘汰”的键。

LRU算法 (最近最少使用算法)

LRU（Least Recently Used）算法，即最近最少使用算法，是一种常用的缓存淘汰策略。这种算法的核心思想是：如果数据在最近一段时间内没有被访问，那么将来被访问的可能性也不大。因此，当缓存空间不足时，LRU算法会优先淘汰那些最长时间没有被访问的数据。

工作原理：

1. 跟踪数据的访问顺序：每当缓存中的一个数据项被访问时，这个数据项就被移到一个记录了访问顺序的列表的前端。
2. 淘汰最久未使用的数据：当需要空间来存储新的数据项时，位于这个列表末端的数据项（即最久未被访问的数据项）首先被淘汰。

标准LRU算法

LRU算法优缺点

上面两张图片可见，URL算法虽然可以淘汰最近一段时间访问频率比较少的数据。但是有些数据可能就某一段时间没有访问，其他时间段高频访问。上图数据如果要淘汰的话，数据B会被淘汰（淘汰了真正热点数据），URL不能准确判断淘汰的数据是否为热点数据。

• 优点:
  • 相对简单，容易理解和实现。
  • 在很多常见的场景下，能有效地预测数据项的访问模式。
• 缺点:
  • 实现和维护代价相对较高，特别是在数据量大时。
  • 并非在所有场景下都是最优的淘汰策略，特别是在访问模式频繁变化的情况下。

Redis LRU实现

全局时钟 每个key内部维护时钟

Redis维护了一个24位时钟，可以简单理解为当前系统的时间戳，每隔一定时间会更新这个时钟。每个key对象内部同样维护了一个24位的时钟，当新增key对象的时候会把系统的时钟赋值到这个内部对象时钟。比如我现在要进行LRU，那么首先拿到当前的全局时钟，然后再找到内部时钟与全局时钟距离时间最久的（差最大）进行淘汰，这里值得注意的是全局时钟只有24位，按秒为单位来表示才能存储194天，所以可能会出现key的时钟大于全局时钟的情况，如果这种情况出现那么就两个相加而不是相减来求最久的key

struct redisServer {
    pid_t pid; 
    char *configfile; 
    //全局时钟
    unsigned lruclock:LRU_BITS; 
    ...
};
typedef struct redisObject {
 unsigned type:4;
 unsigned encoding:4;
 /* key对象内部时钟 */
 unsigned lru:LRU_BITS;
 int refcount;
 void *ptr;
} robj;

Redis中的LRU与常规的LRU实现并不相同，常规LRU会准确的淘汰掉队头的元素，但是Redis的LRU并不维护队列，只是根据配置的策略要么从所有的key中随机选择N个（N可以配置）要么从所有的设置了过期时间的key中选出N个键，然后再从这N个键中选出最久没有使用的一个key进行淘汰

为什么要使用近似LRU

1、性能问题，由于近似LRU算法只是最多随机采样N个key并对其进行排序，如果精准需要对所有key进行排序，这样近似LRU性能更高

2、内存占用问题，redis对内存要求很高，会尽量降低内存使用率，如果是抽样排序可以有效降低内存的占用

3、实际效果基本相等，如果请求符合长尾法则，那么真实LRU与Redis LRU之间表现基本无差异

4、在近似情况下提供可自配置的取样率来提升精准度，例如通过 CONFIG SET maxmemory-samples指令可以设置取样数，取样数越高越精准，如果你的CPU和内存有足够，可以提高取样数看命中率来探测最佳的采样比例

LFU算法 (最不经常使用)

LFU（Least Frequently Used）算法，即最不经常使用算法，是一种用于管理缓存空间的策略。与LRU（最近最少使用）算法不同，LFU算法的核心思想是根据数据项的访问频率来进行淘汰。简而言之，LFU算法会优先移除访问频率最低的数据项。

工作原理：

1. 统计访问频率：每个数据项都有一个计数器，记录该数据项被访问的次数。
2. 淘汰频率最低的数据：当缓存空间不足时，LFU算法会淘汰那些访问频率最低的数据项。如果多个数据项的访问频率相同，则可能根据其他标准（如时间）来决定哪个数据项被淘汰。

实现方式：

1. 计数器和优先队列：实现LFU算法常用的一种方式是为每个数据项维护一个计数器，并使用优先队列来根据访问频率进行排序。
2. 逐渐减少计数器的值：为了应对长时间运行的情况，系统可能会定期减少所有或一部分数据项的计数器值，以便新的或最近变得活跃的数据项不会被过早地淘汰。

LFU算法优缺点

• 优点:
  • 更好地识别和保留那些频繁访问的数据项。
  • 对于长期运行的应用，可能比LRU表现得更好，因为它考虑了整个运行期间的访问历史。
• 缺点:
  • 实现较为复杂，特别是需要准确且高效地处理访问计数和排序。
  • 如果访问模式发生变化，旧的数据项可能由于历史高访问频率而不被淘汰。

LFU算法非常适用于访问模式相对稳定的情况，其中一些数据项显著地比其他数据项被更频繁地访问。通过淘汰那些很少被访问的数据项，LFU算法能够有效地管理有限的缓存空间。

Redis 常见问题汇总

Redis 的优缺点

优点

• 读写性能优异， Redis能读的速度是110000次/s，写的速度是81000次/s
• 支持数据持久化，支持AOF和RDB两种持久化方式
• 单线程执行，Redis的所有操作都是原子性的
• 数据结构丰富，除了支持string类型的value外还支持hash、set、zset、list等数据结构
• 支持主从复制，主机会自动将数据同步到从机，可以进行读写分离

优点主要集中在redis的工作模式上，单线程、内存存储、IO复用、主从复制

缺点

• 内存限制，不能作用于海量数据的高性能读写
• 主机宕机，会丢失部分数据

使用redis有哪些好处

• 数据读写速度快，基于内存操作
• 支持丰富的数据类型，基本类型五种，特殊类型三种
• 所有命令都是原子性操作
• key过期自动删除（定期删除和惰性删除）

为什么要用Redis做缓存

主要从“高性能”和“高并发”这两点来看待这个问题

高性能

​ 相比从DB中获取数据，由DB再到磁盘加载数据的整个过程耗费时间长，用户体验差。使用Redis做缓存直接在内存中获取数据速度快。如果数据库中的对应数据改变的之后，同步改变缓存中相应的数据即可

高并发

​ 当遇到高并发场景时，DB数据库容易达到性能瓶颈。如果将部分数据放在内存缓存中则可以有效降低数据库压力，防止数据库宕机

为什么要用 Redis 而不用 map/guava 做缓存

主要从“本地缓存”、“分布式缓存”、”缓存一致性“作为思考点

• 本地缓存与分布式缓存： 如果缓存数据只是被一个服务独享，那么使用map是最好的选择。但大多数情况下，系统中的一份缓存数据通常被多个服务共同使用，使用Redis的好处就是能让缓存转变为分布式缓存并提供给多个系统使用
• 保证缓存一致性：本地缓存由各服务独立维护，无法统一管理。Redis作为分布是缓存底层就只有一份缓存数据，不存在缓存不一致场景

Redis和Mysql如何保持数据一致性

一致性

• 强一致性： 这种一致性级别是最符合用户直觉的，它要求系统写入什么，读出来的也会是什么，用户体验性好，但实现起来往往对系统的性能影响大
• 弱一致性： 这种一致性级别约束了系统在写入成功后，不承诺立即可以读到写入的值，也不承诺多久之后数据能够达到一致，但会尽可能地保证到某个时间级别(比如秒级别)后，数据能够达到一致状态
• 最终一致性： 最终一致性是弱一致性的一个特例，系统会保证在一定时间内，能够达到一个数据一致的状态。这里之所以将最终一致性单独提出来，是因为它是弱一致性中非常推崇的一种一致性模型，也是业界在大型分布式系统的数据一致性上比较推崇的模型

实际上，没办法做到数据库和缓存的绝对的一致性，只能保证最终一致性

延迟双删

• 优点：数据不一致的时间很短暂，只有休眠那段时间会不一致
• 缺点：需要考虑删除失败造成数据不一致问题

流程

• 1.先删除缓存
• 2.再写数据库
• 3.短暂休眠
• 4.再次删除缓存

为什么需要休眠n毫秒？如何计算？

n毫秒 = 读数据业务逻辑的耗时 + 时间冗余（几百毫秒）+ Redis主从同步延迟时间

休眠延迟删除是为了防止缓存脏读（并发场景下）

在高并发的场景下，即使是在更新数据库之后立即删除缓存，也可能存在其他线程在这两个操作之间查询缓存并将旧数据加载回缓存的情况。

binlog订阅，异步更新

也可以使用消息队列替代binlog，实现异步更新，比如RabbitMq的延迟队列

• 优点：不需要删除多次

• 缺点：实现成本高，延迟受mysql压力影响

流程

A服务

1. 发起修改请求
2. 修改数据库数据
3. 返回响应

缓存服务

1. 订阅binlog
2. 解析binlog
3. 更新缓存

缓存失效

• 写入时使缓存失效：每当数据在 MySQL 中被更新时，而不是更新缓存，而是简单地从 Redis 中删除相应的缓存数据。下次需要这些数据时，从 MySQL 中重新加载并更新缓存。
• 定时使缓存失效：为缓存数据设置过期时间，强制定期从数据库中重新加载数据。

缓存穿透、缓存击穿、缓存雪崩

高可用问题

掘金文章

缓存穿透

缓存没有，数据库也没有数据。高频缓存穿透 恶意攻击：访问数据库不存在的数据

• 缓存空对象
• 布隆过滤器

缓存击穿

某个承载高并发的热点key过期

• 热点数据不过期
• 互斥锁

缓存雪崩

上面说的缓存击穿是一个热点key的失效，而缓存雪崩是多个热点key同时失效或缓存服务崩溃

• 数据预热，缓存时间随机
• 限流降级
• redis集群

RedLock

RedLock 是 Redis 官方推出的一种分布式锁的实现算法。这种算法的目的是在分布式环境中提供一种相对安全和可靠的方式来实现锁机制。RedLock 主要用于那些需要跨多个进程或系统同步资源访问的场景。

RedLock 算法的工作原理:

1. 多个 Redis 实例:

   • RedLock 算法要求部署多个（通常是五个）独立的 Redis 实例。这些实例互不相同，不是主从关系也不是集群模式。

2. 获取锁:

   • 当客户端尝试获取锁时，它会向所有的 Redis 实例发送获取锁的请求。请求包含一个唯一的锁标识符和一个时间戳。
   • 客户端尝试在每个 Redis 实例上使用 SETNX 命令（或 SET 命令的 NX 选项）来设置一个具有过期时间的锁。

3. 锁的获取规则:

   • 客户端需要在大多数（至少三个）Redis 实例上成功设置锁，才被认为是成功获取了锁。
   • 如果客户端在多数实例上未能成功获取锁，它会立即释放在所有实例上的锁。

4. 锁的释放:

   • 当客户端完成其操作时，它会向所有 Redis 实例发送释放锁的命令。

RedLock 的特点:

• 容错性:
  • 由于 RedLock 需要在大多数实例上获取锁，因此即使其中某些实例不可用，仍然可以保证锁的有效性。
• 安全性:
  • RedLock 提供了比单个 Redis 实例更高的安全保障，因为它减少了单点故障的风险。
• 公平性:
  • 通过在多个实例上获取锁，RedLock 尝试确保锁的公平分配，防止单个客户端长时间占用锁。

RedLock 算法在学术和工业界有一些争议。有些专家认为 RedLock 不能提供严格的安全保障，特别是在网络分区和其他极端情况下。

为什么是删除缓存，不是更新缓存

参考文章

• 避免脏数据：在高并发的环境下，直接更新缓存可能导致数据不一致的情况，尤其是当有多个线程同时更新数据时。

• 按需缓存：不是所有数据都需要被频繁访问。通过删除缓存，可以确保只有真正需要的数据被缓存，从而有效管理缓存空间。

• 性能考虑：删除操作通常比更新操作要快，并且更新的数据不一定需要被缓存。

Redis 和 Memcached 有啥区别，为什么选择Redis

数据类型

与Memcached仅支持简单的key-value结构的数据记录不同，Redis支持的数据类型要丰富得多。Memcached基本只支持简单的key-value存储，不支持枚举，不支持持久化和复制等功能

持久性

redis支持数据落地持久化存储,可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用。 memcache不支持数据持久存储

分布式存储

redis支持master-slave复制模式

memcache可以使用一致性hash做分布式

cpu利用

redis单线程执行

memcache支持多核，海量数据性能高于Redis

为什么是Redis

首先Redis支付丰富的数据类型，满足日常开发场景需求，可以做分布式锁，可以使用List做任务队列等。其次提供AOF和RDB持久化数据，数据安全性有保证

Redis 的线程模型

文件事件处理模型

1. 单线程的文件事件处理器:
   • Redis 的核心是基于单线程的文件事件处理模型。这意味着它的网络请求处理、命令执行等，都在一个单线程中顺序执行。
2. IO 多路复用:
   • Redis 使用 IO 多路复用机制来同时监控多个 Socket。这允许单线程高效地处理多个并发网络连接。
3. 文件事件处理器的组成:
   • 文件事件处理器由以下四个主要部分组成：
     • 多 Socket：负责与客户端的网络连接。
     • IO 多路复用程序：监控多个客户端请求，并将请求转为事件放入事件队列。
     • 文件事件分派器：从事件队列获取事件，并根据事件类型将它们分派给对应的事件处理器。
     • 事件处理器：包括连接应答处理器、命令请求处理器、命令回复处理器等，根据不同事件执行相应操作。
4. 事件处理流程:
   • 当多个 Socket 并发产生不同的操作时，这些操作产生的文件事件被 IO 多路复用程序捕获并放入队列。事件分派器然后逐一从队列中取出事件，并将其交给相应的事件处理器进行处理。
5. Redis 6.0 及以上版本的多线程 I/O:
   • 从 Redis 6.0 开始，引入了多线程来处理网络 I/O 的读写操作，但数据的实际读取、处理和回复操作仍在单个主线程中完成。

Redis分布式锁过期了但业务还没有执行完，怎么办

看门狗机制自动延长

参考 redisson 的实现，就是会有watchdog定时判断key是否过期，否则就延长key

• redis分布式锁需要实现一个可延期的策略，相同的客户端可以实现锁延期
• 锁延期的条件，锁的key存在，并且锁的value值和当前的Client要一致
• 每个客户端对应一个定时任务，入参为锁的key和ClientId，定期轮询，比如10秒。当redis中的key相同value和ClientId相同时，自动延长锁时间

客户端宕机

​ 这种时候锁不会自动延期，那就等key的过期时间到期后自动删除其他客户端再申请锁

客户端宕机立即释放分布式锁

思路：客户端宕机说明不能在客户端上有任何期望，可以参考Redis哨兵的实现方式。我们额外开发一个分布式锁哨兵，它独立于Redis和客户端之外

方案：

• 研发一个独立的服务分布式锁哨兵
• 客户端启动的时候将ClientId和Redis连接信息上报给哨兵
• 哨兵定期心跳访问客户端，如果发现客户端宕机则遍历所有分布式锁Key, 判断其value是否为宕机的ClientId，是则删除该key

多个系统同时操作（并发）Redis带来的数据问题

参考

多个系统需要修改同一个key,可能有网络波动，如何保持顺序性

• 加分布式锁，保证一段时间内只有某个服务可读可写
• 使用事务Watch，乐观锁方案

Redis的应用场景

计数器

可以对 String 进行自增自减运算，从而实现计数器功能。Redis 这种内存型数据库的读写性能非常高，很适合存储频繁读写的计数量

缓存

将热点数据放到内存中，减少数据加载时间损耗

会话缓存

可以使用 Redis 来统一存储多台应用服务器的会话信息，实现会话共享

分布式锁实现

在分布式场景下，无法使用单机环境下的锁来对多个节点上的进程进行同步。可以使用 Redis 自带的 SETNX 命令实现分布式锁，除此之外，还可以使用官方提供的 RedLock 分布式锁实现

Redis持久化

持久化就是把内存的数据写到磁盘中去，防止服务宕机了内存数据丢失

• RDB 默认
• AOF
• AOF在重写时如何保证继续服务

如何选择合适的持久化方式

• 如果对数据有安全性要求，同时使用两种持久化功能。在这种情况下，当 Redis 重启的时候会优先载入AOF文件来恢复原始的数据，如果数据量大可以使用RDB做全量数据恢复，AOF做增量恢复

• 如果Redis数据量不大并对数据完整性有严格要求则使用AOF方式,它会记录每个写操作

• 如果Redis数据量很大并且对数据完整性没有强制要求则使用RDB,RDB数据恢复速读比AOF快

• 如只希望数据在服务器运行的时候存在，也可以不使用任何持久化方式

Redis持久化数据和缓存怎么做扩容

如果Redis被当做缓存使用，使用一致性哈希（hash槽）实现动态扩容缩容

• hash环实现 一致性hash

• redis 扩容

Redis的过期键的删除策略 （对过期的数据怎么处理）

键的过期删除策

• 定期删除
• 惰性删除

内存淘汰策略

• 内存满了报错，默认
• LRU
• LRF
• TTL
• 随机

Redis key的过期时间和永久有效分别怎么设置

expire和persist命令

如何保证redis中的数据都是热点数据

这里主要考虑到使用redis的内存淘汰策略实现

内存淘汰策略

Redis如何做内存优化

合理使用数据结构 使用更优秀的序列化技术

可以好好利用hash,list,sorted set,set等集合类型数据，因为通常情况下很多小的Key-Value可以用更紧凑的方式存放到一起。尽可能使用散列表（hashset），散列表（是说散列表里面存储的数少）使用的内存非常小，所以你应该尽可能的将你的数据模型抽象到一个散列表里面。比如你的web系统中有一个用户对象，不要为这个用户的名称，姓氏，邮箱，密码设置单独的key，而是应该把这个用户的所有信息存储到一张散列表里面

什么是Redis事务

Redis 事务的本质是通过MULTI、EXEC、WATCH等一组命令的集合。事务支持一次执行多个命令，一个事务中所有命令都会被序列化。在事务执行过程，会按照顺序串行化执行队列中的命令，其他客户端提交的命令请求不会插入到事务执行命令序列中,不保证原子性

事务

Redis分布式寻址算法

hash槽

在集群模式下，Redis 的 key 是如何寻址的？分布式寻址都有哪些算法？了解一致性 hash 算法吗？如何动态增加和删除一个节点？

hash 算法（大量缓存重建） 一致性 hash 算法（自动缓存迁移）+ 虚拟节点（自动负载均衡） Redis cluster 的 hash slot 算法

hash 算法

来了一个 key，首先计算 hash 值，然后对节点数取模，接着打在不同的 master 节点上。缺点也很明显：某一个 master 节点宕机，所有请求过来，都会基于最新的剩余 master 节点数去取模，尝试去库中取数据进行缓存。这会导致大部分的请求过来，全部无法拿到有效的缓存，导致大量的流量涌入数据库

一致性 hash 算法

将整个 hash 值空间组织成一个虚拟的圆环，整个空间按顺时针方向组织，下一步将各个 master 节点（使用服务器的 ip 或主机名）进行 hash。来了一个 key，首先计算 hash 值，并确定此数据在环上的位置，从此位置沿环顺时针“行走”，遇到的第一个 master 节点就是 key 所在位置，这样就能确定每个节点在其哈希环上的位置。在一致性哈希算法中，如果一个节点挂了，受影响的数据仅仅是此节点到环空间前一个节点（沿着逆时针方向行走遇到的第一个节点）之间的数据，其它不受影响。增加一个节点也同理。 虚拟节点：一致性哈希算法在节点太少时，容易因为节点分布不均匀而造成缓存热点的问题。为了解决这种热点问题，一致性 hash 算法引入了虚拟节点机制，即对每一个节点计算多个 hash，每个计算结果位置都放置一个虚拟节点。这样就实现了数据的均匀分布，负载均衡

hash slot 算法

Redis cluster 有固定的 16384 个 hash slot，slot是槽的概念（理解为数据管理和迁移的基本单位），所有的键根据哈希函数映射到 0~16383 整数槽内，每个节点负责维护一部分槽以及槽所映射的键值数据。 公式：slot = CRC16（key）& 16384。解释：对每个 key 计算 CRC16 值，然后对 16384 取模，可以获取 key 对应的 hash slot。hash slot可以像磁盘分区一样自由分配槽位，在配置文件里可以指定，也可以让redis自己选择分配，结果均匀，这种结构很容易添加或者删除节点。如果增加一个节点，就需要从节点已有的节点 获得部分槽分配到新的节点 上。如果想移除已有的一个节点，需要将节点中的槽移到其他节点上，然后将没有任何槽的节点从集群中移除就可以了。由于缓存的key hash结果是和slot绑定的，而不是和服务器节点绑定，所以节点的更替只需要迁移slot即可平滑过渡。从一个节点将哈希槽移动到另一个节点并不会停止服务，所以无论添加删除或者改变某个节点的哈希槽的数量都不会造成集群不可用的状态