Redis 入门：从实践到理论

Chapter 1. 基本概念和 CLI 使用

1.1 NoSQL

用于存储非结构化数据，不保证 ACID 事务特性（仅有最终一致性 Weak Consistency Model）。

Redis（Remote Dictionary Server）就是一类基于内存的键值型 NoSQL，不保证数据一致性，但可以保证性能。

• 一种 KVStore System，可以方便的存放非结构化数据，这对于缓存各异性数据非常有帮助；

• Handle 网络请求多线程。处理指令单线程，单个指令具有原子性；

• 低延迟，利用 I/O Multiplexing 在单线程中处理多个请求；
• 支持数据持久化；
• 支持主从集群（从备份，读写分离）和分片集群（数据拆分，存储上限提高）；

1.2 Redis Data Structure

Redis Key 一般使用 String，Value 支持：

• 基本类型：String、Hash、List、Set、SortedSet；
• 特殊类型：GEO（地理位置信息格式）、BitMap（位图）、HyperLog；

1.3 Basic Redis CLI Commands

1.3.1 General

命令行指令较多，建议查询官方文档而不是背诵。介绍常见的几个（General）：

• KEYS <pattern>：查询 Key 符合 pattern（R.E.）的键值对；
• DEL/EXISTS <KEY>：删除、判断存在性；
• EXPIRE/TTL <KEY> [sec]：设置/获取键值的有效期（-1 为永久、-2 为已过期）；

1.3.2 String (+int/float)

• SET/MSET <KEY> <VAL>[...(KEY, VAL)]：设置/批量设置键值；
• GET/MGET <KEY>[...KEY]：获取/批量获取；
• INCR/INCRBY <KEY> [STEP]：让存储数值型 String 的 Value 自增 1 或 STEP；
• INCRBYFLOAT <KEY> [incr]：让存储浮点型 String 的 Value 增长指定值；
• SETNX <KEY> <VAL>：仅不存在才插入（决不更改已存在的数据）；
• SETEX <KEY> <sec> <VAL>：设置并指定有效期；

1.3.3 The Hierarchy Structure of Redis Key

Redis Key 允许使用多个单词形成层级结构，常用格式为 <PROJECT_NAME>:<BUSSINESS_NAME>:[TYPE]:<id>；

1.3.4 Hash

String 结构存储时，想要修改其中某个字段不方便。

现在引入 Hash 数据类型，其 value 作为一个无序字典（多了一层 field-value 关系），类似一个 HashMap；

• HSET/HGET <KEY> <FIELD> [VAL]
• HMSET / HMGET
• HGETALL <KEY>：获取这个键的 value 中的所有 field 的值；
• HKEYS：获取这个键的所有 field；
• HVALS：获取这个键的所有 value；
• HSETNX <KEY> <sec> <FIELD> <VAL>：仅不存在才插入；

1.3.5 List

Redis 中的 value 类型为列表，与双向链表很相似（同时支持正向、反向索引）。特性：

• 有序、可重复、增删操作快、查询速度一般；

操作如下：

• LPUSH/RPUSH <KEY> <ELEMENT>[...ELEMENT]：从列表左侧/右侧插入；
• LPOP/RPOP <KEY>：弹出；
• LRANGE <KEY> <START> <END>：获取列表中的角标范围中所有元素；
• BLPOP / BRPOP <KEY>[...KEY] <sec>：和 LPOP/RPOP 类似，但是在没有元素时等待一段时间，而不是直接返回 NIL；

1.3.6 Set

Redis 中的 value 类型为集合，与 HashSet 类似。特性：

• 无序、元素不可重复、查找快、支持交并差集操作；

操作如下：

• SADD/SREM <KEY> <MEMBER>[...MEMBER]；添加/移除集合中的若干元素；
• SCARD <KEY>：获取集合中元素个数；
• SISMEMBER <KEY> <MEMBER>：是否在集合内；
• SMEMBERS <KEY>：获取集合中所有元素；

1.3.7 SortedSet

Redis 中 value 类型为有序集，每个元素携带 score 值（相当于优先级），可以基于 score 排序。

其底层基于 SkipTable + HashTable 实现。特性如下：

• 可排序、元素不重复、查询速度快；

操作如下：

• ZADD <KEY> <score> <MEMBER>：添加一个或多个元素到 sorted set，如果已经存在则更新 score 值；
• ZREM <KEY> <MEMBER>：移除一个指定元素；
• ZSCORE/ZRANK <KEY> <MEMBER>：获取指定元素 score 值 / 排名；

• ZCARD <KEY>：获取元素个数；

