第三阶段：进阶运维与排查
├── Stream 消息队列（143-149）
│   ├── 消费者组 & 消费模型
│   ├── 消息可靠消费 & ACK
│   ├── Pub/Sub 局限性
│   ├── 消息 ID & 核心命令
│   └── 与 Kafka 对比
│
├── 备份与恢复（150-159）
│   ├── RDB & AOF 持久化
│   ├── 备份策略 & 在线备份
│   ├── 误操作恢复
│   ├── 跨机房备份
│   └── 备份对性能的影响
│
├── 排查与应急（160-169）
│   ├── 连接数过高
│   ├── 响应变慢原因
│   ├── 内存满 & OOM
│   ├── 主从复制 & 哨兵频繁切换
│   ├── 集群通信
│   ├── DEBUG 命令问题
│   ├── 客户端异常
│   └── 应急处理

# 消费者组的最佳实践
# 1. 必须 ACK
# 2. 设置合理的 MAXLEN 防止内存膨胀
# 3. 监控 PEL 大小
# 4. 实现死信队列
# 5. 保证业务幂等性（at-least-once 需要）

# 严格避免
# ❌ 忘记 XACK → PEL 膨胀
# ❌ 不设 MAXLEN → Stream 无限增长
# ❌ 消费失败无限重试 → 死循环

# 生产环境推荐配置
# 1. 开启混合持久化（Redis 4.0+）
appendonly yes
appendfsync everysec
aof-use-rdb-preamble yes

# 2. 配置 RDB 快照
save 900 1
save 300 10
save 60 10000

# 3. 从库备份（不影响主库）
# 4. 异地备份 + 定期验证
# 5. 备份保留策略：日备保留 7 天，周备保留 4 周

数据重要性 | RPO     | RTO     | 备份策略
核心数据   | <1秒   | <5分钟  | 混合持久化 + 从库 + 异地实时备份
重要数据   | <1小时 | <30分钟 | RDB 每小时 + 从库
一般数据   | <1天   | <2小时  | RDB 每天一次

注意：RPO 越小，成本越高，根据业务选择

# 必须监控的指标
- 性能指标:
  - connected_clients
  - instantaneous_ops_per_sec
  - latency (99th percentile)
  - slowlog 数量

- 内存指标:
  - used_memory / maxmemory
  - mem_fragmentation_ratio
  - evicted_keys 速率

- 持久化指标:
  - rdb_last_bgsave_status
  - aof_last_rewrite_status
  - latest_fork_usec

- 主从/集群指标:
  - master_link_status
  - slave_lag
  - cluster_state

# 快速健康检查
redis-cli INFO | grep -E \
  "connected_clients|used_memory_human|evicted_keys|rejected_connections|master_link_status"

初学者：
  基本数据类型 → 过期策略 → 持久化 → 主从复制

进阶：
  哨兵 → 集群 → Stream → Lua 脚本

高级运维：
  备份恢复 → 性能调优 → 故障排查 → 应急处理

面试重点：
  高可用架构（哨兵+集群）
  数据安全（持久化+备份）
  性能优化（慢查询+大 Key）
  故障处理（OOM+复制中断）

✅ 适合：
- 缓存（90% 场景）
- 计数器、限流
- 分布式锁、Session 共享
- 轻量级消息队列
- 实时排行榜、地理位置

❌ 不适合：
- 关系型数据存储（用 MySQL）
- 全文搜索（用 ES）
- 海量消息队列（用 Kafka）
- 复杂分析查询（用 ClickHouse）

1. 永远不要在生产环境：
   - 执行 KEYS *（用 SCAN）
   - 执行 DEBUG SEGFAULT
   - 不备份就重启
   - 直接 DEL 大 Key（用 UNLINK）

2. 必须配置：
   - rename-command 禁用危险命令
   - maxmemory + 淘汰策略
   - 合理超时设置
   - 主从 + 哨兵/集群高可用

3. 必须监控：
   - 内存使用率
   - 连接数
   - 复制状态
   - 持久化状态

Redis 启动
  ↓
检查持久化配置
  ↓
如果 AOF 开启：
  ├── 检查 AOF 文件是否存在
  ├── 尝试修复截断的 AOF（如果配置）
  ├── 创建伪客户端（fake client）用于执行命令
  ├── 逐条读取并执行 AOF 中的命令
  │   SET key1 value1
  │   HSET user:1 name tom
  │   LPUSH list1 item1
  │   ...
  └── 加载完成后，Redis 可开始提供服务
  ↓
如果 AOF 关闭但 RDB 存在：
  └── 加载 RDB 文件

Redis 启动时的优先级顺序：
1. 如果 appendonly yes → 加载 AOF
2. 否则，有 RDB 文件 → 加载 RDB
3. 都没有 → 空数据库

# Step 1: 确认 AOF 配置
redis-cli CONFIG GET appendonly
# 1) "appendonly"
# 2) "yes"

redis-cli CONFIG GET appendfilename
# "appendonly.aof"

# Step 2: 确保 AOF 文件在正确位置
ls -la /var/lib/redis/appendonly.aof

# Step 3: 启动/重启 Redis
systemctl restart redis

# Step 4: 验证恢复
redis-cli DBSIZE
redis-cli INFO Persistence | grep aof_current_rewrite_time_sec

# 使用不同的 AOF 文件启动
cp /backup/aof_backup.rdb /tmp/aof_manual_recover.aof

redis-server --port 6380 \
             --appendonly yes \
             --appendfilename /tmp/aof_manual_recover.aof \
             --dir /tmp \
             --daemonize yes

# 1. 找到目标时间点前的命令位置
# 2. 截断 AOF 文件
head -n X /var/lib/redis/appendonly.aof > /tmp/partial.aof

# 3. 用截断的文件恢复
redis-server --port 6380 --appendonly yes --dir /tmp \
             --dbfilename /dev/null \
             --appendfilename partial.aof \

# 方式 1：使用 redis-check-aof 修复
redis-check-aof --fix /var/lib/redis/appendonly.aof

# 方式 2：配置自动修复
$ redis.conf
aof-load-truncated yes

# Step 1: 备份损坏的 AOF
cp /var/lib/redis/appendonly.aof /backup/appendonly.aof.bad

# Step 2: 尝试修复
redis-check-aof --fix appendonly.aof

# Step 3: 如果修复失败，使用 RDB 文件（如果有）
# 或者使用上一个版本的 AOF

AOF 文件大小  | 命令数量    | 恢复时间
--------------|-------------|---------
1GB           | 500万条     | ~30秒
10GB          | 5000万条    | ~5分钟
50GB          | 2.5亿条     | ~30分钟

# 开启混合持久化
redis.conf
aof-use-rdb-preamble yes

混合 AOF 文件结构：
┌─────────────────────────────────┐
│ RDB 格式全量数据（快速加载）    │
├─────────────────────────────────┤
│ AOF 格式增量命令（精确重放）    │
└─────────────────────────────────┘

加载流程：
加载 RDB 部分 → 几秒钟（代替全量重放）
重放 AOF 部分 → 少量增量命令

# 带有 AOF+RDB 混合配置时的启动逻辑
if appendonly == yes:
    load aof file  # 即使有 RDB 也优先加载 AOF
else:
    load rdb file  # 仅加载 RDB

SET key value1
SET key value2    → 最终值为 value2
INCR counter      → 多次重放不会出错

# AOF 恢复时，过期 key 的处理方式：
# 1. 如果有过期时间标记，正常设置 TTL
# 2. 恢复过程中不主动淘汰过期 key
# 3. 恢复完成后，Redis 开始服务时开始淘汰

# 1. 基本数据检查
redis-cli DBSIZE
redis-cli INFO Keyspace

# 2. 特定 key 验证
redis-cli GET critical-key
redis-cli TYPE key-name

# 3. 内存使用检查
redis-cli INFO memory

# 4. AOF 状态检查
redis-cli INFO Persistence | grep aof

# 5. 重复数据检查（确认没有重复）
redis-cli --bigkeys

# 6. 检查过期策略
redis-cli CONFIG GET maxmemory-policy

Redis 启动
  ↓
查找 RDB 文件（dbfilename + dir 配置）
  ↓
加载 RDB 文件
  ├── 检查文件头（REDIS + 版本号）
  ├── 读取校验和验证完整性
  ├── 解压数据（如果压缩）
  └── 逐条加载键值对到内存
  ↓
Redis 服务就绪

# Step 1: 确认配置
redis-cli CONFIG GET dir
# "/var/lib/redis"

redis-cli CONFIG GET dbfilename
# "dump.rdb"

# Step 2: 将备份文件放到正确位置
cp /backup/redis/dump_20240101.rdb /var/lib/redis/dump.rdb

# Step 3: 重启 Redis
systemctl restart redis

# Step 4: 验证恢复
redis-cli DBSIZE
redis-cli KEYS "*" | wc -l

# 使用临时配置启动 Redis 并加载指定 RDB 文件
redis-server --port 6380 \
             --dir /tmp/restore \
             --dbfilename mybackup.rdb \
             --daemonize yes

# 验证数据
redis-cli -p 6380 DBSIZE

# 如果满意，可以停止临时实例并替换正式实例
redis-cli -p 6380 SHUTDOWN

# 从 RDB 文件恢复到运行中的 Redis
redisshake restore -target redis://127.0.0.1:6379 \
                   -type rdb \
                   -file /tmp/dump.rdb

rdbchecksum yes  # 默认开启，加载时验证校验和

# 校验失败的错误日志
# Bad CRC64 signature for the RDB file

rdbcompression yes  # RDB 文件使用 LZF 压缩

恢复时间 ≈ 数据量 / 加载速度

示例：
- 10GB RDB 文件
- 加载速度 ~200MB/s（压缩）
- 恢复时间 ≈ 50 秒 + 10 秒（数据结构重建）

