竹子

Appearance

Redis ​

一、为什么需要 Redis？ ​

传统的磁盘数据库（如 MySQL）虽然持久化可靠，但面对高并发秒杀、热点数据、实时排行榜等场景，磁盘 IO 成为瓶颈。Redis 应运而生，核心优势：

维度传统数据库Redis
存储介质磁盘内存
读写速度毫秒级微秒级（10W+ QPS）
数据结构表（行列）String、Hash、List、Set、Sorted Set 等
典型场景持久化存储缓存、计数器、分布式锁、消息队列

为什么 Redis 单线程还这么快？

  1. 纯内存操作：绝大部分操作在内存中完成，磁盘 IO 只在持久化时发生
  2. IO 多路复用：单线程通过 epoll/kqueue 机制同时监听多个客户端连接，用事件驱动非阻塞 IO
  3. 单线程避免锁竞争：不存在多线程上下文切换和锁开销，但 Redis 6.0+ 引入多线程 IO（网络读写用多线程，命令执行仍单线程）
  4. 高效的数据结构：SDS（简单动态字符串）、ziplist（压缩列表）、skiplist（跳表）等，极致优化内存和性能

二、五种核心数据类型 ​

2.1 String（字符串） ​

底层： SDS（简单动态字符串），二进制安全，最大 512MB。

常用命令：

bash
SET key value [EX seconds] [NX|XX]  # 设置值，可带过期时间和存在判断
GET key                              # 获取值
SETNX key value                      # 不存在才设置（分布式锁基础）
INCR key                             # 原子递增（计数器）
DECR key                             # 原子递减
MSET key1 v1 key2 v2                 # 批量设置
MGET key1 key2                       # 批量获取
SETEX key seconds value              # 设置值并指定过期时间

使用场景：

java
// 1. 分布式锁
String lockKey = "lock:order:" + orderId;
Boolean locked = redisTemplate.opsForValue()
    .setIfAbsent(lockKey, "1", 30, TimeUnit.SECONDS);
if (Boolean.TRUE.equals(locked)) {
    try {
        // 业务逻辑
    } finally {
        // Lua 脚本保证原子性释放
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                        "return redis.call('del', KEYS[1]) else return 0 end";
        redisTemplate.execute(
            new DefaultRedisScript<>(script, Long.class), 
            Collections.singletonList(lockKey), "1");
    }
}

// 2. 计数器（限流、点赞数）
Long count = redisTemplate.opsForValue().increment("article:like:" + articleId);

// 3. 缓存序列化对象
User user = new User(1L, "张三", 25);
redisTemplate.opsForValue().set("user:" + user.getId(), 
    JSON.toJSONString(user), 1, TimeUnit.HOURS);

2.2 Hash（哈希） ​

底层： ziplist（元素少时）或 hashtable（元素多时）。

常用命令：

bash
HSET key field value [field value ...]   # 设置字段
HGET key field                            # 获取字段
HMSET key field1 v1 field2 v2             # 批量设置
HMGET key field1 field2                   # 批量获取
HGETALL key                               # 获取所有字段（慎用，大 key 阻塞）
HINCRBY key field increment               # 原子递增字段值
HDEL key field                            # 删除字段
HEXISTS key field                         # 判断字段是否存在

使用场景：

java
// 1. 存储对象（比 String 序列化更灵活，可部分更新）
redisTemplate.opsForHash().put("user:1001", "name", "张三");
redisTemplate.opsForHash().put("user:1001", "age", "25");
redisTemplate.opsForHash().increment("user:1001", "age", 1); // 只更新年龄

// 2. 购物车（用户ID -> 商品ID:数量）
redisTemplate.opsForHash().put("cart:user:1001", "product:2001", "2");
redisTemplate.opsForHash().put("cart:user:1001", "product:2002", "1");

Hash vs String 存储对象对比：

维度String (JSON)Hash
读写整个对象方便需要 HGETALL
部分字段更新需要序列化反序列化直接 HGET/HSET
内存占用一个 key 一个 value一个 key 多个 field，更省内存
过期只能给整个 key 设过期只能给整个 key 设过期

2.3 List（列表） ​

底层： quicklist（Redis 3.2+，由 linkedlist 和 ziplist 混合组成）。

常用命令：

bash
LPUSH key value [value ...]    # 左侧插入（头部）
RPUSH key value [value ...]    # 右侧插入（尾部）
LPOP key                       # 左侧弹出
RPOP key                       # 右侧弹出
LRANGE key start stop          # 范围查询（O(n)）
LLEN key                       # 长度
LINDEX key index               # 按索引获取
BRPOP key timeout              # 阻塞弹出（消息队列核心）

使用场景：

java
// 1. 消息队列（生产者-消费者）
// 生产者
redisTemplate.opsForList().leftPush("task:queue", taskJson);
// 消费者（阻塞等待）
String task = (String) redisTemplate.opsForList()
    .rightPop("task:queue", 0, TimeUnit.SECONDS);

