07-数据库与中间件

7. 数据库与中间件

7.1 核心知识点

MySQL 索引原理（必考！）

B+树结构：

为什么用 B+ 树而不是 B 树/红黑树/Hash？

1. B+ 树：
   - 非叶子节点只存索引，不存数据 → 每页能存更多索引 → 树更矮
   - 叶子节点用链表连接 → 范围查询高效
   - 3-4 层就能存千万级数据

2. 红黑树：
   - 二叉树，层数太深 → 磁盘 IO 次数多

3. Hash：
   - 只能精确匹配，不支持范围查询

聚簇索引 vs 非聚簇索引（必考！）：

类型	说明	InnoDB	特点
聚簇索引	索引和数据存在一起	主键索引	叶子节点就是数据行
非聚簇索引	索引存主键值	普通索引	叶子节点存主键，需回表

聚簇索引结构图：

B+ 树（主键索引）：

        [10 | 20 | 30]          ← 非叶子节点（只存索引）
       /      |      \
     [5,10] [15,20] [25,30]     ← 叶子节点（存完整数据行！）
       ↓       ↓       ↓
    整行数据  整行数据  整行数据   ← 查到即返回，一次搞定

非聚簇索引结构图：

B+ 树（name 索引）：

        [李 | 王 | 张]           ← 非叶子节点
       /      |      \
    [李,3] [王,1] [张,2]         ← 叶子节点存 (name, 主键id)
       ↓       ↓       ↓
     id=3    id=1    id=2       ← 只存主键，不存完整数据！

---

回表（重要！）：通过非聚簇索引找到主键 → 再通过主键找到数据（两次查询）

-- 假设 name 有索引
SELECT * FROM user WHERE name = '张三';
 
-- 执行过程：
-- 1️⃣ 先走 name 索引，找到 name='张三' 对应的 id=2
-- 2️⃣ 再拿 id=2 去主键索引（聚簇索引）查完整数据
-- ❌ 两次 B+ 树查询 = 回表 = 性能损耗

回表过程图解：

   name 索引              主键索引
  ┌─────────┐           ┌─────────┐
  │ 张三→id=2│ ──────→  │ id=2→✅ │ → 返回完整数据
  └─────────┘   回表     └─────────┘

---

覆盖索引（优化回表）：查询的字段都在索引中，不需要回表

-- 假设有联合索引 (name, age)
CREATE INDEX idx_name_age ON user(name, age);
 
-- ✅ 覆盖索引，不回表：
SELECT name, age FROM user WHERE name = '张三';
-- 因为 name 和 age 都在索引里，直接返回！
 
-- ❌ 需要回表：
SELECT * FROM user WHERE name = '张三';
-- 因为还需要其他字段（如 email），索引里没有

面试回答模板：

“InnoDB 的主键索引是聚簇索引，叶子节点直接存数据行。普通索引是非聚簇索引，叶子节点只存主键值。用普通索引查询时需要回表，优化方法是设计覆盖索引，让查询字段都在索引里。”

最左前缀原则：

-- 联合索引 (a, b, c)
WHERE a = 1                    -- ✅ 使用索引
WHERE a = 1 AND b = 2          -- ✅ 使用索引
WHERE a = 1 AND b = 2 AND c = 3 -- ✅ 使用索引
WHERE b = 2                    -- ❌ 不使用索引（跳过a）
WHERE a = 1 AND c = 3          -- ⚠️ 只用到 a

索引失效场景：

1. 对索引列使用函数：WHERE YEAR(create_time) = 2024
2. 隐式类型转换：WHERE phone = 13800138000（phone是varchar）
3. 左模糊查询：WHERE name LIKE '%张'
4. 使用 OR：WHERE a = 1 OR b = 2（b无索引）
5. 使用 != 或 NOT IN
6. 索引列参与运算：WHERE age + 1 = 18

MySQL 事务（ACID）

特性	英文	中文名称	含义	实现方式	记忆口诀
A	Atomicity	原子性	全部成功或全部失败	undo log	”全有全无”
C	Consistency	一致性	事务前后数据完整性不变	业务逻辑保证	”账不能平不了”
I	Isolation	隔离性	事务之间互不干扰	锁 + MVCC	”各干各的”
D	Durability	持久性	提交后永久保存	redo log	”掉电不丢”

隔离级别：

级别	脏读	不可重复读	幻读
READ UNCOMMITTED	✅	✅	✅
READ COMMITTED	❌	✅	✅
REPEATABLE READ (默认)	❌	❌	✅(部分解决)
SERIALIZABLE	❌	❌	❌

脏读、不可重复读、幻读的区别（必考！）：

问题	现象	原因	记忆口诀
脏读	读到别人未提交的数据	别人回滚了，你读的是脏数据	”读了脏东西”
不可重复读	同一条数据，两次读值不同	别人 UPDATE 了	”数据被改了”
幻读	同一范围，两次读行数不同	别人 INSERT/DELETE 了	”凭空多/少了”