# 1. 增大内核参数提高文件读取性能
sysctl -w vm.max_map_count=262144

# 2. 将 RDB 放在 SSD/NVMe 磁盘
# 3. 考虑使用混合持久化，恢复更快

# 1. 检查 key 数量
redis-cli DBSIZE

# 2. 检查关键 key
redis-cli GET critical_key
redis-cli HGETALL user:1000

# 3. 检查内存使用
redis-cli INFO memory | grep used_memory_human

# 4. 检查是否有过期 key（对比时间戳）
redis-cli INFO keyspace

# 5. 抽样检查（大型实例）
redis-cli --bigkeys

# 6. 检查持久化配置
redis-cli CONFIG GET save
redis-cli CONFIG GET appendonly

# 检查错误日志
tail -f /var/log/redis/redis-server.log

# 常见原因：
# 1. 文件损坏 → 使用 redis-check-rdb 修复
redis-check-rdb /var/lib/redis/dump.rdb

# 2. 文件版本不兼容（低版本 Redis 加载高版本 RDB）
redis-check-rdb --fix /var/lib/redis/dump.rdb  # 尝试修复

# 3. 内存不足
free -h  # 检查可用内存

# 可能原因：
# 1. 备份不是最新版本
# 2. AOF 和 RDB 混合使用导致不一致
# 3. 部分数据被 LRU 淘汰

# 解决方案：
# 1. 确认备份时间点
ls -la /backup/redis/
# 2. 使用更新备份恢复

# RDB 文件加载需要等于数据量的内存
# 检查系统内存
free -h

# 优化：
# 1. 使用 maxmemory 限制
# 2. 增大 swap（不推荐生产）
# 3. 使用更大的实例恢复
# 4. 考虑升级硬件

# 从库恢复步骤
# 1. 停止从库
# 2. 复制 RDB 到从库数据目录
# 3. 启动从库，自动加载 RDB
# 4. 从库自动与主库同步增量数据

负载因子 = 已使用的桶数 / 总桶数

// 负载因子 >= 1，并且可以扩容（没有正在进行的 BGSAVE 等）
// 或负载因子 >= 5，强制扩容

旧哈希表 (size=2^20) → 新哈希表 (size=2^21)
需要重新计算 100 万+ key 的哈希值并插入新表
这期间 Redis 主线程被占用

# 一次性写入 1000 万 key
pipe = r.pipeline()
for i in range(10_000_000):
    pipe.set(f"key:{i}", i)
    if i % 10000 == 0:
        pipe.execute()

# 写入大量 key 后一次性删除
r.flushdb()  # FLUSHDB 本身是 O(N) 操作
# 删除后哈希表缩容，触发 rehash

127.0.0.1:6379> INFO stats
# Stats
latest_fork_usec: 132           -- 最近一次 fork 耗时
# 没有直接显示 rehash 耗时，但可以通过以下方式间接判断

# 查看延迟事件历史
LATENCY LATEST
# 1) 1) "command"    -- 命令延迟
# 2) (integer) 1700000000
# 3) (integer) 15234    -- 15ms 延迟
# 4) (integer) 15234

# 检查 rehash 是否正在进行
INFO keyspace
# 如果 rehash 中有大量 key 需要迁移，会看到 rehash 相关指标

// Redis 中的 rehash 策略
if (dictIsRehashing(d)) {
    // 渐进式 rehash 正在执行
    // 每次操作（增删改查）都迁移一小批数据
    _dictRehashStep(d);
}

# 在大量写入前，强制初始化哈希表大小
# 通过先执行一个空的 SET 操作，然后让 Redis 按需扩容
# 更好的方式：在初始化阶段批量写入，让 rehash 在低负载时发生

# 避免一次性大量写入触发强制 rehash
BATCH_SIZE = 5000
for i in range(0, 1_000_000, BATCH_SIZE):
    pipe = r.pipeline()
    for j in range(i, i + BATCH_SIZE):
        pipe.set(f"key:{j}", j)
    pipe.execute()
    time.sleep(0.001)  # 给 Redis 喘息机会进行渐进式 rehash

# 定期检查是否处于 rehash 状态
redis-cli DEBUG HTSTATS 0
# 查看 key 分布和 rehash 进度

# 使用多个 Redis 实例分担数据量
# 每个实例的 key 数量减少，rehash 影响也变小

10,000,000 key × 重新哈希时间 ≈ 500ms 阻塞

typedef struct dict {
    dictht ht[2];          // 两个哈希表
    long rehashidx;        // rehash 进度，-1 表示未进行
    // ...
} dict;

typedef struct dictht {
    dictEntry **table;     // 哈希表数组
    unsigned long size;    // 哈希表大小
    unsigned long sizemask; // 掩码，用于计算索引
    unsigned long used;    // 已使用的节点数
} dictht;

// 触发条件：负载因子 >= 1（可扩容）或 >= 5（强制扩容）
// 1. 创建新哈希表 ht[1]，大小为 2 的幂且 >= ht[0].used * 2
// 2. 设置 rehashidx = 0，表示开始渐进式 rehash
ht[1] = dictNextPower(ht[0].used * 2);
rehashidx = 0;

ht[0] (旧表) - 正在被使用，有 1000 万 key
ht[1] (新表) - 空表，容量 2000 万
rehashidx = 0

// 每次操作时，自动迁移少量 bucket
static void _dictRehashStep(dict *d) {
    if (d->rehashidx != -1)
        dictRehash(d, 1);  // 迁移 1 个 bucket
}

// 定时任务中也会进行批量迁移
// serverCron 定时函数中会调用
int dictRehashMilliseconds(dict *d, int ms) {
    // 每次定时任务最多执行 ms 毫秒的 rehash
    // 默认每次 rehash 执行 1 毫秒
}

int dictRehash(dict *d, int n) {
    int empty_visits = n * 10;  // 最多跳过 10*n 个空桶
    while (n-- && d->ht[0].used != 0) {
        dictEntry *de, *nextde;

        // 找到下一个非空桶
        while (d->ht[0].table[d->rehashidx] == NULL) {
            d->rehashidx++;
            if (--empty_visits <= 0) return 1;
        }

        // 将桶中所有 entry 迁移到新表
        de = d->ht[0].table[d->rehashidx];
        while (de) {
            uint64_t h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            nextde = de->next;
            // 插入到 ht[1] 的对应位置
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        }
        d->ht[0].table[d->rehashidx] = NULL;
        d->rehashidx++;
    }

    // 检查是否全部迁移完成
    if (d->ht[0].used == 0) {
        dictRelease(d->ht[0]);
        d->ht[0] = d->ht[1];
        dictReset(d->ht[1]);
        d->rehashidx = -1;
        return 0;  // rehash 完成
    }
    return 1;  // rehash 进行中
}

1. 释放 ht[0] 的内存
2. 将 ht[1] 赋值给 ht[0]
3. 重置 ht[1] 为空
4. rehashidx = -1，表示完成

// 查找时，先查 ht[0]，再查 ht[1]
dictEntry *dictFind(dict *d, const void *key) {
    if (d->rehashidx != -1) {
        // 如果正在 rehash，顺便迁移一个 bucket
        _dictRehashStep(d);
    }
    h = dictHashKey(d, key);
    // 先查旧表
    for (table = 0; table <= 1; table++) {
        idx = h & d->ht[table].sizemask;
        he = d->ht[table].table[idx];
        while (he) {
            if (dictCompareKeys(d, key, he->key))
                return he;
            he = he->next;
        }
        if (!dictIsRehashing(d)) break;
    }
    return NULL;
}

// 新增时，直接插入到 ht[1]（新表）
// 避免在旧表创建后还需要迁移
int dictAdd(dict *d, void *key, void *val) {
    if (d->rehashidx != -1) {
        // 直接插入新表
        _dictKeyIndex(d, key, &idx);
        entry = zalloc(sizeof(*entry));
        entry->next = d->ht[1].table[idx];
        d->ht[1].table[idx] = entry;
        d->ht[1].used++;
    }
    // ...
}

// 删除时，从两个表中都查找并删除
int dictDelete(dict *d, const void *key) {
    if (d->rehashidx != -1) _dictRehashStep(d);
    // 查找并删除...
}

# 监控 rehash 期间的内存
import redis
r = redis.Redis()

while True:
    info = r.info('memory')
    print(f"used_memory: {info['used_memory_human']}")
    # 如果 used_memory 突然翻倍，可能正在 rehash
    time.sleep(1)

public String getWithLock(String key) {
    String value = redis.opsForValue().get(key);
    if (value != null) return value;

    String lockKey = "lock:" + key;
    String requestId = UUID.randomUUID().toString();
    Boolean locked = redis.opsForValue()
        .setIfAbsent(lockKey, requestId, 3, TimeUnit.SECONDS);

    if (Boolean.TRUE.equals(locked)) {
        try {
            value = loadFromDB(key);
            redis.opsForValue().set(key, value, 300, TimeUnit.SECONDS);
            return value;
        } finally {
            // 只删除自己加的锁
            String lua = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                         "then return redis.call('del', KEYS[1]) else return 0 end";
            redis.execute(new DefaultRedisScript<>(lua, Long.class), 
                          Arrays.asList(lockKey), requestId);
        }
    } else {
        // 等待 50ms 后重试
        Thread.sleep(50);
        return redis.opsForValue().get(key);
    }
}

public class CacheValue<T> {
    private T data;
    private long expireAt;  // 逻辑过期时间
}

public T getWithSoftExpire(String key) {
    CacheValue<T> cv = (CacheValue<T>) redis.opsForValue().get(key);
    if (cv == null) {
        return loadFromDB(key);  // 兜底
    }
    if (cv.expireAt > System.currentTimeMillis()) {
        return cv.data;  // 逻辑过期时间内，直接返回
    }
    // 逻辑过期，异步刷新，先返回旧数据
    asyncRefresh(key);
    return cv.data;
}