// 2. 最新消息列表（时间线）
// 发布新消息
redisTemplate.opsForList().leftPush("timeline:user:1001", messageJson);
// 只保留最近 100 条
redisTemplate.opsForList().trim("timeline:user:1001", 0, 99);
// 获取前 10 条
List<Object> timeline = redisTemplate.opsForList()
    .range("timeline:user:1001", 0, 9);

// 3. 栈（LPUSH + LPOP）

2.4 Set（集合） ​

底层： intset（整数且元素少时）或 hashtable（其他情况）。

常用命令：

bash
SADD key member [member ...]       # 添加元素
SREM key member [member ...]       # 删除元素
SISMEMBER key member               # 判断是否存在
SCARD key                          # 元素个数
SMEMBERS key                       # 获取所有元素（慎用）
SINTER key1 key2                   # 交集
SUNION key1 key2                   # 并集
SDIFF key1 key2                    # 差集
SRANDMEMBER key [count]            # 随机获取元素
SPOP key [count]                   # 随机弹出

使用场景：

java
// 1. 共同好友（交集）
Set<Object> commonFriends = redisTemplate.opsForSet()
    .intersect("friends:user:1001", "friends:user:1002");

// 2. 抽奖（随机弹出）
// 参与抽奖
redisTemplate.opsForSet().add("lottery:activity:001", userId);
// 随机抽取中奖者
Object winner = redisTemplate.opsForSet().pop("lottery:activity:001");

// 3. 点赞去重（集合不允许重复）
Long result = redisTemplate.opsForSet().add("like:article:5001", userId);
// result == 1 表示点赞成功，result == 0 表示已经点过赞

// 4. 标签系统
redisTemplate.opsForSet().add("user:1001:tags", "Java", "Spring", "Redis");

2.5 Sorted Set（有序集合） ​

底层： ziplist（元素少时）或 skiplist + hashtable（元素多时）。跳表实现 O(log n) 的插入和范围查询。

常用命令：

bash
ZADD key score member [score member ...]   # 添加元素
ZRANGE key start stop [WITHSCORES]         # 按score升序范围查询
ZREVRANGE key start stop [WITHSCORES]      # 按score降序范围查询
ZRANK key member                           # 获取排名（升序，从0开始）
ZREVRANK key member                        # 获取排名（降序）
ZSCORE key member                          # 获取分数
ZREM key member                            # 删除元素
ZINCRBY key increment member               # 增加分数
ZRANGEBYSCORE key min max                  # 按分数范围查询
ZCOUNT key min max                         # 分数范围内元素个数

使用场景：

java
// 1. 排行榜（实时更新排名）
// 用户得分
redisTemplate.opsForZSet().add("rank:game", userId, score);
// 增加分数
redisTemplate.opsForZSet().incrementScore("rank:game", userId, 10);
// Top 10 排行榜
Set<ZSetOperations.TypedTuple<Object>> top10 = redisTemplate.opsForZSet()
    .reverseRangeWithScores("rank:game", 0, 9);
// 用户排名
Long rank = redisTemplate.opsForZSet().reverseRank("rank:game", userId);

// 2. 延迟队列（score 为执行时间戳）
long executeTime = System.currentTimeMillis() + 30 * 60 * 1000; // 30分钟后
redisTemplate.opsForZSet().add("delay:queue", taskId, executeTime);
// 定时任务拉取到期任务
Set<Object> tasks = redisTemplate.opsForZSet()
    .rangeByScore("delay:queue", 0, System.currentTimeMillis());

// 3. 带权重的标签
redisTemplate.opsForZSet().add("article:5001:tags", "Java", 10);
redisTemplate.opsForZSet().add("article:5001:tags", "Redis", 8);

三、高级数据类型 ​

3.1 Bitmap（位图） ​

底层是 String，按位操作，适合二值状态统计。

bash
SETBIT key offset value     # 设置位值
GETBIT key offset           # 获取位值
BITCOUNT key [start end]    # 统计 1 的个数
BITOP AND/OR/XOR dest key1 key2   # 位运算
BITPOS key bit [start end]  # 查找第一个 0/1 的位置
java
// 用户签到（一年 365 天，一个用户只需 46 字节）
int dayOfYear = 185; // 一年的第几天
redisTemplate.opsForValue().setBit("sign:user:1001:2025", dayOfYear, true);
// 统计签到天数
Long signDays = redisTemplate.execute(
    (RedisCallback<Long>) connection ->
        connection.bitCount("sign:user:1001:2025".getBytes()));

3.2 HyperLogLog（基数统计） ​

基于概率算法，统计不重复元素个数，固定 12KB 内存，标准误差 0.81%。

bash
PFADD key element [element ...]   # 添加元素
PFCOUNT key [key ...]             # 统计基数
PFMERGE destkey sourcekey [...]   # 合并多个 HyperLogLog
java
// UV 统计（1000 万 UV 只需 12KB）
redisTemplate.opsForHyperLogLog().add("uv:page:home", userId);
Long uv = redisTemplate.opsForHyperLogLog().size("uv:page:home");