举例理解：

【不可重复读】
事务A第一次查：用户余额 = 100
           事务B修改：UPDATE 余额 = 50，提交
事务A第二次查：用户余额 = 50  ← 同一条数据，值变了！

【幻读】
事务A第一次查：订单数 = 10 条
           事务B插入：INSERT 一条新订单，提交
事务A第二次查：订单数 = 11 条  ← 凭空多了一条！

---

MVCC 多版本并发控制（面试高频追问）

MVCC 是什么？

• 全称：Multi-Version Concurrency Control
• 核心思想：读写不阻塞，通过多版本实现

实现原理：

每行数据有隐藏列：
- DB_TRX_ID：创建/最后修改该行的事务ID
- DB_ROLL_PTR：指向 undo log 的指针（历史版本）

当事务读取时：
1. 根据"快照读"规则判断该版本是否可见
2. 如果不可见，通过 undo log 找到更早的版本
3. 不同事务看到的是不同"快照"

核心价值：

• 读操作不加锁，不阻塞写
• 写操作不阻塞读
• 大大提高并发性能

面试简答：

“MVCC 通过在每行数据保存多个版本，让读操作可以读取历史快照，从而实现读写不阻塞，提高并发性能。“

---

Undo Log 与 Redo Log（事务核心机制）

Undo Log 和 Redo Log 是 InnoDB 实现 ACID 特性的核心机制：

特性	Redo Log	Undo Log
记录内容	修改后的值（物理日志）	修改前的值（逻辑日志）
作用	崩溃恢复（重做）	事务回滚 + MVCC
保证的特性	持久性（D）	原子性（A）、隔离性（I）
生命周期	循环覆盖（Checkpoint后）	事务提交/无读取后清理

---

Redo Log（重做日志）

作用：保证事务的 持久性（Durability），用于 崩溃恢复

WAL（Write-Ahead Logging）原则：

┌─────────────────────────────────────────────────────────────┐
│                      Write-Ahead Logging                     │
├─────────────────────────────────────────────────────────────┤
│  1. 事务修改数据前，先将修改记录写入 Redo Log Buffer         │
│  2. 事务提交时，Redo Log Buffer 强制刷盘（fsync）            │
│  3. 数据页可以延迟刷盘（异步写入）                           │
│  4. 崩溃恢复时，根据 Redo Log 重放未持久化的修改             │
└─────────────────────────────────────────────────────────────┘

为什么用 WAL？
- 数据页随机写入（慢），Redo Log 顺序写入（快）
- 即使数据页未刷盘，事务也能保证持久性

Redo Log 记录结构：

type RedoLogRecord struct {
    LSN        uint64  // Log Sequence Number，全局唯一递增
    TxID       uint64  // 事务ID
    PageID     uint64  // 被修改的页面ID
    Offset     uint16  // 页内偏移
    Len        uint16  // 修改长度
    NewValue   []byte  // 修改后的值（物理层面）
}

关键机制：

1. Checkpoint（检查点）：
   - 定期将脏页刷盘，记录 checkpoint LSN
   - 恢复时只需从 checkpoint LSN 开始重放
   - 减少恢复时间

2. Group Commit（组提交）：
   - 多个事务的 Redo Log 合并一次刷盘
   - 提升 IO 效率

---

Undo Log（回滚日志）

作用：保证事务的 原子性（Atomicity），支持 事务回滚 和 MVCC

Undo Log 链表结构：

   数据行               Undo Log Chain（版本链）
  ┌─────────┐          ┌─────────┐    ┌─────────┐
  │ 当前值  │─────────>│ 旧值 V2 │───>│ 旧值 V1 │───> NULL
  │ V3      │          │ TxID=2  │    │ TxID=1  │
  │ DB_ROLL │          └─────────┘    └─────────┘
  │ _PTR    │
  └─────────┘

- 每次 UPDATE/DELETE 都会生成一条 Undo Log
- 多个版本形成链表，支持 MVCC 读取历史版本

Undo Log 记录结构：

type UndoLogRecord struct {
    UndoNo     uint64  // Undo 日志序号
    TxID       uint64  // 事务ID
    TableID    uint64  // 表ID
    Type       uint8   // INSERT / UPDATE / DELETE
    PrimaryKey []byte  // 主键值
    OldValue   []byte  // 修改前的值（逻辑层面）
    NextUndo   *UndoLogRecord // 链表指针
}

MVCC 读取流程：

func ReadWithMVCC(row *Row, readView *ReadView) *Row {
    current := row
    for current != nil {
        if current.TxID.IsVisibleTo(readView) {
            return current  // 找到可见版本
        }
        current = current.UndoLog  // 沿 Undo 链回溯
    }
    return nil
}

两种 Undo Log：

类型	生成场景	清理时机
Insert Undo	INSERT 操作	事务提交后立即删除
Update Undo	UPDATE/DELETE	无事务需要读取该版本后，由 Purge 线程异步清理