• ZCOUNT/ZRANGEBYSCORE <KEY> <MIN_SCORE> <MAX_SCORE>：统计 score 值在指定范围内的元素个数 / 元素值；

• ZRANGE <KEY> <MIN_RANK> <MAX_RANL>：获取指定排名区间内的元素；
• ZDIFF / ZINTER / ZUNION：求差集、交集、并集；

Chapter 2. Using Redis With Spring Boot

2.1 Basic Usages

引入 Redis Client for Java 的 Spring Boot 依赖：

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-data-redis'
    // 连接池实现
    implementation 'org.apache.commons:commons-pool2'
}

配置 Redis 连接信息：

spring:
    redis:
        host: ...
        post: ...
        password: ...
        # 使用 lettuce 而非 jedis 实现
        lettuce:
            pool:
                max-active: 8	# 最大连接数
                max-idle: 8		# 最大空闲连接
                min-idle: 0		# 最小空闲连接
                max-wait: 100ms	# 最大连接等待时长

使用时自动装配 RedisTemplate 类型。

使用 RedisTemplate 的接口：

• RedisTemplate#opsForValue(Object key, Object val, [long timeout, TimeUnit unit])：获取 Java 一般 Object 类型的操作，这个方法会将 key, val 全部序列化后存储。

  默认 JdkSerializationRedisSerializer 类型作为序列化器，底层使用 ObjectOutputStream#writeObject 完成序列化工作。

  缺点：可读性差、默认序列化器的内存占用大（消息队列中的默认序列化器有同样问题）；

  我们应该少用这种方法，尽量选择下面确定类型的方法。

  还有一种方法是，自定义 Redis 的序列化方式。

• RedisTemplate#opsForXXX()：返回 Spring Data Redis 对于指定数据类型 XXX（String/Hash/…）的可能操作集合 XXXOperations；

XXXOperations 类的对象可以完成对应类型的 set / get 等方法，并选用合适的序列化器进行存储处理。

2.2 Self-defined Serializer for Redis

@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
    /* Create empty template */
    RedisTemplate<String, Object> template = new RedisTemplate<>();
    /* Configure connection factory */
    template.setConnectionFactory(redisConnectionFactory);
    /* Set Serializer for POJO */
    GenericJackson2JsonRedisSerializer jsonRedisSerializer
            = new GenericJackson2JsonRedisSerializer();
    /* Use String serializer for key & hash key (instead of Jdk serializer) */
    template.setKeySerializer(RedisSerializer.string());
    template.setHashKeySerializer(RedisSerializer.string());
    /* Use JSON serializer for value & hash value (instead of Jdk serializer) */
    template.setValueSerializer(jsonRedisSerializer);
    template.setHashValueSerializer(jsonRedisSerializer);
    return template;
}

有个问题，GenericJackson2JsonRedisSerializer 对一类数据会反复保存 @class 字段（反序列化后的类型）。这个信息虽然是必要的，但很多情况下存储 @class 字段甚至比原数据还要占空间！

因此我们一般不会直接使用 JSON 序列化器来序列化我们的 POJO，而是我们开发者对一个层级下的键都约定一个数据类型，然后使用 String 序列化器，最终手动序列化和反序列化，那么可以节省这部分空间。

Spring 中已经帮我们包装好了序列化、反序列化全是 String 的 RedisTemplate，省得我们配置了，它就是 StringRedisTemplate；

也就是 RedisTemplate<String, Object>，并且设置好全部都是 String 的序列化、反序列化器；

Chapter 3. 内存型数据库理论

本章介绍一下以 Redis 为首的 NoSQL 内存型数据库产生、发展的历程，同时介绍其存在的问题以及解决方案。

3.1 为什么需要内存型数据库？

持久化在磁盘上的关系型数据库在存储关系数据、处理事务的多数场合下都非常得力，但免不了存在一些问题。

例如，在电商、文章档案等网页应用中，常常是读请求远多于写请求，即便 MySQL 有 cache buffer pool（InnoDB），在大量数据查询的场合下也会出现频繁的 cache evict，究其原因就是 cache working space 太小了。

人们发现只是读请求造成的 Disk I/O 是可以避免的——通过将数据托管到一个更大的内存空间（这段内存空间可以不连续、甚至可以不在单个物理节点上，由一个程序来管理它）中缓存起来，可以有效提升这些应用的处理效率和吞吐量。

结论 1：在庞大数据量的应用场景下，读多写少、数据时间局部性强的应用访问模式可以通过外置的内存缓冲区统一进行缓存，来提升整体性能和接口承载量。这就是 Redis 要解决的需求痛点。