3.3 Geo（地理位置） ​

基于 Sorted Set 实现，支持经纬度距离计算。

bash
GEOADD key longitude latitude member   # 添加位置
GEOPOS key member                      # 获取位置
GEODIST key member1 member2 [unit]     # 计算距离
GEORADIUS key lon lat radius unit      # 半径内查询
GEORADIUSBYMEMBER key member radius    # 以成员为圆心半径查询
java
// 附近的人
redisTemplate.opsForGeo().add("geo:shop", 
    new Point(116.404, 39.915), "shop:001");
redisTemplate.opsForGeo().add("geo:shop", 
    new Point(116.418, 39.921), "shop:002");
// 查询 shop:001 附近 5km 的店铺
Distance distance = new Distance(5, Metrics.KILOMETERS);
GeoResults<GeoLocation<Object>> nearby = redisTemplate.opsForGeo()
    .radius("geo:shop", "shop:001", distance);

3.4 Stream（流） ​

Redis 5.0 引入，真正的消息队列数据结构，支持消费者组、消息持久化、ACK 确认。

bash
XADD stream key * field value    # 添加消息
XREAD COUNT n STREAMS key id     # 读取消息
XGROUP CREATE key group id       # 创建消费者组
XREADGROUP GROUP group consumer  # 消费者组读取
XACK key group id                # 确认消息
java
// 生产者
Map<String, Object> message = Map.of("orderId", "1001", "amount", "99.9");
redisTemplate.opsForStream().add("stream:order", message);

// 消费者组消费
List<MapRecord<String, Object, Object>> messages = redisTemplate.opsForStream()
    .read(Consumer.from("group:order", "consumer:1"),
          StreamReadOptions.empty().count(5),
          StreamOffset.create("stream:order", ReadOffset.lastConsumed()));

四、持久化机制 ​

4.1 RDB（快照持久化） ​

原理： 在指定时间间隔内，将内存中的数据集快照写入磁盘。Redis 通过 fork 子进程执行写入，父进程继续处理请求（Copy-On-Write）。

触发方式：

bash
# 自动触发（redis.conf）
save 900 1    # 900 秒内至少 1 个 key 变化
save 300 10   # 300 秒内至少 10 个 key 变化
save 60 10000 # 60 秒内至少 10000 个 key 变化

# 手动触发
SAVE     # 主进程执行，阻塞所有请求
BGSAVE   # 子进程执行，不阻塞

RDB 优缺点：

优点缺点
文件紧凑，适合备份和灾难恢复两次快照之间的数据可能丢失
恢复大数据集速度快fork 子进程时内存开销大（COW）
对性能影响小（子进程完成）不适合实时持久化

4.2 AOF（追加文件持久化） ​

原理： 将每条写命令追加到 AOF 文件末尾，重启时回放 AOF 文件恢复数据。

同步策略：

bash
appendfsync always   # 每条命令都 fsync（最安全，最慢）
appendfsync everysec # 每秒 fsync 一次（默认，推荐，最多丢 1 秒数据）
appendfsync no       # 由操作系统决定（最快，不安全）

AOF 重写： 当 AOF 文件过大时，fork 子进程创建新 AOF 文件，用最少命令重建当前数据集，压缩文件体积。

bash
# 自动重写触发条件
auto-aof-rewrite-percentage 100  # 当前 AOF 文件是上次重写后的 2 倍
auto-aof-rewrite-min-size 64mb   # AOF 文件至少 64MB

AOF 优缺点：

优点缺点
数据安全性高，最多丢 1 秒文件体积比 RDB 大
自动重写压缩文件恢复速度比 RDB 慢
文件可读，可手动修复写 QPS 高时 AOF 写入压力大

4.3 混合持久化（Redis 4.0+） ​

原理： AOF 重写时，将重写这一刻的内存数据做 RDB 快照写入 AOF 文件开头，后续增量命令以 AOF 格式追加。兼顾恢复速度和数据安全性。

bash
aof-use-rdb-preamble yes  # 开启混合持久化

三种持久化策略对比：

策略数据安全恢复速度文件大小性能影响
仅 RDB低（可能丢失数分钟）快小小
仅 AOF高（最多 1 秒）慢大中
混合高快中中

生产环境推荐： 混合持久化 + RDB 定期备份。

五、过期策略与内存淘汰 ​

5.1 过期键删除策略 ​

Redis 采用惰性删除 + 定期删除的组合策略：

策略原理优点缺点
定时删除创建定时器，到期立即删除内存友好CPU 不友好（大量 key 到期时严重占用 CPU）
惰性删除访问 key 时检查是否过期CPU 友好内存不友好（过期 key 不访问就永远不删）
定期删除每隔一段时间扫描一批 key，删除过期 key折中方案需要合理设置扫描频率和数量

Redis 实现细节：