---

崩溃恢复流程

┌──────────────────────────────────────────────────────────┐
│                    崩溃恢复 (Crash Recovery)              │
├──────────────────────────────────────────────────────────┤
│  1. 从最近 Checkpoint 读取 Redo Log                       │
│                    ↓                                     │
│  2. REDO 阶段：重放所有 Redo Log（不管事务是否提交）       │
│                    ↓                                     │
│  3. UNDO 阶段：回滚所有未提交事务（使用 Undo Log）         │
│                    ↓                                     │
│  4. 数据库恢复到一致状态                                  │
└──────────────────────────────────────────────────────────┘

---

InnoDB 存储位置

日志	存储位置	说明
Redo Log	ib_logfile0, ib_logfile1	循环写入，固定大小
Undo Log	系统表空间 / Undo 表空间	undo_001 等文件

面试简答：

“Redo Log 用于崩溃恢复，记录修改后的值，保证持久性；Undo Log 用于事务回滚和 MVCC，记录修改前的值，保证原子性和隔离性。事务提交时 Redo Log 必须刷盘，而数据页可以延迟刷盘。“

MySQL vs PostgreSQL

特性	MySQL	PostgreSQL
定位	简单、快速、互联网主流	功能丰富、标准遵从
默认隔离级别	REPEATABLE READ	READ COMMITTED
MVCC实现	undo log	多版本tuple
JSON支持	5.7+ 支持	原生支持，更强大（JSONB）
全文搜索	需要插件	内置支持
复杂查询	一般	更强（窗口函数、CTE等）
存储引擎	InnoDB/MyISAM等	单一（可扩展）
适用场景	Web应用、读多写少	复杂业务、数据分析
Go驱动	go-sql-driver/mysql	lib/pq 或 pgx

Go 连接 PostgreSQL：

import (
    "database/sql"
    _ "github.com/lib/pq"  // 或 github.com/jackc/pgx/v5
)
 
dsn := "host=localhost port=5432 user=postgres password=xxx dbname=mydb sslmode=disable"
db, err := sql.Open("postgres", dsn)

主要区别记忆：

MySQL    = 简单够用，互联网标配
PostgreSQL = 功能强大，复杂业务首选

Redis 数据结构（必考！）

类型	底层实现	典型场景	命令示例
String	SDS	缓存、计数器、分布式锁	SET/GET/INCR
Hash	哈希表/ziplist	对象存储（用户信息）	HSET/HGET/HMGET
List	双向链表/quicklist	消息队列、最新列表	LPUSH/RPOP/LRANGE
Set	哈希表/intset	去重、共同好友	SADD/SMEMBERS/SINTER
ZSet	跳表+哈希表	排行榜、延迟队列	ZADD/ZRANGE/ZRANK

String（字符串）

底层：SDS（Simple Dynamic String）

SDS 比 C 字符串的优势：
1. O(1) 获取长度（len 字段）
2. 二进制安全（可存图片、序列化数据）
3. 自动扩容，避免缓冲区溢出
4. 惰性空间释放，减少内存分配次数

典型场景：

// 1. 缓存
rdb.Set(ctx, "user:1001", userJSON, 1*time.Hour)
 
// 2. 计数器（原子操作）
rdb.Incr(ctx, "article:1001:views")    // 文章浏览量
rdb.IncrBy(ctx, "user:1001:points", 10) // 用户积分
 
// 3. 分布式锁
rdb.SetNX(ctx, "lock:order:123", "uuid", 30*time.Second)
 
// 4. 限流（计数器方式）
count, _ := rdb.Incr(ctx, "limit:user:1001").Result()
if count > 100 {
    return errors.New("请求过于频繁")
}
rdb.Expire(ctx, "limit:user:1001", 1*time.Minute)

Hash（哈希）

适合存储对象：比 String 存 JSON 更省内存，支持单字段更新

// 存储用户信息
rdb.HSet(ctx, "user:1001", map[string]interface{}{
    "name":  "张三",
    "age":   25,
    "email": "zhangsan@example.com",
})
 
// 获取单个字段
name, _ := rdb.HGet(ctx, "user:1001", "name").Result()
 
// 更新单个字段（无需读取整个对象）
rdb.HIncrBy(ctx, "user:1001", "age", 1)
 
// 获取所有字段
user, _ := rdb.HGetAll(ctx, "user:1001").Result()

List（列表）

双向链表：两端操作 O(1)，适合队列场景

// 1. 消息队列（简单实现）
rdb.LPush(ctx, "queue:tasks", "task1", "task2")  // 生产者：左边插入
task, _ := rdb.RPop(ctx, "queue:tasks").Result() // 消费者：右边取出
 
// 2. 阻塞获取（消费者等待）
task, _ := rdb.BRPop(ctx, 0, "queue:tasks").Result() // 0 表示永久等待
 