3.2 缓存策略

在确定使用内存型数据库作为持久化的关系型数据库的缓存后，接下来，和所有缓存机制一样（不仅仅是数据库领域），内存型数据库会遇到两个问题：

1. 采取什么样的读/写缓存策略更好？
2. 当缓冲区占满后，evict 的策略是什么？

3.2.1 缓存读写策略 和 缓存模式

对于第一个问题，区分缓存读策略和缓存写策略。

缓存读策略：大多数情况下比较显然：

• 如果 read cache miss，一定需要从磁盘上读数据，顺带写回缓存；
• 如果 read cache hit，则可以考虑记录数据热点情况；

缓存写策略：主要会有 4 类策略：

• 如果 write cache hit，可以：
  • Write-through：立即将数据写入缓存（即覆盖当前行）并主动刷新（flush）到磁盘；
  • Write-back：先把数据写入缓存，但不立即刷新，直到下一个数据要覆盖这个数据行的时候，才更新到磁盘中（defer write to disk until replacement of line，只是尽可能推迟了写入磁盘的时间）。另外，这个方案需要额外维护 dirty bit 来指示是否与磁盘数据一致；
• 如果 write cache miss，可以：
  • Write-allocate：写分配，在 write-miss 后，先将原数据从磁盘读入缓存，转换为 write-hit 的情况，再 write-back（仅修改缓存 + dirty bit）；
  • No-write-allocate：直接写入磁盘，不加载到缓存（缓存中没有这个数据所在的数据行，因为本来就是 write-miss）；

显然，在 write cache hit 情况下，write-through 和 write-back 策略的优劣势互补：前者保证内存和磁盘的数据较强的一致性，但是同步写回操作免不了降低了操作性能；后者接受了一定程度上的数据不一致性（推迟刷盘时间），换取了短时间内的高并发性能。

write-allocate 和 no-write-allocate 之间的优劣势和 write-through、write-back 的优劣势的对比相同，因此人们常常根据实际情况选择 “write-back + write-allocate” 或 “write-through + no-write-allocate” 的策略中的其中一对。

上述读写策略的不同选择，就形成了以 Redis 为首的内存数据库的 3 个主流的宏观缓存模式，每种模式可以应对一些使用场景：

• 旁路缓存模式（Cache Aside Pattern）：同时维护数据库、缓存，二者中的数据存在强一致性；

  • read：使用上面统一的缓存读策略；

  • write：不存在 write cache hit + no-write-allocate。立即写回数据库，并拒绝缓存。清空写这个数据的缓存信息（使用不缓存手段消除数据不一致性，注意保证顺序先更新磁盘再删除缓存）；

    为什么不采用上述的写缓存策略，而是拒绝缓存？因为考虑到多次盲写的问题。

• 读写穿透模式（Read/Write Through Pattern）：视缓存为主要存储手段，二者中的数据也存在强一致性；

  • read：使用上面统一的读策略；
  • write：write-through + write-allocate；

• 异步缓存写入模式（Write Behind Pattern）：针对读写穿透模式的改进，牺牲一部分数据一致性换取更高的吞吐量；

  • read：使用上面统一的读策略；
  • write：write-back（优化，不使用 dirty-bit，而是异步更新到数据库）；

后文将以如何实现读写穿透模式为例，展示代码，同时加入缓存击穿和缓存雪崩等等问题的应对措施。代码将在文末以附录形式呈现。

结论 2：常用的缓存读写策略有很多种，不过依赖它们制定的缓存模式常见的有 3 种，分别是 旁路缓存、读写穿透、异步缓存；

3.2.2 缓存 Evict 策略：以 Redis 为例

不同内存型数据库的缓存淘汰策略不尽相同。下面以 Redis 为例介绍它的 cache evict 方案：

首先，Redis 正常不会主动 evict 数据项，而是先通过数据过期的方式腾出内存空间：

• 过期时间：对每个数据项可以设置 TTL（Time-To-Live），表示数据过期时间。过期的数据自动被清空；
• 定期清理：Redis 可以配置扫描过期数据的频率，扫描过程称为 Garbage Collection（GC）；
• 随机选取：由于 Redis 管理的缓冲区很大，因此每次 GC 一般不会扫描全表，而是随机选取一部分进行回收；
• 惰性删除：某些键值可能概率原因一直无法被选中删除，因此一旦有查询找到该数据，发现该数据过期后立即删除（被动）；

在此基础上，如果：

• 有些键值始终没被查询，且一直没有被随机选取清理（躲过了定期清理和惰性删除）；
• 过多的键值没有设置过期时间；
• 数据工作集（working set）进一步增大；