bash
# 定期删除：每秒执行 10 次（默认 hz=10）
hz 10  # 删除频率，可调高到 100
# 每次随机抽取 20 个 key，过期比例超过 25% 则继续抽取

5.2 内存淘汰策略（8 种） ​

当内存达到 maxmemory 上限时，Redis 根据配置的策略淘汰 key：

策略说明适用场景
noeviction不淘汰，写请求报错（默认）不允许数据丢失的场景
allkeys-lru所有 key 中淘汰最近最少使用的缓存场景（推荐）
allkeys-lfu所有 key 中淘汰最不频繁使用的缓存场景，热点数据保护
allkeys-random所有 key 中随机淘汰不推荐
volatile-lru有过期时间的 key 中淘汰 LRU部分 key 需持久化
volatile-lfu有过期时间的 key 中淘汰 LFU部分 key 需持久化
volatile-ttl有过期时间的 key 中淘汰 TTL 最短的即将过期的优先淘汰
volatile-random有过期时间的 key 中随机淘汰不推荐

LRU vs LFU：

  • LRU（Least Recently Used）： 最近最少使用。Redis 采用近似 LRU，随机采样 5 个 key（可配置），淘汰其中最久未访问的。
  • LFU（Least Frequently Used）： 最不频繁使用。Redis 4.0 引入，通过访问频率计数器，淘汰使用频率最低的 key，避免偶发访问挤掉热点数据。
bash
maxmemory 4gb                              # 最大内存
maxmemory-policy allkeys-lru               # 淘汰策略
maxmemory-samples 10                       # LRU 采样数（越大越精确，但更耗 CPU）

六、事务 ​

6.1 基本事务 ​

Redis 事务保证命令的顺序执行和隔离性，但不支持回滚（语法错误除外）。

bash
MULTI           # 开始事务
SET k1 v1
SET k2 v2
INCR counter
EXEC            # 执行事务（所有命令一起执行）
DISCARD         # 放弃事务

事务特点：

特性说明
原子性命令入队时不执行，EXEC 时一次性执行，中间不会被其他客户端命令插入
不支持回滚某条命令执行失败，其他命令继续执行（语法错误会全部失败）
隔离性EXEC 执行期间，其他客户端命令不会插入

6.2 乐观锁（WATCH） ​

通过 WATCH 实现 CAS（Compare-And-Swap）乐观锁：

java
// 使用 RedisTemplate 实现乐观锁
redisTemplate.execute(new SessionCallback<Object>() {
    @Override
    public Object execute(RedisOperations operations) {
        operations.watch("balance:user:1001");  // 监控 key
        String balance = (String) operations.opsForValue()
            .get("balance:user:1001");
        int newBalance = Integer.parseInt(balance) - 100;
        if (newBalance < 0) {
            operations.unwatch();
            return "余额不足";
        }
        operations.multi();  // 开启事务
        operations.opsForValue().set("balance:user:1001", 
            String.valueOf(newBalance));
        List<Object> result = operations.exec();  // 执行事务
        // result 为 null 表示 key 被其他客户端修改过，事务失败
        return result != null ? "扣款成功" : "重试";
    }
});

七、Pipeline（管道） ​

7.1 为什么需要 Pipeline？ ​

普通模式下，每个 Redis 命令都需要一次网络往返（RTT）。1000 条命令就是 1000 次 RTT，延迟极高。Pipeline 将多条命令打包发送，一次性返回所有结果，大幅减少 RTT。

java
// 不使用 Pipeline：1000 次 RTT
for (int i = 0; i < 1000; i++) {
    redisTemplate.opsForValue().set("key:" + i, "value:" + i);
}

// 使用 Pipeline：1 次 RTT
redisTemplate.executePipelined(new SessionCallback<Object>() {
    @Override
    public Object execute(RedisOperations operations) {
        for (int i = 0; i < 1000; i++) {
            operations.opsForValue().set("key:" + i, "value:" + i);
        }
        return null;
    }
});

Pipeline vs 事务：

维度Pipeline事务（MULTI/EXEC）
原子性不保证（命令可能被其他客户端插入）保证
返回结果每条命令独立返回结果EXEC 后才返回所有结果
使用场景批量操作，性能优化需要原子性的操作

八、发布订阅 ​

bash
SUBSCRIBE channel [channel ...]     # 订阅频道
PUBLISH channel message             # 发布消息
PSUBSCRIBE pattern [pattern ...]    # 模式订阅（通配符）
UNSUBSCRIBE [channel ...]           # 取消订阅
java
// 配置消息监听器
@Configuration
public class RedisPubSubConfig {
    @Bean
    public RedisMessageListenerContainer container(
            RedisConnectionFactory factory,
            MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = 
            new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener(listenerAdapter, 
            new PatternTopic("order:*"));
        return container;
    }
}

// 消息处理器
@Component
public class OrderMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String channel = new String(message.getChannel());
        String body = new String(message.getBody());
        System.out.println("收到消息：" + channel + " - " + body);
    }
}

// 发布消息
redisTemplate.convertAndSend("order:create", orderJson);