// 3. 最新消息列表
rdb.LPush(ctx, "news:latest", newsJSON)
rdb.LTrim(ctx, "news:latest", 0, 99)  // 只保留最新100条
news, _ := rdb.LRange(ctx, "news:latest", 0, 9).Result() // 获取最新10条

Set（集合）

无序不重复：适合去重、交集/并集运算

// 1. 标签/分类
rdb.SAdd(ctx, "article:1001:tags", "Go", "Redis", "后端")
 
// 2. 共同好友
rdb.SAdd(ctx, "user:1001:friends", "2001", "2002", "2003")
rdb.SAdd(ctx, "user:1002:friends", "2001", "2004", "2005")
 
// 求交集：共同好友
common, _ := rdb.SInter(ctx, "user:1001:friends", "user:1002:friends").Result()
// 结果：["2001"]
 
// 3. 抽奖（随机取出）
winner, _ := rdb.SPop(ctx, "lottery:users").Result()  // 取出一个
winners, _ := rdb.SRandMemberN(ctx, "lottery:users", 3).Result() // 随机3个不取出

ZSet（有序集合）

核心：跳表 + 哈希表，按 score 排序，O(logN) 插入和查询

// 1. 排行榜
rdb.ZAdd(ctx, "rank:points", redis.Z{Score: 1000, Member: "user:1001"})
rdb.ZAdd(ctx, "rank:points", redis.Z{Score: 2000, Member: "user:1002"})
 
// 获取前10名（分数从高到低）
top10, _ := rdb.ZRevRangeWithScores(ctx, "rank:points", 0, 9).Result()
 
// 获取用户排名
rank, _ := rdb.ZRevRank(ctx, "rank:points", "user:1001").Result()
 
// 增加分数
rdb.ZIncrBy(ctx, "rank:points", 100, "user:1001")
 
// 2. 延迟队列（score 存时间戳）
// 添加延迟任务
rdb.ZAdd(ctx, "delay:tasks", redis.Z{
    Score:  float64(time.Now().Add(5*time.Minute).Unix()),
    Member: "send_email:1001",
})
 
// 消费：获取已到期的任务
tasks, _ := rdb.ZRangeByScore(ctx, "delay:tasks", &redis.ZRangeBy{
    Min: "0",
    Max: strconv.FormatInt(time.Now().Unix(), 10),
}).Result()

为什么 ZSet 用跳表而不是红黑树？

1. 范围查询更快：跳表 O(logN + M)，红黑树需要中序遍历
2. 实现更简单：跳表代码量少，易维护
3. 并发友好：跳表更适合并发修改

缓存三大问题（必考！）

问题	现象	原因	解决方案
穿透	查询不存在的数据，直接打到DB	恶意攻击或业务漏洞	布隆过滤器、缓存空值(短TTL)
击穿	热点key过期，大量请求打到DB	热点数据过期	互斥锁、永不过期+异步更新
雪崩	大量key同时过期，DB压力骤增	同一时间大批缓存失效	过期时间加随机值、多级缓存

缓存穿透解决方案代码：

// 方案1：缓存空值
func GetUser(ctx context.Context, userID string) (*User, error) {
    // 1. 查缓存
    val, err := rdb.Get(ctx, "user:"+userID).Result()
    if err == nil {
        if val == "" {  // 缓存了空值
            return nil, nil
        }
        var user User
        json.Unmarshal([]byte(val), &user)
        return &user, nil
    }
    
    // 2. 查数据库
    user, err := db.GetUser(userID)
    if err != nil {
        return nil, err
    }
    
    // 3. 写入缓存
    if user == nil {
        rdb.Set(ctx, "user:"+userID, "", 5*time.Minute)  // 缓存空值，短TTL
    } else {
        data, _ := json.Marshal(user)
        rdb.Set(ctx, "user:"+userID, data, 1*time.Hour)
    }
    
    return user, nil
}

布隆过滤器：

用位数组 + 多个哈希函数判断元素"一定不存在"或"可能存在"
- 查询时：多个哈希位都为1 → 可能存在
- 查询时：任意哈希位为0 → 一定不存在
- 优点：空间效率极高
- 缺点：有误判率，不能删除

Redis 持久化

方式	原理	优点	缺点
RDB	定时快照	恢复快、文件小	可能丢失最后一次快照后的数据
AOF	记录写命令	数据更安全	文件大、恢复慢
混合	RDB+增量AOF	兼顾速度和安全	Redis 4.0+

RDB触发方式：

1. 手动：BGSAVE（后台）/ SAVE（阻塞）
2. 配置自动触发：save 900 1（900秒内1次修改）
3. 关闭Redis时自动触发

AOF 三种同步策略：

appendfsync always   # 每次写入都同步，最安全但最慢
appendfsync everysec # 每秒同步一次（推荐）
appendfsync no       # 由操作系统决定，最快但可能丢数据