导致内存空间还是没法及时腾出，那么 Redis 就会采取主动 evict 的方案。

Redis 主动进行 cache evict 时可以配置 8 种策略（注意，下面的策略都是 “没办法通过数据过期腾出空间” 的主动举措）：

• noneviction：缓冲区占满后报错，不会删除任何键值；
• allkeys-lru：对所有缓存键使用 LRU 策略 evict；
• volatile-lru：从设置了过期时间的数据（不一定过期了）中使用 LRU 策略 evict；
• allkeys/volatile-random：对所有缓存键/设置了过期时间的缓存记录使用随机策略 evict；
• volatile-ttl：从设置过期时间的缓存记录中选择剩余存活时间最少的记录 evict；
• allkeys/volatile-lfu：从所有缓存键/设置了过期时间的缓存记录使用 LFU 策略 evict；

LRU: Least Recently Used；

LFU: Least Frequently Used；

结论 3: Redis 对于缓存使用率过高的解决方案是 数据过期 + 主动 evict。其中数据过期依赖 “定时清理” 和 “惰性删除”，主动 evict 依赖 8 种 evict 策略。

3.3 缓存击穿 & 缓存雪崩 Cache Penetration/Avalanche

注意到以上方案，从缓存读写策略、缓存模式，到缓存 evict 策略，全部都没有考虑到一个问题，或者说一类独特的访问 pattern：如果一直查询并不存在的数据会发生什么。

无论按照上面的哪种策略，都会频繁出现 “cache miss - 查找数据库 - 数据库未找到” 这个流程，这会频繁绕过缓存，增大数据库的 disk I/O，影响正常业务逻辑的时延和吞吐量，尤其是大量的 Client 查找一个并不存在的数据的时候，性能影响更为明显。这种现象被称为 “缓存击穿”。

为了解决缓存击穿的问题，可行的解决方案之一是：根据具体业务逻辑指定一个无效值（Invalid Value），一旦出现一次 read cache miss 并且发现数据库未找到的情况，可以在缓存中写入这个无效值，下次 read cache hit 就知道数据库中没有了。

这种解决方案借鉴了 Bloom Filter 的设计思想。

Google 的一篇论文曾经介绍使用跳表 + Bloom Filter 为 Log-Structure Merged Tree 提升查询速度。

但是，除了 “一直查询并不存在的数据”，还有一类情况会引发缓存击穿：某个热点数据过期被清理。

在过期后的短时间内，没有等上缓存恢复就出现大量的对该数据的并发请求。这些请求会 cache miss 并 fallback 到数据库，造成上述问题。

如果更严重一点，假设一批热点数据同时过期，也就是大批数据出现 cache miss，那么大量请求可能造成拒绝查询甚至宕机的后果，这种现象被称为 “缓存雪崩”。

解决这类缓存击穿，以及缓存雪崩的方案之一，无非是：

• 延长热点数据的 TTL（比如每次访问时增加 TTL），或者热点数据永久驻留缓存；
• 批量设置缓存时，在一定时间范围内随机指定 TTL（例如 10~30 分钟内的均匀分布）；

还有一种情况，如果 Redis 出现宕机，内存缓存数据全部丢失，也会出现缓存雪崩的问题，这个时候仅靠以上的应对方案已经不足以解决了。我们需要对 Redis 中的数据进行适当的持久化（“适当” 指同时保证性能）来尽量避免这个情况。

结论 4：“一直查询不存在的数据” 或者 “某个热点数据被清理” 都会造成缓存击穿、“一批热点数据同时过期”、“内存数据库宕机” 都可能造成缓存雪崩。对应的解决方案是 “添加无效值缓存”、“延长热点数据 TTL”、“随机化批量缓存 TTL”，以及 “适当的缓存持久化”。

3.4 缓存持久化：以 Redis 为例

基于上述原因，我们需要对内存型数据库进行适当的缓存持久化。我们仍然以 Redis 为例说明。

Redis 提供了一套缓存持久化方案：RDB；

首先 RDB 支持全量备份，但是如果 Redis 缓存空间很大，一次全量备份刷盘会耗时很久，虽说 NoSQL 不需要支持完整的 ACID 性质，但也会严重影响查询时延和吞吐量，并且可能出现持久化的数据不一致的情况。也就是说：

• RDB 即便支持全量备份也不能太过频繁，对性能影响较大；
• RDB 以分钟级别进行全量备份，可能短时间内会丢失大量新数据（snapshot 数据不一致）；

因此 RDB 全量备份在多数场合下是不划算的。