注意： Redis Pub/Sub 消息不会持久化，消费者离线期间的消息会丢失。需要可靠消息的场景应使用 Stream 或专业 MQ（RocketMQ、Kafka）。

九、主从复制 ​

9.1 架构 ​

一个 Master 可以有多个 Slave，Slave 还可以有下级 Slave。Master 负责写，Slave 负责读，实现读写分离。

         ┌─────────┐
         │  Master │  ← 写请求
         └────┬────┘
       ┌──────┼──────┐
       ▼      ▼      ▼
   ┌─────┐┌─────┐┌─────┐
   │Slave││Slave││Slave│  ← 读请求
   └─────┘└─────┘└─────┘

9.2 复制流程 ​

全量同步（首次连接）：

  1. Slave 发送 PSYNC ? -1（首次同步）
  2. Master 执行 BGSAVE 生成 RDB 快照，同时记录 RDB 生成期间的写命令到 replication buffer
  3. Master 将 RDB 文件发送给 Slave
  4. Slave 清空旧数据，加载 RDB
  5. Master 将 buffer 中的增量命令发送给 Slave，Slave 执行

增量同步（断开重连）：

  1. Slave 发送 PSYNC <replication_id> <offset>（带上上次同步的偏移量）
  2. Master 检查 repl_backlog（复制积压缓冲区）中是否有该偏移量之后的数据
  3. 如果有：发送增量数据（部分重同步）
  4. 如果没有：触发全量同步

关键配置：

bash
# Master 配置
repl-backlog-size 1mb          # 复制积压缓冲区大小（越大越能支持长时间断线）
repl-backlog-ttl 3600          # 缓冲区存活时间

# Slave 配置
slaveof <masterip> <masterport>  # 或 replicaof
slave-read-only yes               # 从节点只读

十、Sentinel（哨兵） ​

10.1 为什么需要 Sentinel？ ​

主从复制解决了读写分离和备份问题，但 Master 故障时需要手动切换。Sentinel 实现自动故障转移，是高可用的核心组件。

10.2 Sentinel 工作流程 ​

         ┌──────────────────────────┐
         │     Sentinel 集群（至少3个） │
         │  ┌──────┐ ┌──────┐ ┌──────┐│
         │  │Sent-1│ │Sent-2│ │Sent-3││
         │  └──────┘ └──────┘ └──────┘│
         └──────────┬───────────────┘
                    │ 监控
         ┌──────────▼──────────┐
         │       Master        │
         └──────────┬──────────┘
              ┌─────┴─────┐
              ▼           ▼
          ┌──────┐    ┌──────┐
          │Slave1│    │Slave2│
          └──────┘    └──────┘

故障转移流程：

  1. 主观下线（SDOWN）： 单个 Sentinel 认为 Master 不可达（ping 超时）
  2. 客观下线（ODOWN）： 超过 quorum 个 Sentinel 都认为 Master 不可达，Sentinel 之间通过投票确认
  3. 哨兵领导者选举： 通过 Raft 算法选举一个 Sentinel 作为故障转移执行者
  4. 选择新 Master： 根据优先级、复制偏移量、runid 选择最优 Slave
  5. 故障转移： 将选中的 Slave 提升为 Master，其他 Slave 指向新 Master
  6. 通知： 通过 Pub/Sub 通知客户端新 Master 地址

关键配置：

bash
# sentinel.conf
sentinel monitor mymaster 127.0.0.1 6379 2   # quorum=2
sentinel down-after-milliseconds mymaster 30000  # 30 秒无响应判为 SDOWN
sentinel parallel-syncs mymaster 1              # 故障转移时最多 1 个 Slave 同时同步
sentinel failover-timeout mymaster 180000       # 故障转移超时时间

10.3 Spring Boot 集成 Sentinel ​

yaml
spring:
  redis:
    sentinel:
      master: mymaster
      nodes:
        - sentinel1:26379
        - sentinel2:26379
        - sentinel3:26379
    password: redis123
    lettuce:
      pool:
        max-active: 20
        min-idle: 5

十一、Cluster（集群） ​

11.1 为什么需要 Cluster？ ​

主从 + Sentinel 解决了高可用，但所有数据都在一个 Master 上，存在容量瓶颈。Cluster 实现数据分片，将数据分散到多个 Master 节点。

11.2 哈希槽（Hash Slot） ​

Redis Cluster 将整个数据集划分为 16384 个哈希槽：

CRC16(key) % 16384 → 确定 key 属于哪个槽

每个 Master 节点负责一部分槽：

Master1: 槽 0-5460
Master2: 槽 5461-10922
Master3: 槽 10923-16383

为什么是 16384？

  • 心跳包中需要携带槽位信息，16384 个槽用 2KB 位图表示，65536 则需要 8KB
  • 16384 已经足够分片（最多 1000 个节点），500+ 节点时网络开销已很大
  • 槽位不是越多越好，需要权衡心跳包大小和节点数