分布式锁（Redis实现）

// 加锁：SET key value NX EX 30
ok, _ := rdb.SetNX(ctx, "lock:order:123", "uuid", 30*time.Second).Result()
if !ok {
    return errors.New("获取锁失败")
}
 
// 业务逻辑...
 
// 解锁：Lua 脚本保证原子性（防止误删别人的锁）
script := `
if redis.call("get", KEYS[1]) == ARGV[1] then
    return redis.call("del", KEYS[1])
else
    return 0
end
`
rdb.Eval(ctx, script, []string{"lock:order:123"}, "uuid")

为什么解锁要用 Lua 脚本？

如果用两条命令：
1. GET lock:order:123  → "uuid"
2. DEL lock:order:123

在 1 和 2 之间，锁可能已过期被其他线程获取，导致误删别人的锁！
Lua 脚本保证原子性，要么都执行，要么都不执行。

Redis 高可用

方案	说明	适用场景
主从复制	一主多从，读写分离	读多写少
哨兵模式	自动故障转移	高可用要求
Cluster	分片，水平扩展	大数据量

消息队列对比

特性	RabbitMQ	Kafka
吞吐量	万级	百万级
消息顺序	不保证	分区内有序
消息持久化	支持	支持
适用场景	业务解耦、任务队列	日志收集、大数据

7.2 常见面试题

Q1: 如何在Go中使用MySQL？

解题思路：

1. 说明Go中MySQL的常用驱动
2. 演示基本的CRUD操作
3. 说明如何使用连接池
4. 展示事务处理

代码实现：

package main
 
import (
    "database/sql"
    "fmt"
    "log"
    "time"
    
    _ "github.com/go-sql-driver/mysql"
)
 
type User struct {
    ID    int
    Name  string
    Email string
    Age   int
}
 
func main() {
    dsn := "user:password@tcp(127.0.0.1:3306)/dbname?parseTime=true"
    db, err := sql.Open("mysql", dsn)
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()
    
    err = db.Ping()
    if err != nil {
        log.Fatal(err)
    }
    
    db.SetMaxOpenConns(25)
    db.SetMaxIdleConns(25)
    db.SetConnMaxLifetime(5 * time.Minute)
    
    _, err = db.Exec(`
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            name VARCHAR(100) NOT NULL,
            email VARCHAR(100) UNIQUE NOT NULL,
            age INT
        )
    `)
    if err != nil {
        log.Fatal(err)
    }
    
    result, err := db.Exec(
        "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
        "Alice", "alice@example.com", 25,
    )
    if err != nil {
        log.Fatal(err)
    }
    
    id, _ := result.LastInsertId()
    fmt.Printf("Inserted user with ID: %d\n", id)
    
    var user User
    err = db.QueryRow(
        "SELECT id, name, email, age FROM users WHERE id = ?",
        id,
    ).Scan(&user.ID, &user.Name, &user.Email, &user.Age)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("User: %+v\n", user)
    
    rows, err := db.Query("SELECT id, name, email, age FROM users")
    if err != nil {
        log.Fatal(err)
    }
    defer rows.Close()
    
    for rows.Next() {
        var u User
        err := rows.Scan(&u.ID, &u.Name, &u.Email, &u.Age)
        if err != nil {
            log.Fatal(err)
        }
        fmt.Printf("User: %+v\n", u)
    }
    
    tx, err := db.Begin()
    if err != nil {
        log.Fatal(err)
    }
    
    _, err = tx.Exec(
        "UPDATE users SET age = ? WHERE id = ?",
        26, id,
    )
    if err != nil {
        tx.Rollback()
        log.Fatal(err)
    }
    
    err = tx.Commit()
    if err != nil {
        log.Fatal(err)
    }
    
    _, err = db.Exec("DELETE FROM users WHERE id = ?", id)
    if err != nil {
        log.Fatal(err)
    }
}

常见错误分析：

• 错误1：忘记关闭数据库连接
• 错误2：不理解事务的提交和回滚
• 错误3：SQL注入漏洞

Q2: 如何在Go中使用Redis？

常用客户端：

• go-redis/redis：功能最全，推荐使用
• redigo：老牌客户端，API 偏底层

代码实现（带详细注释）：

package main
 
import (
    "context"
    "fmt"
    "log"
    "time"
    
    "github.com/go-redis/redis/v8"  // v8 表示支持 context
)
 