于是人们借鉴了关系型数据库的 Binary Log 的思想，添加了一种 AOF 机制，采用增量备份来备份缓存数据。

AOF 持久化机制（Append Only File），就是一种增量逻辑备份，向日志文件中以二进制形式写入修改缓存的指令，并且使用了 AOF Buffer 来批量写入以提升效率。

我们注意到 AOF Log 随着时间推移也会越来越大、越来越多，加载和存储效率都不高。一种解决方案是，定时对已有的 AOF Log 进行重写压缩（包括删去无效修改、指令重写等等）和轮替。

此外，为了解决备份时数据不一致现象，Redis 再引入 AOF Rewrite Buffer，可以存放在备份期间修改的数据指令，以便子进程对 AOF Log 重写时加入最新的、遗漏的修改指令，维持了一些数据一致性。

结论 5: Redis 提供了 RDB 全量备份和 AOF 增量备份两种缓存持久化方案，和关系型数据库一样，二者配合使用可以一定程度上解决缓存热身和雪崩问题。

虽然缓存持久化能一定程度上解决缓存雪崩、缓存热身等问题，但是无法提升 Redis 的可用性（availability）。为了实现高可用，我们需要引入 Redis 集群，利用多个物理节点 primary-backup 的方式实现高可用支持。

3.5 内存数据库副本 与 高可用支持

一个经典的架构：primary-backup 架构（旧称 “master-slave” 主从架构），可以作为数据 replicas 的模式，在多种分布式系统上都在使用。

例如，我们可以对 Redis 采用这种架构策略。一个 Redis 节点作为 primary，其他两个 redis 作为 backup server；

其中：

• primary 负责统一写操作，再利用类似 RSM（Replicated State Machine）之类的机制向 backup 发送数据 / 以指令传播的形式同步数据的修改；

• primary 和 backup server 都可以处理读操作，有助于减小 primary 负担。

• primary 维护 Write Ahead Log，在 backup 宕机时能够从 primary 的 WAL 以及自身的 Log 中迅速恢复；

如果考虑到 primary 也可能宕机，可以引入分布式协调者（coordinator），决定谁作为 primary。

这个协调者在 Redis 的术语中被称为 “哨兵”（Sentinel）。

另外，如果还需要保证 coordinator 自身的高可用，可以对 coordinator 进行 replications，可以构建经典 “主从-哨兵” 架构。

为了确保不会因为 network partition 而出现多个 primary（“split-brain problem”），可以将 coordinator 中选举 primary、心跳监测的职责分出给唯一的 view server。第一次 coordinator 接受 client 请求时先询问 view server 关于 primary 的信息，然后再向 primary 给出修改请求。

最终，如果还需要保证 view server 的高可用以及数据一致性，还可以将轻量的 view server 进行 replications 并交由 Paxos 或者 ZooKeeper 或者 KRaft 来做分布式协调管理。

一般为了架构简单起见，可以不使用 view server，直接在 sentinels 中引入 primary 投票机制（主观下线、客观下线），粗略地模拟 Paxos 的一致性协调管理。

注：选拔 Primary 的标准可以考虑硬件配置、断开 primary 连接的时间长短等等信息来指定优先级，从而选择。

3.6 内存数据库集群

前面的例子虽然介绍了，内存数据库可以通过建立 replicas 来提升高可用性，但是单个物理节点的存储量总有上限。

在极大的 data working set 场景下，我们可能需要通过集群来实现更大规模数据的缓存。

首先需要解决集群后的缓存位置问题。大多数内存型数据库采用了 一致性哈希（Consistent Hashing）思想：

• 我们先按照常用的 hash 算法将 Key hash 到 $0\sim2^{32}-1$ 个桶中，并且把它们想象成一个环结构；

• 将机器的唯一标识（例如 MAC/IP/HOSTNAME 等信息）以及需要缓存的 KV 都 hash 到环上；

• 于是就能判断信息究竟放在哪一台服务器上了：按顺时针方向，所有对象 hash 的位置距离最近的机器 hash 点就是要存的机器，如下图所示：

  

• 当有机器（t4）加入分布式集群后，t3 - t4 间的缓存将转移至 t4 上（少量数据交换）；

  反之，有机器（t4）从分布式集群中离线后，t3 - t4 间的缓存将重新转移至 t2；

此外，hash 的位置可以根据机器的硬件承载能力适当调整。调整方法可以借助下文介绍的 virtual nodes 来完成。

这样的方案能在分布式场景下尽可能减少缓存失效和变动的比例；