11.3 重定向机制 ​

MOVED 重定向（永久）：

bash
# 客户端连接 Master1，但 key 的槽在 Master2
GET mykey
# 返回：MOVED 12345 192.168.1.2:6379
# 客户端应更新槽位映射，直接请求 Master2

ASK 重定向（临时，槽迁移中）：

bash
# 槽正在从 Master1 迁移到 Master2
GET mykey
# 返回：ASK 12345 192.168.1.2:6379
# 客户端先发送 ASKING 命令，再执行 GET

智能客户端： JedisCluster、Lettuce 等客户端会缓存槽位映射，自动处理重定向。

11.4 数据迁移（Resharding） ​

当需要扩容/缩容时，通过 redis-cli --cluster reshard 将部分槽从一个节点迁移到另一个节点。迁移过程中槽对应的 key 逐个通过 MIGRATE 命令原子迁移。

bash
# 扩容：添加新节点，将部分槽迁移到新节点
redis-cli --cluster add-node 192.168.1.4:6379 192.168.1.1:6379
redis-cli --cluster reshard 192.168.1.1:6379
# 指定迁移多少个槽，从哪些节点迁移到哪些节点

# 缩容：先迁移槽，再删除节点
redis-cli --cluster reshard 192.168.1.1:6379  # 先将槽迁走
redis-cli --cluster del-node 192.168.1.1:6379 <node-id>  # 删除节点

11.5 Cluster 配置 ​

bash
# redis.conf
cluster-enabled yes
cluster-config-file nodes.conf
cluster-node-timeout 15000  # 节点超时时间（毫秒）
cluster-require-full-coverage yes  # 所有槽覆盖才可用（设为 no 允许部分槽不可用）
yaml
# Spring Boot 配置
spring:
  redis:
    cluster:
      nodes:
        - 192.168.1.1:6379
        - 192.168.1.2:6379
        - 192.168.1.3:6379
        - 192.168.1.4:6379
        - 192.168.1.5:6379
        - 192.168.1.6:6379
    password: redis123
    lettuce:
      cluster:
        refresh:
          adaptive: true        # 自适应刷新拓扑
          period: 30s           # 刷新周期

十二、缓存经典问题 ​

12.1 缓存穿透 ​

场景： 查询一个不存在的数据（id=-1），缓存和数据库都没有，每次请求都穿透到数据库。

解决方案：

java
// 方案一：布隆过滤器（Bloom Filter）
// 使用 Redisson 的布隆过滤器
RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter("product:bloom");
bloomFilter.tryInit(1000000L, 0.03);  // 预计 100 万元素，3% 误判率

public Product getProduct(Long id) {
    if (!bloomFilter.contains(id)) {
        return null;  // 一定不存在，直接返回
    }
    // 可能存在，查缓存
    Product product = getFromCache(id);
    if (product == null) {
        product = getFromDB(id);
        if (product != null) {
            setToCache(id, product);
        }
    }
    return product;
}

// 方案二：缓存空值（简单但可能浪费内存）
public Product getProductWithNull(Long id) {
    String cacheKey = "product:" + id;
    String cached = redisTemplate.opsForValue().get(cacheKey);
    if (cached != null) {
        return "NULL".equals(cached) ? null : JSON.parseObject(cached, Product.class);
    }
    Product product = productMapper.selectById(id);
    if (product == null) {
        redisTemplate.opsForValue().set(cacheKey, "NULL", 5, TimeUnit.MINUTES);
    } else {
        redisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(product), 
            1, TimeUnit.HOURS);
    }
    return product;
}

12.2 缓存击穿 ​

场景： 热点 key 过期瞬间，大量请求同时打到数据库。

解决方案：

java
// 方案一：互斥锁（分布式锁）
public Product getProductMutex(Long id) {
    String cacheKey = "product:" + id;
    String lockKey = "lock:product:" + id;
    
    Product product = getFromCache(cacheKey);
    if (product != null) return product;
    
    // 获取分布式锁，只有一个线程去查 DB
    Boolean locked = redisTemplate.opsForValue()
        .setIfAbsent(lockKey, "1", 10, TimeUnit.SECONDS);
    if (Boolean.TRUE.equals(locked)) {
        try {
            // 双重检查
            product = getFromCache(cacheKey);
            if (product != null) return product;
            
            product = productMapper.selectById(id);
            if (product != null) {
                redisTemplate.opsForValue().set(cacheKey, 
                    JSON.toJSONString(product), 1, TimeUnit.HOURS);
            }
        } finally {
            redisTemplate.delete(lockKey);
        }
    } else {
        // 没拿到锁，等待后重试
        Thread.sleep(50);
        return getProductMutex(id);  // 递归重试
    }
    return product;
}