func main() {
    // ==================== 1. 初始化连接 ====================
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",  // Redis 地址
        Password: "",                 // 密码，没有就留空
        DB:       0,                  // 使用哪个数据库（0-15）
        PoolSize: 10,                 // 连接池大小（默认10）
        // 其他常用配置：
        // MinIdleConns: 5,           // 最小空闲连接数
        // MaxRetries: 3,             // 重试次数
        // DialTimeout: 5*time.Second, // 连接超时
    })
    
    // context 用于控制请求超时和取消
    ctx := context.Background()
    
    // 测试连接是否成功
    err := rdb.Ping(ctx).Err()
    if err != nil {
        log.Fatal("Redis 连接失败:", err)
    }
    fmt.Println("Redis 连接成功！")
    
    // ==================== 2. String 操作 ====================
    
    // Set：设置值，0 表示永不过期
    err = rdb.Set(ctx, "name", "张三", 0).Err()
    if err != nil {
        log.Fatal(err)
    }
    
    // Set：带过期时间（5分钟后自动删除）
    err = rdb.Set(ctx, "token", "abc123", 5*time.Minute).Err()
    
    // Get：获取值
    val, err := rdb.Get(ctx, "name").Result()
    if err == redis.Nil {
        fmt.Println("key 不存在")  // 注意：key不存在返回 redis.Nil，不是 error
    } else if err != nil {
        log.Fatal(err)
    } else {
        fmt.Println("name:", val)  // 输出：name: 张三
    }
    
    // Incr：计数器，原子操作
    rdb.Set(ctx, "views", 0, 0)
    rdb.Incr(ctx, "views")      // views = 1
    rdb.IncrBy(ctx, "views", 5) // views = 6
    
    // ==================== 3. Hash 操作 ====================
    
    // HSet：设置多个字段
    err = rdb.HSet(ctx, "user:1001", map[string]interface{}{
        "name":  "李四",
        "age":   25,
        "email": "lisi@example.com",
    }).Err()
    
    // HGet：获取单个字段
    name, _ := rdb.HGet(ctx, "user:1001", "name").Result()
    fmt.Println("用户名:", name)  // 输出：用户名: 李四
    
    // HGetAll：获取所有字段
    user, _ := rdb.HGetAll(ctx, "user:1001").Result()
    fmt.Println("用户信息:", user)  // 输出：map[name:李四 age:25 email:...]
    
    // HIncrBy：给某个字段加数字（比如年龄+1）
    rdb.HIncrBy(ctx, "user:1001", "age", 1)
    
    // ==================== 4. List 操作 ====================
    
    // LPush：从左边插入（最新的在最前面）
    rdb.LPush(ctx, "news:latest", "新闻1", "新闻2", "新闻3")
    // 列表顺序：新闻3, 新闻2, 新闻1
    
    // RPush：从右边插入
    rdb.RPush(ctx, "queue:tasks", "任务1", "任务2")
    
    // LRange：获取范围，0 到 -1 表示全部
    items, _ := rdb.LRange(ctx, "news:latest", 0, 9).Result()  // 获取前10条
    fmt.Println("最新新闻:", items)
    
    // LPop/RPop：取出并删除
    task, _ := rdb.RPop(ctx, "queue:tasks").Result()  // 从右边取出
    fmt.Println("取出任务:", task)
    
    // LTrim：只保留指定范围（常用于限制列表长度）
    rdb.LTrim(ctx, "news:latest", 0, 99)  // 只保留最新100条
    
    // ==================== 5. Set 操作 ====================
    
    // SAdd：添加成员（自动去重）
    rdb.SAdd(ctx, "user:1001:tags", "Go", "Redis", "后端", "Go")  // Go 只会存一次
    
    // SMembers：获取所有成员
    tags, _ := rdb.SMembers(ctx, "user:1001:tags").Result()
    fmt.Println("用户标签:", tags)  // 输出：[Go Redis 后端]
    
    // SIsMember：判断是否存在
    exists, _ := rdb.SIsMember(ctx, "user:1001:tags", "Go").Result()
    fmt.Println("是否有Go标签:", exists)  // true
    
    // SInter：求交集（共同好友）
    rdb.SAdd(ctx, "user:1001:friends", "2001", "2002", "2003")
    rdb.SAdd(ctx, "user:1002:friends", "2001", "2004")
    common, _ := rdb.SInter(ctx, "user:1001:friends", "user:1002:friends").Result()
    fmt.Println("共同好友:", common)  // [2001]
    
    // ==================== 6. ZSet 有序集合 ====================
    
    // ZAdd：添加带分数的成员
    rdb.ZAdd(ctx, "rank:points", &redis.Z{Score: 100, Member: "user:1001"})
    rdb.ZAdd(ctx, "rank:points", &redis.Z{Score: 200, Member: "user:1002"})
    rdb.ZAdd(ctx, "rank:points", &redis.Z{Score: 150, Member: "user:1003"})
    
    // ZRevRange：按分数从高到低获取（排行榜）
    top3, _ := rdb.ZRevRangeWithScores(ctx, "rank:points", 0, 2).Result()
    fmt.Println("积分前3名:", top3)
    
    // ZIncrBy：增加分数
    rdb.ZIncrBy(ctx, "rank:points", 50, "user:1001")  // 1001 的分数 +50
    
    // ZRank：获取排名（从0开始）
    rank, _ := rdb.ZRevRank(ctx, "rank:points", "user:1001").Result()
    fmt.Println("user:1001 排名:", rank+1)
    
    // ==================== 7. Pipeline 批量操作 ====================
    // Pipeline 可以一次发送多个命令，减少网络往返
    
    pipe := rdb.Pipeline()
    
    // 这些命令不会立即执行，而是放入队列
    incr := pipe.Incr(ctx, "counter")
    pipe.Expire(ctx, "counter", time.Hour)
    pipe.Set(ctx, "foo", "bar", 0)
    
    // Exec 一次性发送所有命令
    _, err = pipe.Exec(ctx)
    if err != nil {
        log.Fatal(err)
    }
    
    // 获取结果
    fmt.Println("Counter:", incr.Val())
    
    // ==================== 8. 发布订阅 ====================
    
    // 订阅者（通常在另一个 goroutine 或进程中）
    go func() {
        sub := rdb.Subscribe(ctx, "news:channel")
        defer sub.Close()
        
        for msg := range sub.Channel() {
            fmt.Println("收到消息:", msg.Payload)
        }
    }()
    
    time.Sleep(100 * time.Millisecond)  // 等待订阅者准备好
    
    // 发布者
    rdb.Publish(ctx, "news:channel", "今日头条：Go语言很棒！")
    
    // ==================== 9. 设置过期时间 ====================
    
    rdb.Set(ctx, "session:abc", "user:1001", 0)
    rdb.Expire(ctx, "session:abc", 30*time.Minute)  // 30分钟后过期
    
    // 查看剩余时间
    ttl, _ := rdb.TTL(ctx, "session:abc").Result()
    fmt.Println("剩余时间:", ttl)
    
    // ==================== 10. 删除 key ====================
    
    rdb.Del(ctx, "name", "token")  // 删除多个 key
    rdb.FlushDB(ctx)               // 清空当前数据库（危险！）
}