但这种方案仍然存在问题：当集群中的节点数量较少时，可能会出现节点在哈希空间中分布不平衡的问题（hash 环的倾斜和负载不均），甚至引发雪崩问题（最多数据的 A 故障，全转移给 B，然后 B 故障，并重复下去，造成整个分布式集群崩溃）。

解决 hash 环倾斜的问题的方案之一就是引入 “虚拟节点”（相当于给机器 hash 点创建 “软链接”），将 virtual nodes 和 real nodes 的映射关系记录在 Hash Ring 中；

上面解决方案的具体实现被称为 “Chord 算法”；

我们现在继续以 Redis 为例。在 Redis 集群中，实现的方法略有差别，它一般创建一个超大的数组，例如 struct clusterNode *slots[]，规定哪些 Key Hash 值的范围由指定的 Redis 实例负责。并且只有数组中的所有 entries 都被分配给一个 Redis 实例，整个集群才能认为是上线状态。

因此真正在 Redis 集群中的查询动作通常会先检查数据是否缓存在当前结点中，如果不是则响应 MOVE(IP:PORT) 来指示 client 应该对哪个实例请求该数据。

现在，我们把集群配合上之前提到的 replicas，形成经典的 “三主三从 + 哨兵 架构”，以此来提升集群的整体可用性：

Appendix: Redis + Spring Boot Example

下面是使用 Spring Boot 框架的 Redis 单物理节点代码示例。

public interface CacheService {
    /**
     * Check if the cache(redis) is used and connected.
     */
    boolean isAlive();

    /**
     * Get the data from the cache and database.
     * If cache misses, it will use fallback function to get the current data and cache then.
     *
     * @param keyPrefix The key prefix of the stored items
     * @param id The key ID to identify the specific items from a group of stored items
     * @param type The type of stored items
     * @param fallback The fallback database query function used when cache misses
     * @param time The TTL for the specific cached item
     * @param unit The time unit of the TTL
     *
     * @return Always the specific stored item consistent with database
     */
    <R, ID>Optional<R> queryAndCache(
            String keyPrefix, ID id, TypeReference<R> type,
            Function<ID, Optional<R>> fallback,
            Long time, TimeUnit unit
    );

    /**
     * Get the data from the cache and database.
     *
     * @see CacheService#queryAndCache(String, Object, TypeReference, Function, Long, TimeUnit)
     */
    <R, ID>Optional<R> queryAndCache(
            String keyPrefix, ID id, Class<R> type,
            Function<ID, Optional<R>> fallback,
            Long time, TimeUnit unit
    );

    /**
     * Get a bunch of data from the cache.
     * If cache misses, it will use fallback function to get the current data and cache then.
     *
     * @return The returned list is ordered.
     *  But R could be null if ID is not found in either cache or database.
     *
     * @see CacheService#queryAndCache(String, Object, TypeReference, Function, Long, TimeUnit)
     */
    <R, ID>List<R> queryInBatch(
            String keyPrefix, List<ID> ids, Class<R> type,
            Function<ID, Optional<R>> fallback,
            Long time, TimeUnit unit
    );

    /**
     * Only query the data from the cache.
     *
     * @apiNote [WARNING] We don't suppose you to use this method for the following reasons:
     *  1. It will not cache the results, which may cause cache penetration if you use it heavily;
     *  2. The result may be not consistent with database, for it only query on redis for existence.
     *  So you are advised to:
     *  - Put the result after you finally get the data;
     *  - Retry the query on database if you cannot find it
     *    (because it doesn't mean that the data don't exist in database).
     *
     * @see CacheService#put(String, Object, Object, Long, TimeUnit)
     */
    <R, ID>Optional<R> _query(String keyPrefix, ID id, TypeReference<R> type);

    /**
     * Put the specific item into the cache.
     * It will overwrite the original item with the same key (if exists).
     *
     * @implNote Use asynchronous operation to improve the throughput.
     */
    <R, ID>void put(String keyPrefix, ID id, R val, Long time, TimeUnit unit);
    /**
     * Put cache items in batch.
     * @apiNote We don't use TTL here to avoid a bunch of data use a same TTL,
     *  which may cause cache avalanche.
     * @implNote
     *  1. Use discrete & random TTL for each item.
     *  2. Use asynchronous operation to improve the throughput.
     */
    <R, ID>void putInBatch(String keyPrefix, List<ID> ids, List<R> values);

    /**
     * Invalidate the specific item from the cache.
     *
     * @implNote It will actually invalid the cached item by putting an invalid value to it.
     */
    <ID>Boolean invalidate(String keyPrefix, ID id);
    /**
     * Invalidate the stored items with the same key prefix.
     *
     * @implNote It will actually invalid the cached item by putting an invalid value to it.
     */
    Long invalidateByKeyPrefix(String keyPrefix);