// 方案二：逻辑过期（不设 TTL，value 中存过期时间）
// 异步更新，不阻塞请求
public Product getProductLogicalExpire(Long id) {
    String cacheKey = "product:" + id;
    CacheData data = getFromCache(cacheKey);
    if (data == null) {
        // 缓存未命中，查 DB 并存入
        return rebuildCache(id);
    }
    // 检查是否逻辑过期
    if (data.getExpireTime() < System.currentTimeMillis()) {
        // 异步更新缓存
        threadPool.execute(() -> rebuildCache(id));
    }
    return data.getProduct();  // 返回旧数据，不阻塞
}

12.3 缓存雪崩 ​

场景： 大量 key 同时过期，或 Redis 宕机，所有请求打到数据库。

解决方案：

java
// 方案一：过期时间加随机值（防止同时过期）
public void setWithRandomExpire(String key, Object value, long baseSeconds) {
    long randomSeconds = new Random().nextInt(300);  // 0-300 秒的随机偏移
    long expireSeconds = baseSeconds + randomSeconds;
    redisTemplate.opsForValue().set(key, value, expireSeconds, TimeUnit.SECONDS);
}

// 方案二：多级缓存（本地缓存 + Redis）
// 使用 Caffeine 作为一级缓存
Cache<String, Product> localCache = Caffeine.newBuilder()
    .maximumSize(10000)
    .expireAfterWrite(5, TimeUnit.MINUTES)
    .build();

public Product getProductMultiLevel(Long id) {
    String key = "product:" + id;
    // 一级缓存：本地
    Product product = localCache.getIfPresent(key);
    if (product != null) return product;
    // 二级缓存：Redis
    String json = redisTemplate.opsForValue().get(key);
    if (json != null) {
        product = JSON.parseObject(json, Product.class);
        localCache.put(key, product);
        return product;
    }
    // 三级：数据库
    product = productMapper.selectById(id);
    if (product != null) {
        redisTemplate.opsForValue().set(key, JSON.toJSONString(product), 
            1, TimeUnit.HOURS);
        localCache.put(key, product);
    }
    return product;
}

// 方案三：Redis 高可用（主从 + Sentinel + Cluster）
// 方案四：服务限流降级（Hystrix、Sentinel）

12.4 缓存一致性 ​

问题： 数据库更新后，缓存还是旧数据。

解决方案：

java
// 方案一：Cache Aside（旁路缓存，最常用）
// 先更新数据库，再删除缓存
@Transactional
public void updateProduct(Product product) {
    productMapper.updateById(product);           // 1. 更新数据库
    redisTemplate.delete("product:" + product.getId());  // 2. 删除缓存
}

// 方案二：延迟双删
public void updateProductDoubleDelete(Product product) {
    redisTemplate.delete("product:" + product.getId());  // 1. 先删缓存
    productMapper.updateById(product);                    // 2. 更新数据库
    Thread.sleep(500);  // 3. 等待一段时间
    redisTemplate.delete("product:" + product.getId());  // 4. 再删缓存
}

// 方案三：订阅 MySQL Binlog（Canal）
// 通过 Canal 监听 binlog 变更，异步更新缓存，实现最终一致性

十三、Spring Boot 集成 ​

13.1 基础配置 ​

yaml
spring:
  redis:
    host: localhost
    port: 6379
    password: redis123
    database: 0
    timeout: 3000ms
    lettuce:
      pool:
        max-active: 20      # 最大连接数
        max-idle: 10        # 最大空闲连接
        min-idle: 5         # 最小空闲连接
        max-wait: 3000ms    # 等待连接最大时间

13.2 RedisTemplate 配置 ​

java
@Configuration
public class RedisConfig {
    
    @Bean
    public RedisTemplate<String, Object> redisTemplate(
            RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        
        // Key 序列化
        StringRedisSerializer stringSerializer = new StringRedisSerializer();
        template.setKeySerializer(stringSerializer);
        template.setHashKeySerializer(stringSerializer);
        
        // Value 序列化（JSON）
        Jackson2JsonRedisSerializer<Object> jsonSerializer = 
            new Jackson2JsonRedisSerializer<>(Object.class);
        ObjectMapper mapper = new ObjectMapper();
        mapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        mapper.activateDefaultTyping(
            LaissezFaireSubTypeValidator.instance,
            ObjectMapper.DefaultTyping.NON_FINAL);
        jsonSerializer.setObjectMapper(mapper);
        
        template.setValueSerializer(jsonSerializer);
        template.setHashValueSerializer(jsonSerializer);
        template.afterPropertiesSet();
        return template;
    }
}

13.3 Spring Cache 集成 ​

java
@Configuration
@EnableCaching
public class CacheConfig {
    
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisCacheConfiguration config = RedisCacheConfiguration
            .defaultCacheConfig()
            .entryTtl(Duration.ofHours(1))          // 默认过期 1 小时
            .serializeKeysWith(RedisSerializationContext.SerializationPair
                .fromSerializer(new StringRedisSerializer()))
            .serializeValuesWith(RedisSerializationContext.SerializationPair
                .fromSerializer(new GenericJackson2JsonRedisSerializer()))
            .disableCachingNullValues();             // 不缓存 null
        return RedisCacheManager.builder(factory)
            .cacheDefaults(config)
            .build();
    }
}