常见错误与最佳实践：

错误	正确做法
忘记设置过期时间	除非永久数据，否则都设 TTL
不处理 redis.Nil	Get 不存在的 key 返回 redis.Nil，要特殊处理
每次操作都新建连接	用连接池，程序启动时初始化一次
不用 Pipeline	批量操作用 Pipeline 减少网络往返

常见错误分析：

• 错误1：不理解Redis的数据类型
• 错误2：忘记设置过期时间导致内存泄漏
• 错误3：不理解Pipeline的使用

Q3: 如何在Go中使用消息队列？

消息队列核心概念：

生产者（Producer）→ 消息队列（Queue）→ 消费者（Consumer）

作用：
1. 异步处理：用户下单后立即返回，后台慢慢处理
2. 解耦：订单服务不需要知道短信服务的存在
3. 削峰：双11大量请求先存队列，慢慢消费

RabbitMQ vs Kafka 深度对比：

特性	RabbitMQ (传统 MQ)	Kafka (分布式流处理)
架构本质	基于队列 (Queue-based)。消息存放在内存或磁盘队列中。	基于日志 (Log-based)。消息持久化为顺序日志文件。
消息消费	推模式 (Push)。Broker 主动推给消费者。	拉模式 (Pull)。消费者根据自己速度去拉。
消费后处理	默认消费即删除 (ACK 后删除)。	持久化保留。消息根据配置保留时间，可回溯/重复消费。
吞吐量	万级 QPS。	百万级 QPS (利用顺序读写和零拷贝技术)。
负载均衡	靠镜像队列实现高可用。	分区 (Partition) 机制，天然支持并行和扩展。
适用场景	订单处理、通知、复杂的路由逻辑、任务分发。	日志收集、大数据分析、流式处理、用户行为跟踪。

面试核心追问：为什么 Kafka 这么快？

1. 顺序写磁盘 (Sequential I/O)：磁盘顺序写的速度比随机写快几千倍。
2. 零拷贝 (Zero-Copy)：利用 sendfile 系统调用，数据直接从内核缓冲区发送到网卡，不经过用户空间，减少拷贝次数。
3. 批量发送 (Batching)：将多条消息打包发送，减少网络 IO 次数。
4. 页缓存 (Page Cache)：充分利用操作系统的内存做缓存。

面试核心追问：RabbitMQ 怎么保证消息不丢？

1. 生产者确认：Confirm 机制，确保消息到达 Broker。
2. 持久化：Exchange、Queue、Message 都设置为 Durable/Persistent。
3. 消费者确认：手动 ACK，确保消息处理完再删除。

---

RabbitMQ 代码实现（带详细注释）：

package main
 
import (
    "context"
    "log"
    "time"
    
    "github.com/rabbitmq/amqp091-go"  // RabbitMQ 官方 Go 客户端
)
 