    /**
     * Purge the stored item from the cache.
     */
    <ID>Boolean remove(String keyPrefix, ID id);
}

@Slf4j
@Service
@RequiredArgsConstructor
public class CacheServiceImpl implements CacheService {

    private final StringRedisTemplate template;
    private final ObjectMapper mapper;

    @Override
    public boolean isAlive() {
        try {
            RedisConnectionFactory factory = template.getConnectionFactory();
            return factory != null && !factory.getConnection().isClosed();
        } catch (Exception exception) {
            log.warn("Exception occurs when checking redis connection: {}", exception.getMessage());
            return false;
        }
    }

    @Override
    public <R, ID> Optional<R> queryAndCache(
            String keyPrefix, ID id, Class<R> type,
            Function<ID, Optional<R>> fallback,
            Long time, TimeUnit unit
    ) {
        TypeReference<R> typeReference = new TypeReference<>() {
            @Override
            public Type getType() {
                return type;
            }
        };
        return this.queryAndCache(keyPrefix, id, typeReference, fallback, time, unit);
    }

    @Override
    public <R, ID> Optional<R> queryAndCache(
            String keyPrefix, ID id, TypeReference<R> type,
            Function<ID, Optional<R>> fallback,
            Long time, TimeUnit unit
    ) {
        String key = keyPrefix + id.toString();
        // Stage 1. lookup in redis
        if (this.isAlive()) try {
            String jsonResult;
            // maybe fault
            jsonResult = template.opsForValue().get(key);
            if (jsonResult != null) {
                log.debug("Cache hit when fetching key: {}", key);
                // has non-blank value in redis?
                if (jsonResult.equals(Cache.REDIS_INVALID_VALUE)) {
                    // empty value indicates non-existence.
                    return Optional.empty();
                } else {
                    // TODO: cache hit 后是否更新 TTL?
                    R v = mapper.readValue(jsonResult, type);
                    return Optional.of(v);
                }
            }
        } catch (Exception exception) {
            log.warn("Failed to fetch data from redis due to: {}",
                    exception.getMessage());
        }
        // Stage 2. fallback to database
        Optional<R> dbResult = fallback.apply(id);
        log.debug("Cache miss when fetching key: {}. Cache it now", key);
        // we deliberately write empty value to indicate non-existence.
        this.put(keyPrefix, id, dbResult.isPresent() ? dbResult.get() : Cache.REDIS_INVALID_VALUE, time, unit);
        return dbResult;
    }

    @Override
    public <R, ID> List<R> queryInBatch(
            String keyPrefix, List<ID> ids, Class<R> type,
            Function<ID, Optional<R>> fallback,
            Long time, TimeUnit unit
    ) {
        List<R> result = new ArrayList<>();
        if (this.isAlive()) {
            // query in redis first
            for (ID id: ids) {
                R curRes;
                String jsonResult;
                String key = keyPrefix + id.toString();
                try {
                    // maybe fault
                    jsonResult = template.opsForValue().get(key);
                    if (jsonResult != null) {
                        log.debug("Cache hit when doing batch query: {}", key);
                        curRes = mapper.readValue(jsonResult, type);
                        if (!curRes.equals(Cache.REDIS_INVALID_VALUE)) {
                            result.addLast(curRes);
                        } else {
                            // not exist
                            result.addLast(null);
                        }
                        continue;
                    }
                } catch (Exception exception) {
                    log.warn("Batch query failed: {}. Skip current one.",
                            exception.getMessage());
                }

                // fallback to database
                curRes = fallback.apply(id).orElse(null);
                log.debug("Cache miss when doing batch query: {}. Cache it now", key);
                try {
                    // set cache
                    template.opsForValue().set(
                            key, curRes == null ? Cache.REDIS_INVALID_VALUE
                                    : mapper.writeValueAsString(curRes),
                            time, unit
                    );
                } catch (Exception exception) {
                    log.warn("Cache data from database failed when batching query: {}. Skip current one.",
                            exception.getMessage());
                }
                result.addLast(curRes);
            }
        } else {
            // query in database only
            for (ID id: ids) {
                result.addLast(fallback.apply(id).orElse(null));
            }
        }
        return result;
    }