String key = "user:" + userId;
int ttl = 1800 + ThreadLocalRandom.current().nextInt(600);  // 30min ± 5min
redis.opsForValue().set(key, value, ttl, TimeUnit.SECONDS);

// 结合定时任务 + 热度统计，提前刷新即将到期的热点数据
@Scheduled(fixedRate = 60_000)
public void preRefresh() {
    hotKeys.stream()
        .filter(key -> isExpiringSoon(key))  // TTL < 20% 原始值
        .forEach(key -> {
            String value = loadFromDB(key);
            redis.opsForValue().set(key, value, 1800, TimeUnit.SECONDS);
        });
}

# 错误做法：所有缓存设置相同的过期时间
redis.setex("hot:news:1", 3600, data1)
redis.setex("hot:news:2", 3600, data2)
redis.setex("hot:news:3", 3600, data3)
# ... 一小时后所有 Key 同时过期

Redis 主节点宕机 → 哨兵故障转移（10-30秒不可用）
                 → 期间所有请求穿透到数据库
                 → 数据库可能被压垮

import random

def set_cache_with_skew(key, value, base_ttl):
    # 基础过期时间 ± 随机偏移
    ttl = base_ttl + random.randint(-300, 300)  # ±5 分钟随机
    redis.setex(key, ttl, value)

def batch_set_cache(items, base_ttl):
    for i, (key, value) in enumerate(items):
        # 每个 Key 的偏移量逐步递增
        ttl = base_ttl + i * random.randint(10, 60)
        redis.setex(key, ttl, value)

未错峰：00:00 → 10000 个 Key 同时过期 → 数据库 10000 QPS
已错峰：00:00 → 逐渐过期 → 数据库 100 QPS

def preload_cache(business_data):
    """
    在大促、秒杀等活动前执行
    """
    for item in business_data:
        # 设置不同过期时间
        ttl = 3600 + random.uniform(0, 1800)
        redis.setex(f"hot:{item.id}", int(ttl), item.to_json())

# redis.conf
save 900 1       # 900 秒内至少 1 次修改则保存
save 300 10      # 300 秒内至少 10 次修改则保存
save 60 10000    # 60 秒内至少 10000 次修改则保存

appendonly yes   # 开启 AOF
appendfsync everysec  # 每秒 fsync

def get_data_with_fallback(key):
    # 尝试 Redis
    try:
        data = redis.get(key)
        if data:
            return data
    except (RedisConnectionError, TimeoutError):
        # Redis 不可用，走降级逻辑
        pass

    # 降级到本地缓存
    data = local_cache.get(key)
    if data:
        return data

    # 本地缓存也没有，查数据库（限流）
    if rate_limiter.try_acquire("db_query", 100):  # 限流 100 QPS
        data = query_db(key)
        local_cache.put(key, data, 30)  # 本地缓存 30 秒
        return data
    else:
        # 超过限流，返回默认值或错误
        return DEFAULT_VALUE

# 对数据库查询做限流
@RateLimiter(max_calls=500, period=1)  # 每秒最多 500 个查询
def query_db(key):
    return db.execute("SELECT * FROM data WHERE id = ?", key)

# 熔断器：5 秒内错误率 > 50% 则打开熔断
@CircuitBreaker(failure_threshold=0.5, recovery_timeout=30)
def get_data(key):
    data = redis.get(key)
    if data is None:
        data = query_db(key)
        redis.setex(key, 3600, data)
    return data

# 熔断打开时，直接返回默认值
def fallback(key):
    return {"id": key, "status": "unavailable"}

# 多节点连接（客户端侧高可用）
redis_clients = [
    Redis("192.168.1.1:6379"),
    Redis("192.168.1.2:6379"),  # 备用
    Redis("192.168.1.3:6379"),  # 备用
]

def get_data_safe(key):
    for client in redis_clients:
        try:
            return client.get(key)
        except ConnectionError:
            continue
    # 所有 Redis 都不可用
    return fallback_to_db_with_rate_limit(key)

请求 → 客户端路由（多节点）
        ↓
    限流层（单 IP/用户限流）
        ↓
    本地缓存（Caffeine，30秒）
        ↓
    Redis Cluster（高可用 + 持久化）
        ↓
    数据库限流（500 QPS）
        ↓
    数据库
        ↓
熔断器（保护数据库）

import redis
import time
import threading

def get_data(key):
    # 1. 查缓存
    data = redis.get(key)
    if data is not None:
        return data

    # 2. 缓存未命中，尝试获取锁
    lock_key = f"lock:{key}"
    # 使用 SET NX 实现分布式锁，过期时间 5 秒防止死锁
    if redis.set(lock_key, "1", nx=True, ex=5):
        try:
            # 3. 双重检查：拿到锁后再次查缓存（防止等待时已被重建）
            data = redis.get(key)
            if data is not None:
                return data

            # 4. 查询数据库
            data = db.query("SELECT * FROM data WHERE id = ?", key)
            # 5. 重建缓存
            if data:
                redis.setex(key, 3600, data)
            else:
                redis.setex(key, 30, "NULL")  # 缓存空对象
            return data
        finally:
            redis.delete(lock_key)  # 释放锁
    else:
        # 6. 没拿到锁，等待后重试或返回旧数据
        time.sleep(0.1)
        return get_data(key)  # 递归重试（实际生产应防止无限递归）

def get_hot_data(key):
    # 1. 查缓存（永不过期）
    cached = redis.get(key)
    if cached is None:
        # 首次加载——正常查库
        return load_data_to_cache(key)

    # 2. 解析逻辑过期时间
    data = json.loads(cached)
    if data["expire_time"] > current_timestamp():
        # 3. 未到逻辑过期时间，直接返回
        return data["value"]

    # 4. 逻辑已过期，尝试获取锁重建
    lock_key = f"lock:{key}"
    if redis.set(lock_key, "1", nx=True, ex=3):
        try:
            # 后台异步更新
            thread = threading.Thread(target=refresh_cache, args=(key,))
            thread.start()
        finally:
            redis.delete(lock_key)

    # 5. 返回旧数据（不阻塞）
    return data["value"]

def refresh_cache(key):
    # 查询数据库，更新缓存
    db_data = db.query("SELECT * FROM data WHERE id = ?", key)
    cache_entry = {
        "value": db_data,
        "expire_time": current_timestamp() + 3600
    }
    redis.set(key, json.dumps(cache_entry))

# 方案 A：定时任务更新
@cron("0 */5 * * * *")  # 每 5 分钟执行一次
def refresh_hot_key():
    data = db.query("SELECT * FROM hot_items")
    redis.set("hot:items", json.dumps(data))  # 永不过期

# 方案 B：数据变更时更新（Cache-Aside style）
def update_hot_items(new_items):
    # 更新数据库
    db.execute("UPDATE hot_items SET ...")
    # 更新缓存（写入时更新，而非过期时重建）
    redis.set("hot:items", json.dumps(new_items))

请求 → [本地缓存（Caffeine）] → [Redis] → [数据库]

def get_hot_data_with_fallback(key):
    # 1. 查二级缓存
    local_data = local_cache.get(key)
    if local_data:
        return local_data

    # 2. 查 Redis
    redis_data = redis.get(key)
    if redis_data:
        local_cache.put(key, redis_data, 5)  # 本地缓存 5 秒
        return redis_data

    # 3. Redis 未命中，互斥锁查库
    with distributed_lock(f"lock:{key}", timeout=3):
        # 双重检查
        redis_data = redis.get(key)
        if redis_data:
            return redis_data
        # 查数据库
        db_data = query_db(key)
        redis.setex(key, 3600, db_data)
        local_cache.put(key, db_data, 5)
        return db_data

正常：请求 → [缓存命中] → 返回数据

击穿瞬间（某个热点 Key 过期）：
请求 1 → 缓存未命中（刚过期）→ 查数据库
请求 2 → 缓存未命中         → 查数据库
请求 3 → 缓存未命中         → 查数据库
...（海量请求同时打穿缓存）
请求 N → 缓存未命中         → 查数据库

数据库瞬间 QPS 暴增：
正常：数据库 200 QPS → 击穿瞬间：数据库 10000 QPS

场景：大量缓存设置了相同的过期时间（如凌晨 0 点全部过期）

00:00: 所有缓存 Key 同时过期
00:00:01 → 百万请求同时穿透到数据库
           ↓
        数据库连接池耗尽，响应变慢
           ↓
        更多请求堆积，数据库雪崩

场景：Redis 主节点宕机

Redis 宕机 → 所有缓存不可用 → 所有请求穿透到数据库
           → 数据库 50 万 QPS 压垮

正常：数据库 200 QPS → 雪崩时：数据库 100000 QPS（全量流量）

请求数据 → 查询缓存（未命中，cache miss）
           ↓
        查询数据库（也不存在）
           ↓
        返回 null，不缓存
           ↓
        下次同样的请求再次穿透缓存

正常情况下：
[10000 QPS] → [缓存层命中 9500] → [数据库 500 QPS] ✅

穿透攻击时：
[10000 QPS] → [缓存命中 0] → [数据库 10000 QPS] ❌

def get_user(user_id):
    # 查缓存
    user = redis.get(f"user:{user_id}")
    if user is not None:
        if user == "NULL":  # 缓存空对象
            return None
        return user

    # 查数据库
    user = db.query("SELECT * FROM users WHERE id = ?", user_id)
    if user is None:
        # 缓存空对象，30 秒过期
        redis.setex(f"user:{user_id}", 30, "NULL")
        return None

    # 缓存真实数据
    redis.setex(f"user:{user_id}", 3600, user)
    return user