// 使用缓存注解
@Service
public class ProductService {
    
    @Cacheable(value = "product", key = "#id", unless = "#result == null")
    public Product getById(Long id) {
        return productMapper.selectById(id);
    }
    
    @CachePut(value = "product", key = "#product.id")
    public Product update(Product product) {
        productMapper.updateById(product);
        return product;
    }
    
    @CacheEvict(value = "product", key = "#id")
    public void delete(Long id) {
        productMapper.deleteById(id);
    }
    
    @Caching(evict = {
        @CacheEvict(value = "product", key = "#product.id"),
        @CacheEvict(value = "product:list", allEntries = true)
    })
    public Product save(Product product) {
        productMapper.insert(product);
        return product;
    }
}

13.4 Redisson 分布式锁 ​

java
@Configuration
public class RedissonConfig {
    
    @Bean
    public RedissonClient redissonClient() {
        Config config = new Config();
        config.useSingleServer()
            .setAddress("redis://localhost:6379")
            .setPassword("redis123")
            .setConnectionPoolSize(20);
        return Redisson.create(config);
    }
}

@Service
public class OrderService {
    
    @Autowired
    private RedissonClient redissonClient;
    
    public void createOrder(Long orderId) {
        String lockKey = "lock:order:" + orderId;
        RLock lock = redissonClient.getLock(lockKey);
        
        try {
            // 尝试加锁，最多等待 10 秒，锁 30 秒后自动释放
            if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
                // 业务逻辑
                System.out.println("创建订单：" + orderId);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            // 释放锁（只有持有锁的线程才能释放）
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
    
    // Redisson 看门狗机制：自动续期，防止业务执行超时锁过期
    // 不指定 leaseTime 时，默认 30 秒，每 10 秒续期一次
    public void longRunningTask() {
        RLock lock = redissonClient.getLock("lock:task");
        lock.lock();  // 看门狗自动续期
        try {
            // 长时间业务逻辑
            Thread.sleep(60000);
        } finally {
            lock.unlock();
        }
    }
}

十四、面试要点 ​

Q1：Redis 为什么快？

核心四点：纯内存操作（微秒级）；IO 多路复用（单线程高效处理并发连接）；高效数据结构（SDS、ziplist、skiplist 等）；单线程避免锁竞争和上下文切换。Redis 6.0 后多线程只用于网络 IO 读写，命令执行仍是单线程。

Q2：RDB 和 AOF 的区别？如何选择？

RDB 是定时快照，恢复快但可能丢数据；AOF 是追加写命令，数据安全但文件大、恢复慢。生产环境推荐 Redis 4.0 的混合持久化（RDB 快照 + AOF 增量），兼顾两者优势。同时保留 RDB 定期备份用于灾难恢复。

Q3：缓存穿透、击穿、雪崩的区别和解决方案？

  • 穿透：查询不存在的数据 → 布隆过滤器 / 缓存空值
  • 击穿：热点 key 过期 → 互斥锁 / 逻辑过期（异步更新）
  • 雪崩：大量 key 同时过期 / Redis 宕机 → 随机 TTL / 多级缓存 / 高可用集群

Q4：Redis 如何实现分布式锁？

使用 SETNX + expire 保证原子性加锁，用 Lua 脚本保证原子性释放（判断 value 是否匹配再删除）。Redisson 封装了分布式锁，支持可重入、看门狗自动续期、公平锁、联锁等高级功能。生产环境推荐使用 Redisson。

Q5：Redis Cluster 如何实现数据分片？

使用 16384 个哈希槽，CRC16(key) % 16384 确定 key 归属的槽，每个 Master 节点负责一部分槽。客户端缓存槽位映射，直接路由到目标节点。扩缩容时通过 reshard 迁移槽，MOVED 和 ASK 重定向保证迁移期间的可用性。

Q6：Redis 主从复制的全量同步和增量同步流程？

全量同步：Slave 首次连接，Master 执行 BGSAVE 生成 RDB 发送给 Slave，Slave 加载 RDB 后，Master 发送 buffer 中的增量命令。增量同步：Slave 重连时带上 offset，Master 检查 repl_backlog 缓冲区，有则发送增量，无则触发全量同步。

Q7：Redis Sentinel 的故障转移流程？

SDOWN（单个 Sentinel 认为 Master 不可达）→ ODOWN（quorum 个 Sentinel 确认）→ Sentinel 领导者选举（Raft）→ 选择最优 Slave（优先级 > 复制偏移量 > runid）→ 提升为新 Master → 通知客户端。

Q8：Spring Cache 的 @Cacheable、@CachePut、@CacheEvict 的区别？

  • @Cacheable：先查缓存，命中则返回，未命中则执行方法并将结果存入缓存
  • @CachePut：总是执行方法，并将结果存入缓存（更新缓存）
  • @CacheEvict：删除缓存，可指定 key 或清空整个缓存（allEntries=true）