    @Override
    public <R, ID> Optional<R> _query(String keyPrefix, ID id, TypeReference<R> type) {
        String key = keyPrefix + id.toString();
        if (this.isAlive()) try {
            String jsonResult;
            // may be fault even we have already checked the liveness
            jsonResult = template.opsForValue().get(key);
            if (jsonResult != null) {
                log.debug("Cache hit when fetching key with _query: {}", key);
                // has non-blank value in redis?
                if (!jsonResult.equals(Cache.REDIS_INVALID_VALUE)) {
                    return Optional.of(mapper.readValue(jsonResult, type));
                }
                // Sadly we cannot use empty value to indicate non-existence. :(
            }
        } catch (Exception exception) {
            log.warn("Failed to do simple query on redis due to: {}",
                    exception.getMessage());
        }
        log.debug("Cache miss when fetching key with _query: {}", key);
        return Optional.empty();
    }

    @Override
    public <R, ID> void put(String keyPrefix, ID id, R val, Long time, TimeUnit unit) {
        try {
            if (this.isAlive()) {
                template.opsForValue().set(
                        keyPrefix + id.toString(),
                        mapper.writeValueAsString(val),
                        time, unit
                );
                log.debug("Set cache for key: {}; TTL: {} {}", keyPrefix + id, time, unit);
            } else {
                log.warn("Redis not alive. Skip writing to it.");
            }
        } catch (Exception exception) {
            log.warn("Exception occurs when setting value for redis: {}",
                    exception.getMessage());
        }
    }

    @Override
    public <R, ID> void putInBatch(String keyPrefix, List<ID> ids, List<R> values) {
        Random randomGen = new Random();
        try {
            assert ids.size() == values.size();
            if (this.isAlive()) {
                int idx = 0;
                for (ID id: ids) {
                    // center: REDIS_DEFAULT_CACHE_TTL
                    // [center - RANGE, center + RANGE]
                    long randomTime = randomGen.nextLong(2 * Cache.REDIS_BATCH_RANDOM_TTL_RANGE)
                            + Cache.REDIS_DEFAULT_CACHE_TTL - Cache.REDIS_BATCH_RANDOM_TTL_RANGE;
                    template.opsForValue().set(
                            keyPrefix + id.toString(),
                            mapper.writeValueAsString(values.get(idx)),
                            randomTime, Cache.REDIS_TTL_UNIT
                    );
                    log.debug("Set cache for key in batch: {}; TTL: {} {}",
                            keyPrefix + id, randomTime, Cache.REDIS_TTL_UNIT);
                    ++idx;
                }
            } else {
                log.warn("Redis not alive. Skip batch process.");
            }
        } catch (Exception exception) {
            log.warn("Exception occurs when doing batch process for redis: {}",
                    exception.getMessage());
        }
    }

    @Override
    public <ID> Boolean invalidate(String keyPrefix, ID id) {
        if (!this.isAlive()) {
            log.warn("Redis not alive. Skip invalidating process.");
            return false;
        }
        try {
            // return template.delete(key);
            // no delete but set empty value
            template.opsForValue().set(
                    keyPrefix + id.toString(), Cache.REDIS_INVALID_VALUE,
                    Cache.REDIS_DEFAULT_CACHE_TTL, Cache.REDIS_TTL_UNIT
            );
            log.debug("Invalid cache for key: {}; TTL: {} {}",
                    keyPrefix + id, Cache.REDIS_DEFAULT_CACHE_TTL, Cache.REDIS_TTL_UNIT);
            return true;
        } catch (Exception exception) {
            log.warn("Exception occurs when invalidating value from redis: {}",
                    exception.getMessage());
        }
        return false;
    }

    @Override
    public Long invalidateByKeyPrefix(String keyPrefix) {
        long res = 0L;
        try {
            Set<String> keys = template.keys(keyPrefix);
            // if (keys != null) res = template.delete(keys);
            if (keys != null) {
                res = keys.size();
                List<String> empties = Collections.nCopies(keys.size(), Cache.REDIS_INVALID_VALUE);
                log.debug("--- Start invalid cache by key prefix ---");
                // tricky point: use empty key prefix
                this.putInBatch("", keys.stream().toList(), empties);
                log.debug("--- Successfully invalid cache for keys: {} ---", keys);
            }
        } catch (Exception exception) {
            log.warn("Exception occurs when invalidateByPrefix: {}",
                    exception.getMessage());
        }
        return res;
    }

    @Override
    public <ID> Boolean remove(String keyPrefix, ID id) {
        if (this.isAlive()) {
            try {
                log.debug("Remove cache for key: {}", keyPrefix + id.toString());
                return template.delete(keyPrefix + id.toString());
            } catch (Exception exception) {
                log.warn("Exception occurs when removing cache: {}",
                        exception.getMessage());
            }
        } else log.warn("Redis not alive. Skip removing process.");
        return false;
    }
}