// ==================== 生产者代码 ====================
func producer() {
    // 1. 建立连接
    // amqp://用户名:密码@地址:端口/虚拟主机
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal("连接失败:", err)
    }
    defer conn.Close()  // ⚠️ 重要：用完必须关闭！
    
    // 2. 创建 Channel（通道）
    // 一个连接可以有多个 Channel，Channel 才是真正收发消息的
    ch, err := conn.Channel()
    if err != nil {
        log.Fatal("创建通道失败:", err)
    }
    defer ch.Close()
    
    // 3. 声明队列
    // 如果队列不存在会创建，存在则跳过
    q, err := ch.QueueDeclare(
        "order_queue",  // 队列名称
        true,           // durable：是否持久化（重启后还在）
        false,          // autoDelete：没有消费者时是否自动删除
        false,          // exclusive：是否排他（只有创建者能用）
        false,          // noWait：是否等待服务器确认
        nil,            // args：额外参数
    )
    if err != nil {
        log.Fatal("声明队列失败:", err)
    }
    
    // 4. 发送消息
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    
    // 模拟订单消息
    orderJSON := `{"order_id": "12345", "user_id": "1001", "amount": 99.99}`
    
    err = ch.PublishWithContext(ctx,
        "",             // exchange：交换机（空字符串=默认交换机）
        q.Name,         // routing key：路由键（这里是队列名）
        false,          // mandatory：是否强制投递
        false,          // immediate：是否立即投递
        amqp091.Publishing{
            ContentType:  "application/json",
            Body:         []byte(orderJSON),
            DeliveryMode: amqp091.Persistent,  // 消息持久化
        },
    )
    if err != nil {
        log.Fatal("发送消息失败:", err)
    }
    
    log.Println("消息发送成功:", orderJSON)
}
 
// ==================== 消费者代码 ====================
func consumer() {
    // 1. 建立连接（和生产者一样）
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatal("连接失败:", err)
    }
    defer conn.Close()
    
    ch, err := conn.Channel()
    if err != nil {
        log.Fatal("创建通道失败:", err)
    }
    defer ch.Close()
    
    // 2. 声明队列（确保队列存在）
    q, err := ch.QueueDeclare(
        "order_queue",
        true, false, false, false, nil,
    )
    if err != nil {
        log.Fatal(err)
    }
    
    // 3. 设置 QoS（服务质量）
    // prefetchCount：一次最多取多少条消息（防止某个消费者堆积太多）
    err = ch.Qos(
        1,     // prefetchCount：一次只取1条
        0,     // prefetchSize：不限制大小
        false, // global：是否应用于整个连接
    )
    if err != nil {
        log.Fatal(err)
    }
    
    // 4. 开始消费
    msgs, err := ch.Consume(
        q.Name,  // 队列名
        "",      // consumer：消费者标签（空=自动生成）
        false,   // autoAck：是否自动确认 ⚠️ 重要！false=手动确认
        false,   // exclusive：是否排他
        false,   // noLocal：不消费同一连接发的消息
        false,   // noWait：是否等待服务器确认
        nil,     // args：额外参数
    )
    if err != nil {
        log.Fatal(err)
    }
    
    log.Println("等待消息...")
    
    // 5. 处理消息
    for msg := range msgs {
        log.Printf("收到消息: %s", msg.Body)
        
        // 处理业务逻辑...
        processOrder(msg.Body)
        
        // ⚠️ 关键：手动确认消息
        // 告诉 RabbitMQ "我处理完了，可以删除这条消息了"
        msg.Ack(false)  // false=只确认当前消息
        
        // 如果处理失败，可以拒绝：
        // msg.Nack(false, true)  // true=重新入队
        // msg.Reject(true)       // 拒绝并重新入队
    }
}
 
func processOrder(body []byte) {
    // 模拟处理订单
    time.Sleep(1 * time.Second)
    log.Println("订单处理完成")
}
 
func main() {
    // 通常生产者和消费者是两个独立的程序
    go producer()
    consumer()
}

---

消息确认机制（重要！）：

模式	说明	风险
自动确认	消息发出去就删除	消费者崩溃会丢消息
手动确认	处理完再 Ack	更安全，推荐使用

// 自动确认（危险！）
ch.Consume(queue, "", true, ...)  // autoAck=true
 
// 手动确认（推荐）
ch.Consume(queue, "", false, ...)  // autoAck=false
// 处理完后：
msg.Ack(false)   // 成功，删除消息
msg.Nack(false, true)  // 失败，重新入队

---

常见应用场景：

场景	做法
异步发短信	用户下单 → 发消息 → 短信服务消费
订单超时取消	下单 → 发延迟消息（30分钟）→ 消费者检查是否支付
日志收集	各服务 → Kafka → ELK
秒杀削峰	请求 → 队列 → 按能力消费

---

常见错误与最佳实践：

错误	正确做法
忘记关闭连接	用 defer conn.Close()
自动确认丢消息	用手动确认 autoAck=false
消息没持久化	队列 durable=true + 消息 Persistent
单连接多 Channel	通常够用，别开太多连接
消费者处理太慢	设置 QoS，多开几个消费者