请求 → 布隆过滤器（判断 Key 是否存在）
        ├── "一定不存在" → 直接返回 null（不查缓存和数据库）
        └── "可能存在" → 查缓存 → 查数据库

def get_user(user_id):
    # 参数校验
    if user_id is None or user_id <= 0:
        return None
    if len(str(user_id)) > 10:
        return None
    # 后续正常查询...

# 对单个 IP 或用户做限流
if is_rate_limited(client_ip, 100, "1s"):
    return "请求过于频繁"

[入口] → 参数校验 → 布隆过滤器 → 缓存查询 → [缓存空对象] → 数据库
                         ↓              ↓
              不存在则直接返回    未命中查库

补充：接口限流作为全局防护

[RDB 快照] + [增量 AOF 命令]

# redis.conf
save 900 1       # 900秒（15分钟）内至少1次key变更
save 300 10      # 300秒（5分钟）内至少10次变更
save 60 10000    # 60秒内至少10000次变更

save 900 1
save 300 10
save 60 10000

# 方式一：配置 save ""
save ""

# 方式二：config set 运行时禁用
redis> CONFIG SET save ""

if (dirty >= changes AND (now - lastsave) >= seconds) → trigger bgsave

redis-check-aof /var/lib/redis/appendonly.aof

AOF analyzed: size=2147483648, ok_up_to=2147000000
AOF is not valid

redis-check-aof --fix /var/lib/redis/appendonly.aof

The AOF appears to be truncated. Do you want to fix it? (y/n): y

cp appendonly.aof appendonly.aof.bak

redis-check-rdb /var/lib/redis/dump.rdb

BGSAVE 期间：
父进程：继续处理所有客户端请求（读/写）
子进程：遍历内存数据，生成 RDB 文件

flowchart LR
    subgraph "父进程（继续服务）"
        A["客户端读写请求"]
        B["过期 Key 清理"]
        C["定时任务"]
        D["AOF 写入（如开启）"]
    end

    subgraph "fork 时刻"
        E["fork() 系统调用
父进程阻塞"]
    end

    subgraph "子进程（RDB 快照）"
        F["遍历内存数据"]
        G["写临时 RDB 文件"]
        H["完成后通知父进程"]
    end

    A --> E --> A
    E --> F

// 简化伪代码
void bgsaveCommand() {
    // fork() 这一步会阻塞！复制父进程的页表
    pid_t pid = fork();

    if (pid == 0) {
        // 子进程代码：写 RDB
        rdbSave();
        exit(0);
    } else if (pid > 0) {
        // 父进程：继续处理其他请求
        return OK;
    }
}

fork 阻塞时间的影响因素：

Redis 占用内存    fork 阻塞时间（估算）
1GB               ~10~20ms
10GB              ~100~300ms
50GB              ~500ms~3s

为什么 fork 需要时间？
fork 需要复制父进程的页表
页表大小 ≈ 物理内存 / 页大小（4KB）
10GB 内存 → 页表约 2.5 MB → 复制需要时间

为什么越大越慢？
页表与内存成正比
内存越大 → 页表越大 → 复制越慢

fork 阻塞期间：
- 父进程不处理任何新请求
- 所有排队请求延迟增加
- 如果是 Redis Cluster，可能触发超时重连

所以：
1. 大实例（10GB+）避免在高峰期 BGSAVE
2. 监控 fork 耗时（INFO 命令查看 latest_fork_usec）
3. 启动时自动 RDB 加载也可能较长

COW 的影响：

内存开销：
父进程修改 1 个 Key → 复制 4KB 的内存页
修改 100 万个 Key → 最多复制 100万×4KB = 4GB 内存

性能影响：
COW 涉及缺页中断 + 页面复制 + TLB 刷新
每条写操作的延迟略有增加（增加几微秒）

关键：只影响被修改的数据页，不影响只读操作

# SAVE 和 BGSAVE 的区别
SAVE            # ❌ 主线程同步执行，所有请求被阻塞！
BGSAVE          # ✅ fork 子进程异步执行，主线程继续服务

SAVE 期间 Redis 表现：
- 完全不响应任何命令
- 所有客户端连接被阻塞
- 对于生产环境是灾难性的

什么时候用 SAVE？
1. 系统维护窗口（夜间停机维护）
2. 集群故障转移期间新 Master 的 RDB 加载
3. 明确知道可以接受停服的场景

BGREWRITEAOF            # 后台重写，可以继续服务

# Redis 启动时加载 RDB 文件
# 这个过程父进程也阻塞
# 加载完 RDB 后才开始接受请求

RDB 加载速度：
1GB RDB → 约 1~3秒（SSD）
10GB RDB → 约 10~30秒
50GB RDB → 约 50~150秒

加载期间 Redis 不处理任何请求
这是正常的，因为 Redis 还没初始化完成

# 1. 监控 fork 耗时
127.0.0.1:6379> INFO stats
# latest_fork_usec: 1532  ← 最近一次 fork 耗时（微秒）
# latest_fork_rate: 0

# 2. 调整 RDB 生成频率
save 900 1       # 15分钟一次（默认）
save 300 10      # 5分钟一次（默认）
save 60 10000    # 1分钟一次（默认）

大内存 Redis 实例建议：
1. 内存 > 20GB 时，延长 save 间隔
2. 避免高峰期自动触发 save
3. 手动在低峰期执行 BGSAVE
4. 确保机器内存 > Redis 内存 × 2（COW 需要额外内存）
5. 考虑关闭 RDB，改用 AOF everysec

写命令的执行路径：

Redis 执行命令 → 修改内存 → 追加到 AOF 缓冲区（内存）
                                      ↓
                             调用 write() 写入内核缓冲区
                                      ↓
                             调用 fsync() 刷到磁盘（物理写入）

# redis.conf 配置
appendfsync always     # 每次写命令都 fsync
appendfsync everysec   # 每秒 fsync 一次（默认）
appendfsync no         # 由操作系统决定

流程图：

SET user:1001 "张三"
→ 写入 AOF 缓冲区
→ write() 写入内核缓冲区
→ fsync() 刷到磁盘 ✅ ← 命令执行完毕后才返回给客户端

SET user:1002 "李四"
→ 同上流程

核心特点：
每条写命令都确认写入磁盘后才返回客户端

性能测试对比（SSD，单线程写入）：

appendfsync always:    ~2~3万 QPS 写入
appendfsync everysec:  ~8~10万 QPS 写入
appendfsync no:        ~10~12万 QPS 写入

always 策略性能下降 3~5 倍！

流程：

T0: SET user:1001 → 写入缓冲区，标记需要 fsync
T0: SET user:1002 → 同上
T0: SET user:1003 → 同上
   ...
T1: 每秒一次的 fsync 到来
   → 将这一秒内积累的所有数据一次性 fsync 到磁盘

核心特点：
1 秒内的数据先缓存在内存缓冲区
每秒调用 1 次 fsync，批量刷盘

数据丢失分析：
如果 Redis 在 fsync 之前崩溃（距离上次 fsync 最多 1 秒）：
- 丢失上一秒内的所有数据
- 因此最多丢失 1 秒数据

实际情况通常 < 1 秒
如果 fsync 时崩溃：丢失上次 fsync 到崩溃之间的数据
平均 ≈ 0.5 秒

流程：

SET user:1001 → write() 到内核缓冲区
            → 内核缓冲区满了？或者 30 秒到了？
            → 操作系统自动刷盘

核心特点：
Redis 只调用 write()，不主动调用 fsync()
刷盘时机完全由操作系统决定

数据丢失分析：
操作系统缓冲区默认保留 30 秒的数据
如果崩溃：
- 最多丢失 30 秒的数据
- 实际可能更多（如果内核缓冲区有未刷数据）

特殊风险：
写操作突然中断（断电），缓冲区数据全部丢失
可能丢失几分钟的作品（写操作频繁时）

策略       fsync时机    写入性能    数据丢失    磁盘负载    推荐场景
always    每条命令       低         最多1条命令   高        支付级场景（极少用）
everysec  每秒1次        高         最多1秒      中        通用生产环境（推荐）
no        操作系统决定    最高       最多30秒+    低        纯缓存（可接受大量丢数据）

正常情况下 everysec 由后台线程执行 fsync
但如果 fsync 耗时过长（如机械硬盘有寻道延迟）：
→ 后台线程执行 fsync 时被阻塞
→ 主线程再次尝试 fsync 也会被阻塞
→ → 导致主线程处理命令变慢

解决方案：
1. 使用 SSD（fsync 通常在 1~5ms）
2. 生产环境请勿使用机械硬盘做 Redis 持久化

# redis.conf
no-appendfsync-on-rewrite no

当 AOF 重写或 RDB 快照正在执行时：
- 磁盘 IO 负载很高
- 此时再频繁 fsync 会加剧性能问题

no-appendfsync-on-rewrite yes 的含义：
在重写过程中不执行 fsync
由操作系统自己决定何时刷盘
重写结束后恢复 fsync 策略

代价：重写期间的 AOF 数据安全性降低到 "no" 级别

虽然 no 策略不调用 fsync
但 write() 系统调用仍然有成本
只是让操作系统合并 fsync 操作

实际性能提升：
always  → everysec：提升 3~5 倍
everysec → no：提升 10%~30%（差距不大）

因为 Redis 主线程的瓶颈通常不在 fsync
而在命令执行本身

# 通用生产环境
appendfsync everysec              # 推荐，性能和安全兼顾
no-appendfsync-on-rewrite yes     # 重写时降级为 no，保护主线程

# 金融级别数据（谨慎评估性能损失后使用）
appendfsync always               # 每条命令都刷盘
# 配套：使用高性能 SSD，减少刷盘延迟

# 纯缓存场景（不开启持久化）
appendonly no                     # 关掉最好
# 或：
appendfsync no                    # 需要 AOF 但性能优先

RDB 是"拍照片"（保存某一时刻的全量数据）
AOF 是"记日记"（记录每个写命令，重启时重放）

flowchart LR
    A[客户端命令] --> B[Redis 执行命令]
    B --> C[写入内存数据结构]
    B --> D[追加到 AOF 缓冲区]
    D --> E{多久写入文件？}
    E -->|always| F[每次写入磁盘]
    E -->|everysec| G[每秒写入磁盘]
    E -->|no| H[由操作系统决定]

    D --> I[AOF 文件持续增长]
    I --> J{AOF 重写}
    J --> K[压缩 AOF 文件，去除冗余命令]

# 当执行 SET 时：
127.0.0.1:6379> SET user:1001 "张三"
+OK

# AOF 文件中记录：
*3
$3
SET
$10
user:1001
$6
张三

127.0.0.1:6379> BGREWRITEAOF
# 在后台启动 AOF 重写（不会阻塞主进程）

*3          ← 数组长度（3个参数）
$3          ← 第一个参数长度
SET         ← 第一个参数（命令）
$10         ← 第二个参数长度
user:1001   ← 第二个参数（Key）
$6          ← 第三个参数长度
张三        ← 第三个参数（Value）

实际 AOF 文件内容（可读）：

*2
$6
SELECT
$1
0
*3
$3
SET
$10
user:1001
$6
张三
*3
$3
SET
$10
user:1002
$3
李四

Redis 启动时 AOF 恢复流程：

1. 创建伪客户端（fake client）
2. 读取 AOF 文件
3. 解析 RESP 协议格式的命令
4. 逐条执行命令（模拟客户端操作）
5. 直到所有命令执行完毕

第 4 步是整个恢复过程最慢的环节：
RDB 恢复：直接加载内存 → 毫秒级
AOF 恢复：逐条重放命令 → 几秒到几分钟（取决于数据量）

假设有 1000 万条写操作记录：

RDB：
- 加载 1GB 的快照到内存 → 约 1~3 秒

AOF（未重写）：
- 1000 万条 SET 命令
- 每条执行 ≈ 几微秒
- 总时间 ≈ 几十秒到几分钟

AOF（已重写）：
- 合并后只剩几十万条记录
- 恢复速度 ≈ RDB 的 3~5 倍慢

优点：
1. 数据安全性高
   everysec 策略下最多丢失 1 秒的数据
   always 策略下最多丢失 1 条命令

2. 可读性强
   AOF 文件是文本协议格式
   可以直接用文本编辑器打开查看

3. 支持修复（redis-check-aof）
   如果 AOF 文件损坏（如磁盘故障）
   可以使用工具修复
   redis-check-aof --fix appendonly.aof

4. 支持逐条同步
   相比 RDB 的块级同步更灵活

缺点：
1. 文件体积大
   相同数据量的 AOF 比 RDB 大 3~10 倍
   即使重写后也通常比 RDB 大

2. 恢复速度慢
   逐条执行命令，不如 RDB 直接加载内存快

3. 写入开销
   AOF 每次写命令都有磁盘 IO 开销
   （但通过缓冲区 + 批量写入，影响可控）

测试环境：8GB 数据量，SSD 固态硬盘

维度            RDB              AOF（everysec）
写入性能        高（COW 在后台）   影响较小（缓冲区写入）
数据安全        最多丢失几分钟数据  最多丢失 1 秒数据
文件大小        1GB               ~3~6GB（取决于压缩）
恢复速度        ~2秒               ~15~60秒（取决于配置）
CPU 影响        fork 时的 COW     主进程追加写入（较轻微）
磁盘 IO         周期性写满 1GB     持续小量写入

Redis 启动时决定使用哪种方式恢复优先级的逻辑：

if AOF 功能开启 && AOF 文件存在:
    使用 AOF 恢复（优先）
else if RDB 文件存在:
    使用 RDB 恢复
else:
    数据为空（新实例）

为什么优先用 AOF？
因为 AOF 的数据更完整（丢失更少的数据）

flowchart TD
    A[Redis 持久化方案] --> B[RDB 快照]
    A --> C[AOF 日志]
    A --> D[混合持久化 4.0+]

    B --> B1["优点：恢复快、文件小"]
    B --> B2["缺点：可能丢失几分钟数据"]

    C --> C1["优点：最多丢1秒数据"]
    C --> C2["缺点：恢复慢、文件大"]

    D --> D1["RDB 做全量 + AOF 做增量"]
    D --> D2["恢复快 + 数据安全兼顾"]

维度            RDB                        AOF
文件数量        一个 dump.rdb               一个或多个 appendonly.aof
文件格式        二进制快照                   文本（RESP 协议）
文件大小        较小（紧凑二进制）            较大（文本协议，通常 3~5 倍）
数据完整性      可能丢失上一次快照后的所有数据  最多丢失 1 秒数据（或 1 条命令）
恢复速度        快（直接加载到内存）          慢（逐条执行命令）
写入性能        好（子进程写，COW）           中（主进程追加缓冲区 + fsync）
磁盘 IO        周期性大 IO                   持续写入（策略可控）
CPU 消耗        fork 时有一定开销              追加写入，消耗较少
是否可读        不可读（二进制）              可读（文本文件可查看）
修复支持        ❌ 修复困难                   ✅ redis-check-aof 修复
数据量级        适合所有规模                   大文件恢复慢是瓶颈

场景：支付系统、库存系统、账户系统

推荐：AOF（always 刷盘策略）
关闭：RDB（或仅作为补充）

理由：
- 最多丢一条命令
- 虽然恢复慢，但业务能接受重启时间
- 混合持久化更佳：RDB 加速恢复 + AOF 保护数据

# 配置示例
appendonly yes
appendfsync always    # 每条命令都刷盘
save ""               # 关闭 RDB（也可以保留，作为补充）

场景：读多写少的缓存、广告推荐、只读缓存

推荐：RDB（或关闭持久化）
关闭：AOF（可关可不关）

理由：
- 数据丢失可以通过回源数据库恢复
- 追求最高性能
- 恢复速度快，宕机后快速上线

# 配置示例
save 900 1
save 300 10
save 60 10000
appendonly no            # 关闭 AOF

场景：典型 Web 应用缓存、用户 Session、非关键计数器

推荐：混合持久化（RDB + AOF everysec）

# 配置示例
# RDB 配置
save 900 1
save 300 10

# AOF 配置（推荐 everysec）
appendonly yes
appendfsync everysec

# 混合持久化（4.0+）
aof-use-rdb-preamble yes

场景：临时数据、验证码、Session（可重建）、局域网 Redis

推荐：关闭持久化

理由：
- 省掉磁盘 IO 和 CPU 开销
- 提升 5%~15% 写入性能
- 数据丢了也不影响业务

# 配置示例
save ""              # 关闭所有 RDB
appendonly no        # 关闭 AOF

原则一：能开混合持久化就开
  4.0+ 标配 aof-use-rdb-preamble yes
  恢复快 + 数据全，两全其美

原则二：默认 everysec
  always 太激进（每次写都 fsync），性能下降明显
  no 太危险（操作系统决定，可能丢很多数据）
  everysec 是平衡点

原则三：大内存实例考虑 RDB
  50GB+ 的 Redis 用 AOF everysec 写入量不小
  AOF 文件大到几十 GB，恢复要几分钟

原则四：缓存场景只用 RDB 或不用
  你的业务接受数据丢失吗？
  能接受 → 不开持久化（最快方案）
  需要快速恢复备灾 → 用 RDB

原则五：流量低谷执行 RDB
  配置 save 参数，让 RDB 在流量低谷自动触发
  避免高峰期 fork 造成延迟尖刺

原则六：文件备份和监控
  无论选什么，定期备份 RDB/AOF 文件到异地
  监控磁盘使用率（AOF 增长可能导致磁盘满）

原则七：混合持久化不能替代备份
  混合持久化是指 Redis 内部的持久化方式
  不等于异地容灾或数据备份
  定期将 RDB 传输到备份服务器仍必不可少

# redis.conf 生产推荐配置

# 开启 RDB（用于快速恢复）
save 900 1
save 300 10
save 60 10000

# 开启 AOF（everysec 平衡性能和安全）
appendonly yes
appendfilename "appendonly.aof"
appendfsync everysec
no-appendfsync-on-rewrite no   # 重写时允许 fsync

# 自动 AOF 重写
auto-aof-rewrite-percentage 100   # AOF 比上次重写增长 100% 时触发
auto-aof-rewrite-min-size 64mb    # AOF 至少 64M 才考虑重写

# 混合持久化（4.0+ 关键）
aof-use-rdb-preamble yes

# 持久化文件路径
dir /var/lib/redis/
dbfilename dump.rdb

RDB 的核心思想：在某个时刻拍一张"内存快照"
相当于给 Redis 的内存数据拍了一张全貌照片

# SAVE：同步阻塞，主进程执行（不推荐生产环境使用）
127.0.0.1:6379> SAVE
# 处于阻塞状态，直到 RDB 生成完毕

# BGSAVE：后台异步，fork 子进程执行（推荐）
127.0.0.1:6379> BGSAVE
# 立即返回 "Background saving started"
# 子进程在后台生成 RDB 文件

# redis.conf
# 格式：save <秒数> <修改次数>
save 900 1       # 900秒内至少有1次写入操作
save 300 10      # 300秒内至少有10次写入操作
save 60 10000    # 60秒内至少有10000次写入操作

# 满足任意一条即触发 BGSAVE
# 可以关闭自动保存（注释所有 save 行）

# 主从复制：全量同步时，Master 自动 BGSAVE
# 执行 SHUTDOWN 时：如果没有开启 AOF，自动 SAVE 后关机
# DEBUG RELOAD：重新加载 RDB

flowchart TD
    A[执行 BGSAVE] --> B[父进程检查: 是否有子进程在持久化?]
    B -->|有| C[返回错误]
    B -->|无| D[调用 fork() 系统调用]
    D --> E[创建子进程]
    E --> F[子进程: 将内存数据写入临时 RDB 文件]
    F --> G[子进程: 用临时文件覆盖正式 RDB 文件]
    G --> H[子进程: 向父进程发送信号完成通知]
    H --> I[父进程: 更新持久化统计信息]

    subgraph "fork 时刻的父子进程状态"
        E --> J["父进程继续
处理客户端请求"]
        E --> K["子进程拥有
fork 时刻的内存快照"]
    end

fork 创建子进程时：
- 父子进程共享同一份物理内存（不是复制）
- 内存页标记为"只读"

当父进程需要修改某个内存页时：
1. 触发 page fault（缺页异常）
2. 内核复制该内存页
3. 父进程修改复制后的页面
4. 子进程看到的是修改前的页面（fork 时的快照）

未修改的页面：父子进程依然共享
修改的页面：复制后再修改

假设 Redis 占用 10GB 内存：
- BGSAVE 期间修改了 20% 的数据（2GB）
- 实际需要额外 2GB 内存用于 COW
- 机器需要至少有 12GB 可用内存

极端情况（全量修改）：
- 修改了 100% 的数据（10GB）
- 需要额外 10GB 内存
- 总共需要 20GB

公式：
RDB 期间额外内存 ≈ 修改数据量 × 页大小（4KB 的粒度）

fork 的耗时：
Redis 占用 10GB → fork 时间 ≈ 100~300ms
Redis 占用 50GB → fork 时间 ≈ 1~2s

fork 期间父进程阻塞（不能处理请求）
所以对于大内存实例，要关注 fork 耗时

RDB 文件结构：

| REDIS | rdb_version | databases | EOF | checksum |

databases:
| SELECTDB | db_number | key_value_pairs |
  key_value_pairs:
  | EXPIRETIME | expire_time | TYPE | key | value |
  | TYPE | key | value | ...

文件特点：
- 紧凑的二进制格式
- 不同数据类型有不同的编码方式
- 末尾有校验和（checksum）
- 支持压缩（某些类型用 LZF 压缩）

# 启动时自动恢复
# Redis 启动时会检查：
# 1. 如果配置了 AOF 且 AOF 文件存在 → 用 AOF 恢复（优先）
# 2. 否则如果 RDB 文件存在 → 用 RDB 恢复

# 手动恢复
# 将 dump.rdb 复制到 Redis 的数据目录（dir 配置项）
# 重启 Redis

优点：
1. 文件紧凑，适合备份
   RDB 是一个单一的二进制文件（dump.rdb）
   可以直接复制、传输、归档

2. 恢复速度快
   直接加载内存，比 AOF（需要逐条执行命令）快得多
   对于大型数据集，RDB 恢复比 AOF 快 10~20 倍

3. 对性能影响小
   BGSAVE 通过 fork 子进程生成 RDB
   父进程仅需处理 COW 开销

4. 跨版本兼容
   RDB 文件格式相对稳定

缺点：
1. 数据丢失风险大
   两次快照之间的数据可能丢失
   如每 5 分钟一次快照，故障最多丢失 5 分钟数据

2. 生成 RDB 的 COW 开销
   大内存实例 fork 可能耗时数秒（阻塞）
   COW 需要额外内存

3. 不支持逐 Key 级操作
   不能只恢复或备份某一个 Key
   要么全量备份，要么全量不备份

Redis 3.2 之前：
- ziplist：内存紧凑，但大量元素时增删改差（连锁更新风险）
- linked list（双向链表）：增删改好，但每个节点都是 24 字节指针，内存开销大

痛点：
- ziplist 太大时修改性能差
- linked list 每个节点都分配独立内存，碎片多、内存大
- 两种方案选择困难

flowchart LR
    subgraph "quicklist"
        A["head"] --> B["Node 1
[ziplist 50个元素]"]
        B --> C["Node 2
[ziplist 50个元素]"]
        C --> D["Node 3
[ziplist 50个元素]"]
        D --> E["...N 个 ziplist"]
        E --> F["tail"]
        F -.-> A
    end

每个 ziplist 节点存储一定数量的元素（默认 8KB 或 200 个元素）
多个 ziplist 通过双向链表连接起来

// quicklist 头
typedef struct quicklist {
    quicklistNode *head;       // 指向第一个 quicklistNode
    quicklistNode *tail;       // 指向最后一个 quicklistNode
    unsigned long count;       // 所有 ziplist 中的元素总数
    unsigned long len;         // quicklistNode 的数量（双向链表的长度）
    int fill : QL_FILL_BITS;  // ziplist 大小控制（正数：元素个数；负数：字节限制）
    unsigned int compress : QL_COMP_BITS;  // 压缩深度
    unsigned int bookmark_count : QL_BM_BITS;
    quicklistBookmark bookmarks[];
} quicklist;

// 每个 quicklistNode
typedef struct quicklistNode {
    struct quicklistNode *prev;      // 前驱节点
    struct quicklistNode *next;      // 后继节点
    unsigned char *zl;               // 指向 ziplist（或 listpack 在 7.0+）
    unsigned int sz;                 // ziplist 的字节数
    unsigned int count : 16;         // ziplist 中的元素数量
    unsigned int encoding : 2;       // 编码方式：1=RAW, 2=LZF 压缩
    unsigned int container : 2;      // 存储容器：1=NONE, 2=ZIPLIST
    unsigned int recompress : 1;     // 是否需要重新压缩
    unsigned int attempted_compress : 1;
    unsigned int extra : 10;
} quicklistNode;

# redis.conf
list-max-ziplist-size -2

fill 参数含义：
正数（如 5）：每个 ziplist 最多包含 5 个元素
负数（编码字节限制）：
  -1：4 KB
  -2：8 KB（默认值）
  -3：16 KB
  -4：32 KB
  -5：64 KB

fill 为 5 时：每个 ziplist 最多 5 个元素
  优点：增删改快（ziplist 内部操作少）
  缺点：链表节点多（更多指针开销）

fill 为 -2 时：每个 ziplist 最多 8KB
  优点：链表节点少（节约指针）
  缺点：ziplist 内容多（增删改内存拷贝大）

# redis.conf
list-compress-depth 0

compress 参数含义：
0：不压缩（默认值）
1：两头各 1 个节点不压缩，中间的节点用 LZF 算法压缩
2：两头各 2 个节点不压缩，中间的节点压缩
N：两头各 N 个节点不压缩

List 的使用模式通常是：
- 两端操作最频繁（LPUSH/LPOP/RPUSH/RPOP/LTRIM）
- 中间元素很少访问

压缩中间节点以内存换性能：
- 压缩后的 ziplist 可以节省 60%~90% 的内存
- 访问中间元素时先解压（略有开销，但很少发生）

# LPUSH 在头部插入
127.0.0.1:6379> LPUSH mylist a b c

执行步骤：
1. 检查 head 节点的 ziplist 是否有空间（不超 fill 限制）
2. 如果有 → 直接在 ziplist 头部插入
3. 如果 ziplist 满了 → 创建一个新的 quicklistNode
4. 新节点挂到 linked list 头部

当插入元素导致 ziplist 超过 fill 限制：
1. 将当前 ziplist 拆分为两个 ziplist
2. 在两个 ziplist 之间创建一个新的节点
3. 调整双向链表的指针

类比：ArrayList 扩容时复制 → ziplist 扩容时拆分

# 访问列表中间的元素
127.0.0.1:6379> LINDEX mylist 100

1. 定位到包含该元素的 quicklistNode
2. 检查该节点是否被 LZF 压缩
3. 如果是 → 解压缩 ziplist
4. 从 ziplist 中取出元素
5. 访问完毕后标记为 recompress
6. 下次访问重新压缩

优点：访问完不会立刻压缩，如果连续访问同一个节点只解压一次

维度         quicklist                原 LinkedList     原 ziplist
---
内存效率     ⭐⭐⭐⭐                  ⭐⭐              ⭐⭐⭐⭐⭐
（头尾访问不压缩时差一些， （每个节点 24 字节指针， 连续内存，零碎片
 中间压缩后与 ziplist 相当） 内存碎片多）

增删头尾     ⭐⭐⭐⭐⭐                ⭐⭐⭐⭐⭐        （头部插入需 O(N) 移动
（直接在 ziplist 操作，                                         内存）
 如满则新加节点）

增删中间     ⭐⭐⭐                   ⭐⭐⭐⭐⭐        ⭐⭐⭐
（拆分/合并 ziplist          （O(1) 修改指针）     （O(N) 内存移动）
 有 O(N) 开销）

遍历         ⭐⭐⭐⭐⭐                ⭐⭐⭐⭐            ⭐⭐⭐⭐
（分段遍历 + 连续遍历                                              （段内连续）
 ziplist 内部连续，比
 linked list 缓存友好）

范围查询     ⭐⭐⭐⭐                  ⭐⭐                ⭐⭐⭐⭐⭐
（定位到段后连续读）          （每个节点跳转）          （游标扫全量）

Redis 6.x 及之前：
quicklist → 内部使用 ziplist（有连锁更新风险）

Redis 7.0+：
quicklist → 内部使用 listpack（无连锁更新，更安全）

使用 ziplist/listpack 的场景：
- Hash：field 数量 < 512 且每个 field-value < 64 字节 → ziplist
- ZSet：元素数量 < 128 个且元素大小 < 64 字节 → ziplist
- List：元素数量 < 某个阈值 → quicklist 中的 ziplist

ziplist 的内存布局：

| zlbytes | zltail | zllen | entry1 | entry2 | ... | entryN | zlend |
|---------|--------|-------|--------|--------|-----|--------|-------|
| 4字节    | 4字节  | 2字节  | 可变    | 可变    |     | 可变    | 1字节 |

zlbytes：整个 ziplist 占用的字节数
zltail：最后一个 entry 的偏移量（用于从后往前遍历）
zllen：entry 的数量（如果 > 65535 则需要遍历得到真实数量）
zlend：结束标记（255）

每个 entry 的结构：

| prevlen | encoding | content |
|---------|----------|---------|

prevlen：前一个 entry 的长度（用于从后往前遍历）
  - 前一个节点长度 < 254 字节 → 1 字节
  - 前一个节点长度 >= 254 字节 → 5 字节（第 1 字节标记 254，后 4 字节存真实长度）

encoding：编码类型和数据长度
  - 字符串编码：前 2 位表示编码，后 6 位/2 字节/5 字节存数据长度
  - 整数编码：直接编码整数类型（如 int16/int32/int64）

content：实际的数据内容（字符串或整数）

flowchart LR
    A["前提：多个 entry 的 prevlen 恰好都是 1 字节"]
    B["场景：删除/插入一个元素"]
    C["问题：某个 entry 变长（1→5 字节）"]
    D["后果：后续 entry 的长度变化逐个传递"]

    A --> B --> C --> D

    D --> E["时间：O(N²) 的最差更新性能"]
    D --> F["空间：连续插入时多次 realloc"]

初始状态（每个 entry 的 prevlen 都刚好 1 字节）：

entry1 [prevlen=252] → entry2 [prevlen=252]
    前一个 252 字节，prevlen 用 1 字节（< 254）

意外情况：在 entry4 前面插入了一个很大的 entry（300 字节）

entry5 [prevlen=252 → 变成 prevlen=300 → 需要 5 字节]
    因为前一个 entry 的大小从 252 变成了 300
    prevlen 需要从 1 字节扩展到 5 字节
    这个扩展导致 entry5 本身也变大了 4 字节

→ entry6 的 prevlen 也因为 entry5 变大而需要扩展
→ entry7 的 prevlen 也因此需要扩展
→ ... 连锁反应，可能波及整个 ziplist

最坏情况：
- 如果 N 个 entry 的 prevlen 都是 [250, 253] 字节（刚好用 1 字节）
- 插入一个 > 254 字节的 entry
- 第一个 entry 的 prevlen → 变成 5 字节
- 该 entry 长度增加 4 字节 → 下一个 entry 的 prevlen 也需扩展
- ... 影响 N 个 entry → 时间复杂度 O(N²)
- 每次扩展都需要 realloc → 内存碎片

实际影响？
根据 antirez 的分析：
- 平均 prevlen 分布下发生连锁更新的概率极低
- 即使发生，影响范围也有限
- 但极端情况理论上存在

listpack 的内存布局：

| totbytes | num_elements | entry1 | entry2 | ... | entryN | end |
|----------|--------------|--------|--------|-----|--------|-----|
| 4字节     | 2字节        | 可变    | 可变    |     | 可变    | 1字节|

ziplist entry:     | prevlen | encoding | content |
listpack entry:    | encoding | content | backlen |

⚠️ 关键区别：
- ziplist：prevlen 记录前一个 entry 的长度（导致连锁更新）
- listpack：backlen 记录当前 entry 的长度（只影响自身）
  = 从后往前遍历时读取 backlen 知道要"跳过"多少字节

ziplist 的问题来源：
  prevlen 依赖前一个 entry 的大小
  前一个 entry 变化 → 当前 entry 的 prevlen 也要变化
  当前 entry 变化 → 下一个 entry 的 prevlen 也要变化
  → 连锁反应

listpack 的解决思路：
  backlen 只记录当前 entry 自身的长度
  当前 entry 大小变化 → 只更新自己的 backlen
  不影响任何其他 entry
  → 没有连锁更新

backlen 使用类似 UTF-8 的变长编码：
- 每个字节 7 位数据 + 1 位 continue 标记
- 较大值：用更多字节编码

例如：
值 0-127：1 字节
值 128-16383：2 字节
值 16384-2097151：3 字节
..

好处：backlen 本身不会因为数值变化而改变编码长度
（而 ziplist 的 prevlen 在阈值处跳跃：1 字节 → 5 字节）

维度         ziplist                      listpack
引入版本     Redis 2.x                     Redis 5.0
struct      |prevlen|enc|content|         |enc|content|backlen|
遍历方向     双向（prevlen 回溯）          双向（backlen 回溯）
连锁更新     有（设计缺陷）                无（根本解决）
最后 entry   
定位         zltail 偏移量                从 totbytes 计算
内存效率     略高（backlen 编码额外开销）   略低于 ziplist（但差异极小）
使用场景     Hash/ZSet/List(quicklist)     Redis Stream + 未来替代
演进趋势     逐渐被 listpack 取代          新的紧凑型数据结构标准

# Redis 7.0 默认配置变化

# 6.x 及之前：
# Hash 用 ziplist
hash-max-ziplist-entries 512
hash-max-ziplist-value 64

# 7.0 之后：
# Hash 默认使用 listpack（替代 ziplist）
hash-max-listpack-entries 512
hash-max-listpack-value 64

# ZSet 同理
zset-max-listpack-entries 128
zset-max-listpack-value 64

对用户的影响：
- ziplist 和 listpack 对用户完全透明
- 用户直接操作 HSET、ZADD 等命令
- Redis 内部自动判断是否转换（阈值内用紧凑结构，超了用哈希表/跳表）
- listpack 在 5.0 引入 Stream，7.0 全面替换 ziplist

Set 的两种底层存储：
1. intset（整数集合）— 元素全是整数且数量少
2. dict（字典/hashtable）— 有其他元素或数量大

转换条件（默认配置）：
set-max-intset-entries 512  # 超过 512 个元素 → 转为 hashtable
# 或插入非整数元素 → 转为 hashtable

// Redis intset 结构体
typedef struct intset {
    uint32_t encoding;   // 编码方式：INTSET_ENC_INT16 / INT32 / INT64
    uint32_t length;     // 元素个数
    int8_t contents[];   // 元素数组（柔性数组）
} intset;

INTSET_ENC_INT16 (2字节)：元素范围 [-32768, 32767]
INTSET_ENC_INT32 (4字节)：元素范围 [-2147483648, 2147483647]
INTSET_ENC_INT64 (8字节)：元素范围 [-2^63, 2^63-1]

intset 包含 4 个 int16 整数 [1, 3, 5, 7]：

| encoding=2 | length=4 | [1] | [3] | [5] | [7] |
|------------|----------|-----|-----|-----|-----|
| 4字节       | 4字节     | 2B  | 2B  | 2B  | 2B  |

总占用：4 + 4 + 4×2 = 16 字节

对比 hashtable 存储相同数据：
哈希表头：~64 字节
4 个 dictEntry：~4×32 = 128 字节
总计：~192 字节

intset 内存效率：是 hashtable 的 1/12

假设 intset 当前用 int16 编码（所有元素在 -32768~32767 之间）：

[1, 3, 5, 7]  → encoding = INT16

插入元素 70000（超出 int16 范围）：
→ 需要升级到 INT32

升级过程：
1. 检查新元素是否超出当前编码范围
2. 确定新编码（INT32）
3. 重新分配内存（每个元素从 2 字节变为 4 字节）
4. 将原数组元素转换为新编码并重新排列
5. 插入新元素
6. 更新 encoding 字段

127.0.0.1:6379> SADD nums 1 3 5 7
(integer) 4

127.0.0.1:6379> OBJECT ENCODING nums
"intset"

127.0.0.1:6379> SADD nums 70000  # 超出 int16 范围
(integer) 1

# 内部自动升级为 int32
# 用户无感知，但内存占用翻倍

升级前（int16，4个元素）：
| encoding=2 | length=4 | [1] [3] [5] [7] |
内存：4 + 4 + 4×2 = 16 字节

升级中（将原 4 个元素扩展为 4 字节）：
| encoding=4 | length=4 | [1] [3] [5] [7] |
内存：4 + 4 + 4×4 = 24 字节

升级后（插入 70000）：
| encoding=4 | length=5 | [1] [3] [5] [7] [70000] |
内存：4 + 4 + 5×4 = 28 字节

1. 升级不可逆
   即使删除了导致升级的 70000，编码不会降级回 int16
   （Redis 没有降级操作）

2. 升级耗时
   升级需要 O(N) 时间重新分配和复制所有元素
   但 intset 设计用于小数据量（<512），升级开销可控

3. 不支持降级
   原因：实现降级会引入复杂度，且 intset 本身是小数据量场景
   降级的收益有限

intset 存储 [7, 3, 1, 5]：

插入时：
1. 插入 7 → [7]
2. 插入 3 → 找到插入位置 [3, 7]
3. 插入 1 → [1, 3, 7]
4. 插入 5 → [1, 3, 5, 7]

始终保证有序 → 查找可以用二分查找

// intset 的查找操作使用二分查找
static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
    int min = 0, max = intrev32ifbe(is->length) - 1, mid;

    // 检查是否超出范围
    if (intrev32ifbe(is->length) == 0 || 
        value < _intsetGet(is, min) || 
        value > _intsetGet(is, max)) {
        *pos = intrev32ifbe(is->length);
        return 0;
    }

    // 二分查找
    while (max >= min) {
        mid = (min + max) / 2;
        int64_t cur = _intsetGet(is, mid);
        if (value > cur) {
            min = mid + 1;
        } else if (value < cur) {
            max = mid - 1;
        } else {
            *pos = mid;
            return 1;  // 找到
        }
    }
    *pos = min;  // 未找到，返回插入位置
    return 0;
}

flowchart TD
    A[SADD 添加元素] --> B{元素都是整数吗？}
    B -->|是| C{数量 < 512？}
    B -->|否| E[直接使用 dict]
    C -->|是| D[使用 intset 存储]
    C -->|否| E
    D --> F{后续插入非整数？}
    F -->|是| G[升级为 dict]
    D --> H{后续超过 512 个？}
    H -->|是| G

# 验证存储类型
127.0.0.1:6379> SADD myset 1 2 3 4 5
(integer) 5
127.0.0.1:6379> OBJECT ENCODING myset
"intset"

127.0.0.1:6379> SADD myset 6 7 8 ...  # 超过 512 个
127.0.0.1:6379> OBJECT ENCODING myset
"hashtable"  # 自动转换为 hashtable

1. 内存紧凑
   连续内存存储，无指针碎片
   相比 hash table 节省 90%+ 内存

2. 二分查找 O(logN) 
   有序数组 + 二分查找，小数据量性能极佳

3. 自动升级
   对用户透明，存储大整数时自动扩大编码

4. CPU 缓存友好
   连续内存，空间局部性好，CPU 缓存命中率高

数据结构         插入      查找      范围查询
平衡树（AVL）    O(logN)  O(logN)  O(logN+K)
红黑树          O(logN)  O(logN)  O(logN+K)
跳表            O(logN)  O(logN)  O(logN+K)
数组           O(N)      O(logN)   O(logN+K)（二分但插入慢）
链表           O(1)头插入 O(N)     O(N)

1. 实现简单
   跳表 ≈ 链表增强，代码容易理解和维护
   红黑树 ≈ 节点染色 + 旋转 + 分裂，实现复杂

2. 范围查询高效
   跳表找到起点后，沿链表遍历即可
   红黑树范围查询需要中序遍历，实现复杂

3. 区间锁粒度更细
   跳表只需要锁定相邻节点
   树可能需要锁定更广范围

普通链表查找：
┌───┐   ┌───┐   ┌───┐   ┌───┐   ┌───┐   ┌───┐
│ 1 │ → │ 3 │ → │ 5 │ → │ 7 │ → │ 9 │ → │ 15│ → null
└───┘   └───┘   └───┘   └───┘   └───┘   └───┘
查找 9：从 1→3→5→7→9，遍历 5 次，O(N)

flowchart TD
    subgraph "L2 最高层（20%节点有索引）"
        A1["head
-∞"] --> A2[3] --> A3[9] --> A4["null"]
    end
    subgraph "L1 中间层（50%节点有索引）"
        B1["head
-∞"] --> B2[1] --> B3[3] --> B4[5] --> B5[7] --> B6[9] --> B7[15] --> B8["null"]
    end
    subgraph "L0 完整数据层"
        C1["head
-∞"] --> C2[1] --> C3[3] --> C4[5] --> C5[7] --> C6[9] --> C7[15] --> C8["null"]
    end

    A1 -.-> B1
    B2 -.-> C2
    B3 -.-> C3
    A2 -.-> B3
    B4 -.-> C4
    B5 -.-> C5
    A3 -.-> B6
    B7 -.-> C7

查找 9 的路径（跳表）：

从 L2 开始：head → 3 → 9！
只用了 2 次跳跃就找到了（普通链表要 5 次）

// Redis 跳表节点
typedef struct zskiplistNode {
    sds ele;                              // 存储的实际数据（ZSet 的 member）
    double score;                         // 排序分数
    struct zskiplistNode *backward;       // 后退指针（L0 层使用，用于逆向遍历）
    struct zskiplistLevel {
        struct zskiplistNode *forward;    // 前进指针
        unsigned int span;                // 跨度（到下一个节点的距离）
    } level[];                            // 层级数组（柔性数组）
} zskiplistNode;

typedef struct zskiplist {
    struct zskiplistNode *header, *tail;  // 头尾指针
    unsigned long length;                 // 节点总数
    int level;                            // 当前最大层数
} zskiplist;

// Redis 的随机层高生成算法
int zslRandomLevel(void) {
    int level = 1;
    // 每次有 25%（1/4）的概率增加一层
    while ((random() & 0xFFFF) < 0.25 * 0xFFFF) {
        level += 1;
    }
    // 最终不超过最大层数（ZSKIPLIST_MAXLEVEL = 32）
    return (level < ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

概率分布：
层高=1：75%
层高=2：18.75%
层高=3：4.69%
层高=4：1.17%
...
层高=32：几乎为零

平均层高 ≈ 1/(1-0.25) = 1.33

空间效率：每个节点的平均指针数 ≈ 1.33
时间复杂度：查找 O(logN)（基于概率保证）

查找分数为 9 的元素：

从最高层 L2 开始：
1. L2: head(3) → 3. 3 < 9 且 3 的 L2 下一个是 9 → 进入 L1
2. L1: 3 → 5. 5 < 9 且 5 的 L1 下一个是 7 → 到 7
3. L1: 7 → 9. 7 < 9 且 7 的 L1 下一个是 9 → 到 9
4. 9 == 9 → 找到！

路径：3次比较
普通链表需要 5 次

# ZADD leaderboard 88 "user_1002"
# Redis 内部执行：
ZADD leaderboard 88 "user_1002"

插入步骤：
1. 查找插入位置 → O(logN)
2. 创建新节点（随机生成层高）
3. 调整前后节点的指针和跨度 → O(1)
4. 更新最高层数

不用像平衡树那样做旋转/颜色调整

ZREM leaderboard "user_1002"

删除步骤：
1. 查找目标节点 → O(logN)
2. 调整前后节点指针和跨度
3. 释放节点内存
4. 更新最高层数（如果需要）

# ZSet 的范围查询是跳表最大的优势之一
ZRANGEBYSCORE leaderboard 80 100  # 分数 80~100

1. 找到起始分数 80 的位置 → O(logN)
2. 从该位置沿 L0 层 forward 指针往后遍历 → O(K)
3. 直到遇到 >100 的元素停止

时间复杂度：O(logN + K)
N = 总节点数，K = 返回的节点数

维度       跳表            平衡树（AVL/红黑树）
实现难度   简单（约 400 行代码）  复杂（约 2000+ 行代码）
平均查找   O(logN)         O(logN)
插入      O(logN)          O(logN) + 旋转调整
删除      O(logN)          O(logN) + 旋转调整
范围查询   O(logN+K)       O(logN+K)（需中序遍历）
排序      ⭐天然有序       ⭐天然有序
内存      指针平均 1.33 个  每个节点 2~3 个指针 + 颜色标记
区间锁    细粒度           粗粒度

import "unsafe"

// Sizeof：返回类型的大小（字节）
unsafe.Sizeof(x)     // → uintptr

// Alignof：返回类型的对齐要求（字节）
unsafe.Alignof(x)    // → uintptr

// Offsetof：返回结构体字段的偏移量（仅结构体字段）
unsafe.Offsetof(s.f) // → uintptr

package main

import (
    "fmt"
    "unsafe"
)

func main() {
    var s string     // string: 16 字节（ptr + len）
    var sl []int     // slice: 24 字节（ptr + len + cap）
    var m map[int]int // map: 8 字节（指向 hmap 的指针）

    fmt.Printf("string: %d\n", unsafe.Sizeof(s))   // 16
    fmt.Printf("slice:  %d\n", unsafe.Sizeof(sl))  // 24
    fmt.Printf("map:    %d\n", unsafe.Sizeof(m))   // 8

    // 注意：s 是字符串头，实际字符数据在堆上
    s = "hello world" // Sizeof(s) 仍然是 16
}

func checkAlign() {
    fmt.Printf("int8:  align=%d\n", unsafe.Alignof(int8(0)))   // 1
    fmt.Printf("int16: align=%d\n", unsafe.Alignof(int16(0)))  // 2
    fmt.Printf("int64: align=%d\n", unsafe.Alignof(int64(0)))  // 8

    // 结构体的对齐等于其最大字段对齐
    type S struct { a bool; b int64 }
    fmt.Printf("struct S align=%d\n", unsafe.Alignof(S{}))    // 8
}

type Packet struct {
    Version uint8  // 偏移 0
    ID      uint16 // 偏移 2（有 1 字节填充）
    Length  uint32 // 偏移 4
    Data    [4]byte // 偏移 8
    CRC     uint32 // 偏移 12
}

func main() {
    p := Packet{}
    fmt.Printf("Version: offset=%d size=%d\n",
        unsafe.Offsetof(p.Version), unsafe.Sizeof(p.Version))
    fmt.Printf("ID:      offset=%d size=%d\n",
        unsafe.Offsetof(p.ID), unsafe.Sizeof(p.ID))
    fmt.Printf("Length:  offset=%d size=%d\n",
        unsafe.Offsetof(p.Length), unsafe.Sizeof(p.Length))
    fmt.Printf("Data:    offset=%d size=%d\n",
        unsafe.Offsetof(p.Data), unsafe.Sizeof(p.Data))
    fmt.Printf("CRC:     offset=%d size=%d\n",
        unsafe.Offsetof(p.CRC), unsafe.Sizeof(p.CRC))
    fmt.Printf("Total:   size=%d\n", unsafe.Sizeof(p))
    // Total: size=16
}

Version: offset=0 size=1
ID:      offset=2 size=2    # 偏移 1 被 padding 填补
Length:  offset=4 size=4
Data:    offset=8 size=4
CRC:     offset=12 size=4
Total:   size=16            # 16 = 13（数据）+ 3（尾部 padding）

graph LR
    subgraph "Packet 内存布局"
        P0["偏移 0
Version(1B)"] --> P1["偏移 1-2
🟤 padding(1B)"]
        P1 --> P2["偏移 2-4
ID(2B)"]
        P2 --> P3["偏移 4-8
Length(4B)"]
        P3 --> P4["偏移 8-12
Data(4B)"]
        P4 --> P5["偏移 12-16
CRC(4B)"]
    end