Client 实现分析

time：2026_1_21

核心文件：

• apps/client_main.cpp
• src/app/InputPrediction.cpp
• src/app/ClientRender.cpp

客户端的职责可以概括为：

1
2
3

先预测，让本地操作立刻生效；
再接受服务端权威状态；
最后从权威快照回滚重放，追上当前 tick。

1. ClientCtx 保存了什么

ClientCtx 是客户端运行态上下文。

它包含几类状态：

这说明客户端不是只保存“当前位置”，而是保存：

• 输入历史
• 状态历史
• 网络进度
• 权威 tick
• 本地预测 tick

这些状态共同支撑 rollback。

2. 未开局阶段：hello / Start

客户端启动后如果还没收到 Start，会定期发送空 InputPacket：

1

SendHello()

服务端收到后按 UDP 地址分配 slot。

客户端收到 StartPacket 后执行：

1

ApplyStart()

主要动作：

1. 记录 localPlayerId。
2. 设置 startTick。
3. 把客户端本地 tick 对齐到 startTick。
4. 初始化世界快照并写入 stateHist。

注意：

Start 不是权威状态同步包，它只是告诉客户端“你是谁，以及从哪个 tick 开始”。

3. 每帧预测流程

客户端 OnTick 里使用 accumulator：

1
2
3
4
5
6

frame = now - prev
acc += min(frame, maxFrame)

while acc >= dt:
  推进一个逻辑 tick
  acc -= dt

每个逻辑 tick 做四件事：

1. 从 SDL 键盘状态采样本地输入。
2. 写入 localHist。
3. 用本地输入和远端预测输入调用 worldPred.Step。
4. 保存预测快照到 stateHist，然后发送冗余 Input 包。

核心链路：

1
2
3
4
5
6
7

PollInput
  -> localHist.Put(localCmd)
  -> GetRemoteCmdForTick
  -> BuildCmdVec
  -> worldPred.Step
  -> stateHist.Put
  -> EncodeInput + SendTo

4. BuildCmdVec 的意义

World::Step 需要的是“所有玩家同一个 tick 的输入数组”。

但是客户端本地只真正知道自己的输入，对手输入只能预测。

所以 BuildCmdVec 的任务是：

1
2
3

把本地真实输入放到自己的 player slot；
把远端预测输入放到其他 player slot；
形成 vector<InputCmd> 交给 World::Step。

这也是服务端和客户端能复用同一个 World::Step 的原因。

5. 远端输入预测

远端玩家的真实输入不会直接发给客户端。

客户端根据服务端 State 里的远端状态估计远端输入：

• 如果远端在 Hitstun，预测不动。
• 如果速度明显向右，预测 moveX = 1。
• 如果速度明显向左，预测 moveX = -1。
• 如果速度接近 0，预测停止。
• 缺少新预测时，短时间 HoldLast。

相关函数：

• PredictMoveXFromState
• PredictMoveYFromState
• GetRemoteCmdForTick

注意：

远端输入预测只是为了让画面连续，不是权威逻辑。最终仍然以服务端 State 为准。

6. 收包处理：ACK 和 State

客户端收包入口：

1

OnUdp()

它会依次尝试解析：

1. StartPacket
2. AckPacket
3. StatePacket

ACK

ACK 更新：

• lastServerTick
• lastServerHash
• serverLastInputTick

它主要表示服务端处理进度，不携带完整权威状态。

客户端可以用：

1

lead = localTick - lastServerTick

观察自己本地预测领先服务端多少 tick。

State

State 才是真正的权威状态包。

客户端收到 State 后：

1. 还原成 WorldSnapshot auth。
2. 校验 Hasher::Hash(auth) 是否等于 stateHash。
3. 更新远端输入预测。
4. 判断本地玩家是否明显偏离。
5. 写入权威快照。
6. RestoreAndReplay。

7. RestoreAndReplay

回滚重放流程：

1
2
3
4
5
6
7
8

worldPred.Restore(authoritativeSnapshot)

for t = authoritativeTick + 1; t < localNextTick; ++t:
  localCmd = localHist[t] or default
  remoteCmd = remoteHist[t] or hold/default
  cmds = BuildCmdVec(...)
  worldPred.Step(cmds, dt)
  stateHist.Put(worldPred.Snapshot())

重点：

• 回滚不是把玩家瞬移到服务端位置。
• 回滚是恢复到历史权威点，然后重新模拟未来输入。
• 这样玩家本地输入仍然能保持即时反馈。

8. 为什么只统计本地玩家 rollback

代码里判断 needRollback 时只比较本地玩家：

• 位置误差
• HP
• action
• onGround

原因是远端玩家本来就是预测出来的。如果远端误差也算 rollback，会导致统计频繁增长，调试信息失真。

但是注意：

即使没有计入 rollbackCount，客户端仍然会 rebase + replay。

这是为了让远端玩家和子弹尽快采用权威状态。

9. 客户端注意事项

• localHist 只保存本地玩家输入，不保存整局所有玩家真实输入。
• remoteHist 保存的是远端预测输入，不是权威输入。
• stateHist 既会存预测快照，也会被权威快照覆盖。
• lastAuthoritativeTick 用来丢弃旧 State，防止延迟包覆盖新状态。
• 新增会影响模拟的玩家字段时，要同步改 WorldSnapshot、StatePacket、Hasher 和测试。
• 不要把远端预测当成权威结果。
• 不要用真实帧耗时直接推进 World::Step。
• rollbackCount 是调试指标，不等于所有 rebase 次数。

模拟层笔记

核心文件：

• include/lab/sim/StateSnapshot.h
• include/lab/sim/InputCmd.h
• include/lab/sim/InputBuffer.h
• include/lab/sim/StateHistory.h
• src/sim/World.cpp
• src/sim/Hasher.cpp

模拟层是整个项目最核心的部分。网络同步、客户端预测、服务端权威、回滚重放和压力测试，最后都依赖同一个模拟入口：

1

World::Step(cmds, dt)

1. 模拟层的目标

模拟层要做到：

• 同样初始状态 + 同样输入 + 同样 tick = 同样结果。
• 不依赖 SDL、socket、真实网络。
• 能生成完整快照。
• 能从快照恢复。
• 能被 hash 检查一致性。
• 能被测试直接调用。

这就是确定性模拟的基础。

2. InputCmd：最小输入命令

InputCmd 表示某个玩家在某个 tick 的输入。

字段：

• tick
• buttons
• moveX
• moveY

设计重点：

• 输入必须带 tick。
• 输入只描述玩家意图，不描述最终结果。
• 输入要足够小，适合 UDP 高频发送。
• 输入可以重复发送，服务端按 tick 写入缓冲。

3. WorldSnapshot：状态快照

WorldSnapshot 包含：

• tick
• 玩家数组
• 子弹数组
• mazeSeed
• 迷宫宽高
• 迷宫网格

它用于：

• 服务端下发权威状态。
• 客户端回滚恢复。
• 压测保存权威历史。
• hash 对账。

判断一个字段是否应该进入 Snapshot，可以问：

1

这个字段是否会影响未来模拟结果？

如果答案是会，就必须进入 Snapshot。

例如 aimX/aimY 很重要。玩家停下后开火时，开火方向依赖上一次瞄准方向。如果回滚时没有恢复这个字段，子弹方向就可能和服务端不同。

4. World::Step 推进流程

World::Step 做的事情：

1. 检查输入数量是否等于玩家数量。
2. 检查所有输入 tick 是否一致。
3. 确保迷宫存在。
4. 设置当前快照 tick。
5. 对每个玩家处理 hitstun、移动、摩擦、墙体碰撞、朝向、射击冷却。
6. 如果有攻击输入且冷却结束，生成子弹。
7. 处理玩家之间 pushbox 分离。
8. 推进子弹，处理撞墙和命中。
9. 同步子弹和迷宫到快照。

一句话：

1

World::Step 是唯一的模拟推进入口。

5. 玩家移动

玩家移动不是瞬间设置速度，而是向目标速度插值：

1
2

targetVx = moveX * speed
v += (targetVx - v) * alpha

其中：

• speed = 4.5
• friction = 8.0
• dt = 1 / 60

这样移动会更平滑。

墙体碰撞采用分轴处理：

1
2
3

先尝试 x 方向移动；
再尝试 y 方向移动；
某方向撞墙就清掉该方向速度。

6. 子弹和命中

攻击输入会生成子弹：

• 子弹 owner 是玩家编号，1-based。
• 子弹有位置、速度、life。
• 子弹从玩家前方生成。
• 子弹撞墙或命中后消失。

命中玩家后：

• 扣 HP。
• 设置 Action::Hitstun。
• 设置 stateTimer。
• 给目标一点击退速度。

注意：

子弹状态必须进入 Snapshot、网络包和 hash。否则子弹命中会造成服务端和客户端状态分叉。

7. 迷宫生成

迷宫由 mazeSeed 驱动，用 DFS carve 方式生成。

服务端 State 里会下发 mazeSeed，客户端用 seed 生成同样迷宫。

但注意：

当前实现使用 std::shuffle 和 std::mt19937。同机测试通常没问题，但跨编译器、跨标准库时最好不要只依赖 seed。更稳的做法是：

• 同步完整迷宫网格。
• 或实现项目自己的确定性随机和 shuffle。

项目的 Snapshot 和 hash 已经包含迷宫网格，这是正确方向。

8. InputBuffer 和 StateHistory

两者都是按 tick 索引的环形缓冲。

核心公式：

1

index = tick % capacity

读取时必须校验真实 tick：

1

slot.tick == requestedTick

否则环形缓冲覆盖后，可能把旧帧误认为新帧。

用途：

9. Hasher

Hasher 用于状态一致性检查。

它覆盖：

• tick
• 玩家状态
• 子弹状态
• 迷宫 seed
• 迷宫尺寸
• 迷宫网格

float 字段不会直接按内存 hash，而是做毫米量化：

1

round(value * 1000)

这样和网络包里的毫米整数保持一致，避免微小浮点差异造成假 mismatch。

10. 模拟层注意事项

• 新增会影响未来的字段时，要同步改 Snapshot、StatePacket、Hasher、Restore、测试。
• World::Step 必须使用固定 dt，不要传真实帧耗时。
• 不要在模拟层访问 SDL、socket、event loop。
• 不要直接 hash float 内存。
• 不要让随机数在服务端和客户端各自自由推进，随机结果必须可恢复、可同步。
• 回滚恢复时不只要恢复玩家，也要恢复子弹、迷宫、冷却、朝向等隐性状态。
• 玩家 pushbox 后可能需要再次考虑墙体约束，这是后续可改进点。

渲染层笔记

核心文件：

• include/lab/app/ClientRender.h
• src/app/ClientRender.cpp
• apps/client_main.cpp

渲染层不是项目的同步核心，但它负责把预测世界展示出来，并把回滚、hash mismatch 等调试信息显示给开发者。

1. RenderCtx

RenderCtx 保存 SDL 渲染资源：

初始化：

1

InitRenderer()

释放：

1

ShutdownRenderer()

注意：

SDL 资源要成对创建和释放。后续如果继续完善，可以把 window、renderer、font 封装成 RAII 对象。

2. RenderFrame 输入

渲染函数只接收：

1

RenderFrame(renderCtx, worldSnapshot, rollbackCount, hashMismatchCount)

这说明渲染层不直接修改世界。

它只读 WorldSnapshot，然后画：

• 迷宫墙体
• 玩家方块
• 子弹
• HUD 文本

这是一个很好的边界：

1
2
3

模拟层产出状态；
渲染层消费状态；
渲染层不反向修改模拟。

3. 世界坐标到屏幕坐标

项目用简单的顶视角坐标转换：

1
2

screenX = windowWidth / 2 + worldX * scale
screenY = windowHeight / 2 - worldY * scale

Y 方向取反，是因为屏幕坐标通常向下为正，而游戏世界里向上为正。

scale 根据迷宫尺寸和窗口大小计算：

1

scale = min(width / mazeWidth, height / mazeHeight) * 0.8

这样不同迷宫尺寸都能大致居中显示。

4. HUD 调试信息

HUD 显示：

• 当前 tick
• rollback 次数
• hash mismatch 次数
• maze seed
• 每个玩家 HP
• 每个玩家 action
• stateTimer

这些信息是联机同步调试里很关键的观察点。

尤其是：

• rollback 次数持续上升，说明预测和权威经常偏离。
• hash mismatch 出现，说明状态还原或确定性有问题。
• tick/HP/action 可以帮助定位是哪一帧开始分叉。

5. 渲染和模拟的关系

客户端每个 tick 会先推进 worldPred，然后 RenderFrame 画当前预测快照。

收到服务端 State 后，客户端会：

1
2
3

Restore(auth)
Replay(...)
RenderFrame(...)

所以屏幕上看到的是“经过本地预测和权威校正后的世界”。

注意：

渲染只展示预测结果。真正权威状态仍然在服务端。

6. 渲染层注意事项

• 渲染层不要保存会影响模拟的状态。
• 渲染层不要修改 WorldSnapshot。
• HUD 是调试工具，不应该参与同步逻辑。
• 如果增加插值/平滑，最好只做视觉层修正，不要改权威模拟状态。
• 字体路径当前写死为 /System/Library/Fonts/Menlo.ttc，跨平台需要改成配置项。
• 如果窗口卡顿，不应该让模拟用真实大 dt 推进，固定帧 accumulator 已经负责处理。
• 后续可以补充摄像机、缩放、玩家标识、命中反馈，但仍应保持渲染和模拟解耦。

生产者-消费者模式与阻塞队列

时间：2026/04/09

关键词：mutex、condition_variable、阻塞队列、bounded queue、shutdown、spurious wakeup
核心目标：写出一个正确、可复用的生产者-消费者队列，而不是“能跑但容易死锁或卡住”的版本。

1. 这个模式在解决什么问题

生产者-消费者模式适用于：

• 生产方产生任务或消息
• 消费方异步处理
• 双方速度不一致

典型场景：

• 日志队列
• 任务队列
• 网络消息分发
• 线程池任务提交

它的核心不只是“一个队列”，而是三件事：

• 互斥
• 条件通知
• 生命周期关闭

2. 最小阻塞队列骨架

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

#include <condition_variable>
#include <mutex>
#include <queue>

template <class T>
class BlockingQueue {
public:
    void push(T value) {
        {
            std::lock_guard<std::mutex> lk(mutex_);
            queue_.push(std::move(value));
        }
        cv_.notify_one();
    }

    T pop() {
        std::unique_lock<std::mutex> lk(mutex_);
        cv_.wait(lk, [&] { return !queue_.empty(); });
        T value = std::move(queue_.front());
        queue_.pop();
        return value;
    }

private:
    std::mutex mutex_;
    std::condition_variable cv_;
    std::queue<T> queue_;
};

3. 为什么一定要用谓词版 wait

错误直觉是：

1

cv.wait(lock);

然后醒来就认为一定有数据。
这是不安全的，因为存在：

• 虚假唤醒
• 多个线程竞争同一个条件

正确写法：

1

cv.wait(lock, [&] { return !queue_.empty(); });

也就是：

• 醒来后重新检查条件

4. bounded queue：为什么需要容量上限

如果生产速度远大于消费速度，无界队列会不断膨胀。
这时常需要有界队列：

• 队列满时，生产者阻塞或失败

1
2
3

while (queue_.size() >= capacity_) {
    not_full_.wait(lock);
}

这样可以建立：

• 背压
• 内存上限

5. 一个更完整的阻塞队列设计

更工程化的队列通常需要这些接口：

• push
• try_push
• pop
• try_pop
• close

close() 很关键，因为消费者可能永远在等：

1
2
3

if (closed_ && queue_.empty()) {
    return std::nullopt;
}

否则程序退出时很容易卡死在线程等待上。

6. 推荐的关闭语义

常见设计是：

• 关闭后不允许再 push
• 还能把队列里剩余任务消费完
• 队列空且关闭时，pop 返回“结束”

示意：

1
2
3
4
5
6

std::optional<T> pop() {
    std::unique_lock<std::mutex> lk(mutex_);
    cv_.wait(lk, [&] { return closed_ || !queue_.empty(); });
    if (queue_.empty()) return std::nullopt;
    ...
}

这比靠塞一个 "EXIT" 哨兵值更通用。

7. 什么时候用 notify_one，什么时候用 notify_all

经验上：

• 普通入队，通常 notify_one
• 全局状态变化，比如 close()，通常 notify_all

因为关闭时可能有多个线程都在等待，需要全部唤醒重新判断。

8. 一个更稳妥的示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

#include <condition_variable>
#include <mutex>
#include <optional>
#include <queue>

template <class T>
class BlockingQueue {
public:
    explicit BlockingQueue(std::size_t capacity) : capacity_(capacity) {}

    bool push(T value) {
        std::unique_lock<std::mutex> lk(mutex_);
        not_full_.wait(lk, [&] { return closed_ || queue_.size() < capacity_; });
        if (closed_) return false;
        queue_.push(std::move(value));
        lk.unlock();
        not_empty_.notify_one();
        return true;
    }

    std::optional<T> pop() {
        std::unique_lock<std::mutex> lk(mutex_);
        not_empty_.wait(lk, [&] { return closed_ || !queue_.empty(); });
        if (queue_.empty()) return std::nullopt;
        T value = std::move(queue_.front());
        queue_.pop();
        lk.unlock();
        not_full_.notify_one();
        return value;
    }

    void close() {
        std::lock_guard<std::mutex> lk(mutex_);
        closed_ = true;
        not_empty_.notify_all();
        not_full_.notify_all();
    }

private:
    std::size_t capacity_;
    std::queue<T> queue_;
    bool closed_ = false;
    std::mutex mutex_;
    std::condition_variable not_empty_;
    std::condition_variable not_full_;
};

9. 阻塞队列如何安全扩容

这里的“扩容”通常指：

有界队列原来最多只能放 capacity_ 个元素，现在允许它放更多元素。

扩容本身不需要搬迁队列数据，因为 std::queue 会自己管理内部存储。我们真正要保护的是：

• capacity_ 的读写
• 正在等待 not_full_ 的生产者
• 队列的关闭状态

9.1 最小扩容接口

可以给上面的 BlockingQueue 加一个 reserve_capacity()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

bool reserve_capacity(std::size_t new_capacity) {
    std::unique_lock<std::mutex> lk(mutex_);

    if (closed_) {
        return false;
    }

    if (new_capacity <= capacity_) {
        return true;
    }

    capacity_ = new_capacity;
    lk.unlock();
    not_full_.notify_all();
    return true;
}

这里的关键点是：

• 修改 capacity_ 必须持有同一把 mutex_
• 扩容后要通知等待中的生产者
• 队列已经 close() 后，不再允许扩容

为什么用 notify_all()？

因为扩容可能一次释放多个可写入位置，多个生产者都可能从“队列满”变成“可以写入”。

9.2 按增量扩容

有时也会写成“增加多少容量”：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

bool grow_capacity(std::size_t extra) {
    std::unique_lock<std::mutex> lk(mutex_);

    if (closed_) {
        return false;
    }

    if (extra == 0) {
        return true;
    }

    capacity_ += extra;
    lk.unlock();
    not_full_.notify_all();
    return true;
}

工程里还要考虑：

• 最大容量上限
• capacity_ + extra 是否溢出
• 是否允许扩容太频繁
• 扩容后内存压力是否可接受

否则有界队列可能又退化成“无界队列”。

9.3 自动扩容的思路

如果希望 push() 遇到满队列时自动扩容，可以把逻辑写在同一把锁里：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

bool push(T value) {
    std::unique_lock<std::mutex> lk(mutex_);
    bool expanded = false;

    if (!closed_ &&
        queue_.size() >= capacity_ &&
        capacity_ < max_capacity_) {
        capacity_ = std::min(capacity_ * 2, max_capacity_);
        expanded = true;
    }

    not_full_.wait(lk, [&] {
        return closed_ || queue_.size() < capacity_;
    });

    if (closed_) return false;

    queue_.push(std::move(value));
    lk.unlock();
    not_empty_.notify_one();
    if (expanded) {
        not_full_.notify_all();
    }
    return true;
}

对应成员变量可以是：

1
2

std::size_t capacity_;
std::size_t max_capacity_;

如果用 std::min，记得包含：

1

#include <algorithm>

这个版本的意思是：

• 队列满时，先尝试扩到更大的容量
• 如果已经到最大容量，就继续阻塞等待
• 所有判断和容量更新都在锁内完成
• 扩容后唤醒其他可能还在等待的生产者

注意不要在持锁的 push() 里再调用一个内部也会加锁的 grow_capacity()，否则很容易把自己锁死。

9.4 缩容要更谨慎

扩容比较简单，因为它只会让等待生产者更容易继续。

缩容更麻烦，因为可能出现：

• 当前元素数量已经超过新容量
• 生产者和消费者对“满”的判断突然变化
• 缩容策略和关闭流程互相影响

一个保守策略是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

bool shrink_capacity(std::size_t new_capacity) {
    std::lock_guard<std::mutex> lk(mutex_);

    if (closed_) {
        return false;
    }

    if (new_capacity < queue_.size()) {
        return false;
    }

    capacity_ = new_capacity;
    return true;
}

也就是说：

不把容量缩到当前队列大小以下。

这样可以避免队列瞬间进入一种“已经超载但无法解释”的状态。

10. 常见坑

10.1 if 代替 while/谓词

这是最常见的条件变量错误。

10.2 持锁太久

如果拿着锁做重计算或 I/O，会严重拖慢并发吞吐。

10.3 没有关闭语义

线程可能永远阻塞退出不了。

10.4 用哨兵值替代通用关闭协议

对简单 demo 可以，但扩展性差。

11. 一页总结

生产者-消费者模式的核心不是“有个队列”，而是：

1. 用互斥保护共享队列
2. 用条件变量等待状态变化
3. 用关闭协议管理线程退出
4. 必要时用容量上限建立背压
5. 动态扩容时，要在同一把锁内修改容量并唤醒等待生产者

如果只记一句：

条件变量永远和“共享状态 + 谓词检查”一起使用，不能只靠通知本身。

线程同步消息队列与线程池

时间：2026/04/09

关键词：任务队列、worker thread、future、停止协议、背压、线程池
核心目标：理解线程池为什么几乎总是“队列 + 工作线程 + 生命周期管理”的组合。

1. 为什么线程池比“每个任务一个线程”更常见

直接为每个任务创建线程的问题在于：

• 创建销毁开销高
• 线程数不可控
• 容易把系统调度器压爆

线程池的思路是：

• 预先创建固定数量 worker
• 任务进入共享队列
• worker 从队列取任务执行

2. 线程池最小结构

一个线程池通常包含：

• 任务队列
• 多个工作线程
• 停止标志
• 提交接口

示意：

1

producer -> task queue -> workers

3. 推荐的任务表示

最常见的是：

1

std::function<void()>

这样线程池不关心任务具体类型，只负责执行。

如果要返回值，可以把真实任务包装进：

• std::packaged_task
• std::promise
• std::future

4. 一个最小线程池骨架

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

class ThreadPool {
public:
    explicit ThreadPool(std::size_t n) {
        for (std::size_t i = 0; i < n; ++i) {
            workers_.emplace_back([this] { worker_loop(); });
        }
    }

    ~ThreadPool() {
        {
            std::lock_guard<std::mutex> lk(mutex_);
            stop_ = true;
        }
        cv_.notify_all();
        for (auto& t : workers_) {
            if (t.joinable()) t.join();
        }
    }

    void submit(std::function<void()> task) {
        {
            std::lock_guard<std::mutex> lk(mutex_);
            tasks_.push(std::move(task));
        }
        cv_.notify_one();
    }

private:
    void worker_loop() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lk(mutex_);
                cv_.wait(lk, [&] { return stop_ || !tasks_.empty(); });
                if (stop_ && tasks_.empty()) return;
                task = std::move(tasks_.front());
                tasks_.pop();
            }
            task();
        }
    }

    bool stop_ = false;
    std::mutex mutex_;
    std::condition_variable cv_;
    std::queue<std::function<void()>> tasks_;
    std::vector<std::thread> workers_;
};

5. 为什么停止协议很重要

如果没有明确的停止逻辑，线程池很容易在析构时：

• worker 永远等在 wait
• 主线程 join 不回来

正确退出条件通常是：

• stop_ == true
• 并且队列已空

6. 返回值怎么做

常见写法是：

• 把用户任务包装成 packaged_task
• 返回对应 future

这样提交方既能异步执行，也能之后 get() 结果。

线程池的接口常见长这样：

1
2

template <class F, class... Args>
auto enqueue(F&& f, Args&&... args) -> std::future<...>;

这也是完美转发的高频实战场景。

7. 有界任务队列与背压

如果任务生产速度远大于消费速度，线程池也可能把内存吃爆。
所以工程上经常要考虑：

• 队列容量上限
• 超限后阻塞
• 超限后丢弃
• 超限后降级

这其实就是背压策略。

8. 线程池不是越多线程越好

线程数通常取决于：

• CPU 核心数
• 任务是否 CPU 密集
• 任务是否经常阻塞 I/O

经验上：

• CPU 密集型：线程数通常接近核心数
• I/O 密集型：线程数可适当更大

9. 线程池如何安全扩容

线程池扩容的本质是：

在不破坏任务队列、不影响已有 worker、不和析构/停止流程打架的前提下，增加新的 worker 线程。

只“扩容”通常比“缩容”简单，因为扩容不需要强行打断已有线程，只需要让更多线程开始消费同一个任务队列。

9.1 安全扩容要守住的几个点

1. 扩容时要和停止状态互斥
2. 新 worker 必须复用同一套 worker_loop
3. 不要在任务执行期间持有队列锁
4. 不允许线程池已经停止后继续扩容
5. 如果有最大线程数，要在锁内检查和更新

最核心的判断是：

1
2
3

if (stop_) {
    throw std::runtime_error("thread pool already stopped");
}

否则可能出现这种危险情况：

1. 析构线程设置 stop_ = true
2. 另一个线程又新增 worker
3. 析构只 join 了旧线程或生命周期已经混乱

9.2 一个简单的扩容接口

可以给线程池加一个 add_workers()：

1
2
3
4
5
6
7
8
9
10
11

void add_workers(std::size_t count) {
    std::lock_guard<std::mutex> lk(mutex_);

    if (stop_) {
        throw std::runtime_error("thread pool already stopped");
    }

    for (std::size_t i = 0; i < count; ++i) {
        workers_.emplace_back([this] { worker_loop(); });
    }
}

这段代码的关键点是：

• stop_ 和 workers_ 的修改放在同一把锁保护下
• 新线程执行的还是原来的 worker_loop()
• 新 worker 会自动从同一个 tasks_ 队列里抢任务

不过这只是最小写法。正式项目里通常还会加：

• max_workers_
• 当前线程数统计
• 扩容失败处理
• 线程池生命周期约束

9.3 带最大线程数的版本

更工程化一点：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

void add_workers(std::size_t count) {
    std::lock_guard<std::mutex> lk(mutex_);

    if (stop_) {
        throw std::runtime_error("thread pool already stopped");
    }

    const std::size_t current = workers_.size();
    const std::size_t allowed = max_workers_ > current
        ? max_workers_ - current
        : 0;

    const std::size_t actual = std::min(count, allowed);

    for (std::size_t i = 0; i < actual; ++i) {
        workers_.emplace_back([this] { worker_loop(); });
    }
}

对应成员变量：

1

std::size_t max_workers_ = std::thread::hardware_concurrency() * 2;

这里的重点不是公式，而是：

扩容不能无限扩，否则线程池会退化成“每个任务都创建线程”的混乱状态。

9.4 扩容后需要 notify_all() 吗

通常不一定需要。

因为新 worker 创建后会进入 worker_loop()，它自己会检查队列：

1

cv_.wait(lk, [&] { return stop_ || !tasks_.empty(); });

如果队列里已经有任务，谓词为真，新 worker 不会一直睡着。

但如果你的实现不是带谓词的 wait，或者扩容逻辑还改变了其他调度状态，就要重新检查通知逻辑。

9.5 自动扩容的常见触发条件

如果做成动态线程池，常见策略是：

• 队列积压超过阈值
• 当前线程数小于最大线程数
• 最近一段时间任务消费速度跟不上提交速度
• 任务是 I/O 密集型，worker 经常阻塞

伪代码：

1
2
3
4

if (tasks_.size() > high_watermark &&
    workers_.size() < max_workers_) {
    add_workers(1);
}

注意这个判断必须在锁保护下完成，避免多个提交线程同时发现“需要扩容”，然后一起扩太多。

9.6 缩容比扩容更麻烦

扩容是“增加消费者”，一般比较安全。

缩容是“让某些 worker 退出”，要设计额外协议，例如：

• 空闲超时退出
• 投递特殊退出任务
• 设置目标线程数，让多余 worker 在空闲时自然退出

不要粗暴强杀线程。C++ 标准线程没有安全的强制 kill 机制，强行终止线程很容易破坏锁、资源和对象状态。

10. 消息队列 vs 线程池

这两个概念经常一起出现，但不完全一样。

• 消息队列：强调数据传递与同步
• 线程池：强调任务执行与线程复用

线程池内部几乎总会用到任务队列，但消息队列本身不一定等于线程池。

11. 常见坑

11.1 任务里抛异常没人管

如果没有 future 或显式捕获，异常可能直接导致线程终止。

11.2 析构时仍允许提交任务

这会让生命周期变得混乱。

11.3 持锁执行任务

这是严重错误。
正确做法是：

• 取出任务后释放锁
• 再执行任务

11.4 线程池里再无限提交内部任务

这可能制造级联膨胀和死锁风险。

12. 一页总结

线程池最关键的不是模板技巧，而是三个工程点：

1. 任务队列
2. worker 生命周期
3. 明确的停止与背压策略

如果只记一句：

线程池本质上是“用受控线程数去消费一个受控任务流”。

MySQL 学习笔记

1. MySQL 是什么

MySQL 是一个关系型数据库，用来存储：

• 用户信息
• 订单数据
• 文档元数据
• 任务记录
• 聊天记录
• 各种需要长期保存的结构化数据

先抓住它和 Redis 的分工差异：

MySQL 更适合存“长期、结构化、可查询的数据”，Redis 更适合存“高频、临时、快速访问的数据”。

2. 为什么项目里经常会有 MySQL

因为很多业务数据都需要：

• 长期保存
• 有明确字段结构
• 支持条件查询
• 支持关联查询
• 保证数据一致性

例如：

• 一个用户有哪些会话
• 一个文档属于哪个用户
• 一条任务记录当前是什么状态
• 某段时间创建了多少条消息

这些都很适合用 MySQL 来存。

3. 在 AI 项目里 MySQL 常见用法

在 AI 模型开发场景里，MySQL 常见用途有：

• 存用户表
• 存会话表
• 存消息表
• 存文档表
• 存任务记录表
• 存模型调用日志
• 存权限和配置

放回整套系统里看，各组件的分工大致是：

• FastAPI / Drogon：接请求
• Redis：做缓存、状态层、中间层
• Celery：处理后台任务
• MySQL：存长期结构化数据

4. MySQL 和 Redis 的区别

简单理解：

• 用户资料、文档记录、消息历史更适合 MySQL
• 缓存结果、限流计数、短期任务状态更适合 Redis

5. 安装和启动

如果本地已经装好了 MySQL，可以直接启动服务。

如果只是临时学习，Docker 会更省事：

1
2
3
4
5
6

docker run -d \
  --name mysql8 \
  -p 3306:3306 \
  -e MYSQL_ROOT_PASSWORD=123456 \
  -e MYSQL_DATABASE=demo \
  mysql:8

进入 MySQL：

1

mysql -h 127.0.0.1 -P 3306 -u root -p

查看数据库：

1

SHOW DATABASES;

选择数据库：

1

USE demo;

6. Python 连接 MySQL

Python 项目里很常见的是 pymysql。

安装：

1

pip install pymysql

最基础的连接方式：

1
2
3
4
5
6
7
8
9
10
11

import pymysql

conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="123456",
    database="demo",
    charset="utf8mb4",
    autocommit=False,
)

几个常见参数：

• host：数据库地址
• port：端口，MySQL 默认通常是 3306
• user / password：用户名密码
• database：要连接的库
• charset="utf8mb4"：推荐写上，避免中文和 emoji 编码问题
• autocommit=False：默认手动提交事务

7. 连接和游标 cursor

你原来笔记里提到“游标对象的具体作用是什么”，可以这样理解：

连接 conn 负责连数据库，游标 cursor 负责执行 SQL 和取结果。

最常见写法：

1

cursor = conn.cursor()

更推荐：

1
2
3

with conn.cursor() as cursor:
    cursor.execute("SELECT 1")
    result = cursor.fetchone()

cursor 主要负责：

• 执行 SQL
• 传递参数
• 获取查询结果
• 提供 lastrowid、rowcount 等信息

常见方法：

• cursor.execute()：执行一条 SQL
• cursor.executemany()：批量执行
• cursor.fetchone()：取一条
• cursor.fetchmany(n)：取前 n 条
• cursor.fetchall()：取全部

如果你希望查询结果按字典返回，而不是元组，可以这样写：

1
2
3
4
5
6
7
8
9
10

import pymysql.cursors

conn = pymysql.connect(
    host="127.0.0.1",
    user="root",
    password="123456",
    database="demo",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
)

这样查出来的数据会更接近：

1

{"id": 1, "name": "tom"}

8. 建表和常见字段类型

MySQL 最基础的 SQL 之一就是建表。

示例：

1
2
3
4
5
6
7

CREATE TABLE users (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    username VARCHAR(100) NOT NULL UNIQUE,
    email VARCHAR(200) DEFAULT NULL,
    age INT DEFAULT 0,
    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);

常见字段类型：

• INT：整数
• BIGINT：更大的整数
• VARCHAR(n)：变长字符串
• TEXT：长文本
• DATETIME：日期时间
• DATE：日期
• DECIMAL(10,2)：精确小数
• BOOLEAN：布尔，底层通常可看成 TINYINT

常见约束：

• PRIMARY KEY：主键
• AUTO_INCREMENT：自增
• NOT NULL：不能为空
• UNIQUE：唯一
• DEFAULT：默认值
• FOREIGN KEY：外键

9. 表结构相关 SQL

这部分在实际开发里很常用。

9.1 创建表

1
2
3
4

CREATE TABLE users (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL
);

9.2 如果表已存在就跳过

你前面提到的“表已存在怎么办”，对应的就是这个场景。

如果担心“已经有同名表，再创建会报错”，可以写：

1
2
3
4

CREATE TABLE IF NOT EXISTS users (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(100) NOT NULL
);

这样当表已存在时，会直接跳过，不会因为同名表报错。

9.3 修改表

1

ALTER TABLE users ADD COLUMN email VARCHAR(200);

9.4 删除表

1

DROP TABLE users;

如果担心表不存在报错：

1

DROP TABLE IF EXISTS users;

9.5 查看建表语句

1

SHOW CREATE TABLE users;

这个命令非常实用，因为它能直接看到当前表的真实结构、索引、字符集等信息。

10. 插入数据 INSERT

最基础的插入写法：

1
2

INSERT INTO users (username, email, age)
VALUES ('tom', 'tom@example.com', 18);

在 Python 里更推荐参数化写法：

1
2
3
4
5
6
7

with conn.cursor() as cursor:
    sql = """
    INSERT INTO users (username, email, age)
    VALUES (%s, %s, %s)
    """
    cursor.execute(sql, ("tom", "tom@example.com", 18))
    conn.commit()

插入后拿自增主键：

1

user_id = cursor.lastrowid

11. 查询数据 SELECT

11.1 查全部字段

1

SELECT * FROM users;

11.2 只查部分字段

1

SELECT id, username FROM users;

11.3 带条件查询

1
2
3

SELECT id, username
FROM users
WHERE age >= 18;

11.4 排序

1
2
3

SELECT id, username
FROM users
ORDER BY created_at DESC;

11.5 分页

1
2
3
4

SELECT id, username
FROM users
ORDER BY id DESC
LIMIT 10 OFFSET 0;

也常写成：

1

LIMIT 0, 10;

12. 更新数据 UPDATE

1
2
3

UPDATE users
SET email = 'new@example.com'
WHERE id = 1;

一定要注意：

UPDATE 不写 WHERE，会更新整张表。

Python 示例：

1
2
3
4
5

with conn.cursor() as cursor:
    sql = "UPDATE users SET email = %s WHERE id = %s"
    affected_rows = cursor.execute(sql, ("new@example.com", 1))
    conn.commit()
    print(affected_rows)

13. 删除数据 DELETE

1
2

DELETE FROM users
WHERE id = 1;

同样要注意：

DELETE 不写 WHERE，会删整张表的数据。

如果你是想删整张表但保留结构，也可以：

1

TRUNCATE TABLE users;

14. 条件查询、排序、分组

这些是 SQL 里非常高频的能力。

14.1 WHERE

1

SELECT * FROM users WHERE age >= 18;

14.2 ORDER BY

1

SELECT * FROM users ORDER BY id DESC;

14.3 LIMIT

1

SELECT * FROM users LIMIT 20;

14.4 GROUP BY

1
2
3

SELECT status, COUNT(*) AS total
FROM tasks
GROUP BY status;

14.5 HAVING

HAVING 通常配合分组结果使用：

1
2
3
4

SELECT status, COUNT(*) AS total
FROM tasks
GROUP BY status
HAVING total > 10;

15. JOIN 关联查询

这是关系型数据库最重要的优势之一。

假设：

• users 表存用户
• documents 表存文档

查询“每篇文档属于哪个用户”：

1
2
3

SELECT d.id, d.title, u.username
FROM documents d
JOIN users u ON d.user_id = u.id;

常见 JOIN：

• JOIN / INNER JOIN：两边都能匹配到才返回
• LEFT JOIN：左表全部保留，右表匹配不到则为 NULL

例如：

1
2
3

SELECT u.id, u.username, d.title
FROM users u
LEFT JOIN documents d ON u.id = d.user_id;

16. 索引 INDEX

索引的作用可以粗暴理解成：

让查询更快，但会增加写入成本和占用空间。

例如给用户名加索引：

1

CREATE INDEX idx_users_username ON users(username);

适合加索引的字段通常有：

• 经常出现在 WHERE 里的字段
• 经常用于排序的字段
• 经常用于 JOIN 的字段
• 唯一约束字段

常见误区：

• 不是索引越多越好
• 很小的表不一定需要索引
• 更新频繁的字段加太多索引会拖慢写入

17. 外键 FOREIGN KEY

外键用于表达表和表之间的引用关系。

例如：

1
2
3
4
5
6
7

CREATE TABLE documents (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id BIGINT NOT NULL,
    title VARCHAR(200) NOT NULL,
    CONSTRAINT fk_documents_user
        FOREIGN KEY (user_id) REFERENCES users(id)
);

它的作用是：

• 保证引用关系合法
• 避免出现“文档指向一个不存在用户”的情况

但很多实际项目里，也会选择在业务层保证关系，而不是强依赖数据库外键。

18. 事务 commit 和 rollback

你原来的笔记里提到“事务提交”，这部分非常重要。

事务适合用于：

• 多条 SQL 必须一起成功
• 中间某一步失败时要整体撤回

示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

try:
    with conn.cursor() as cursor:
        cursor.execute(
            "INSERT INTO accounts (name, balance) VALUES (%s, %s)",
            ("alice", 1000),
        )
        cursor.execute(
            "INSERT INTO accounts (name, balance) VALUES (%s, %s)",
            ("bob", 500),
        )
    conn.commit()
except Exception:
    conn.rollback()
    raise

常见规则：

• 成功后 conn.commit()
• 失败时 conn.rollback()

如果连接时配置：

1

autocommit=True

就表示每条 SQL 默认自动提交。

学习阶段可以用，但项目里要清楚：

一旦涉及多步写操作，通常还是要认真控制事务边界。

19. 查询结果获取

这是你原来笔记里已经提到的内容，这里整理成更清晰的版本。

19.1 fetchone()

只取一条：

1

row = cursor.fetchone()

19.2 fetchmany(n)

取前 n 条：

1

rows = cursor.fetchmany(10)

19.3 fetchall()

取全部：

1

rows = cursor.fetchall()

19.4 lastrowid

拿最后插入的自增 ID：

1

new_id = cursor.lastrowid

19.5 rowcount

看本次 SQL 影响了多少行：

1

count = cursor.rowcount

20. execute() 和 executemany()

20.1 execute()

执行一条 SQL：

1
2
3
4

cursor.execute(
    "INSERT INTO users (username, age) VALUES (%s, %s)",
    ("tom", 18),
)

20.2 executemany()

批量执行：

1
2
3
4
5
6
7
8
9
10

data = [
    ("tom", 18),
    ("alice", 20),
    ("bob", 22),
]

cursor.executemany(
    "INSERT INTO users (username, age) VALUES (%s, %s)",
    data,
)

适合：

• 批量插入
• 批量更新同类数据

21. 一段比较完整的 PyMySQL 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

import pymysql
import pymysql.cursors

conn = pymysql.connect(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="123456",
    database="demo",
    charset="utf8mb4",
    cursorclass=pymysql.cursors.DictCursor,
    autocommit=False,
)

try:
    with conn.cursor() as cursor:
        cursor.execute(
            """
            CREATE TABLE IF NOT EXISTS users (
                id BIGINT PRIMARY KEY AUTO_INCREMENT,
                username VARCHAR(100) NOT NULL UNIQUE,
                age INT NOT NULL DEFAULT 0,
                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
            )
            """
        )

        cursor.execute(
            "INSERT INTO users (username, age) VALUES (%s, %s)",
            ("tom", 18),
        )

        new_id = cursor.lastrowid

        cursor.execute(
            "SELECT id, username, age FROM users WHERE id = %s",
            (new_id,),
        )
        row = cursor.fetchone()
        print(row)

    conn.commit()
except Exception:
    conn.rollback()
    raise
finally:
    conn.close()

22. FastAPI 里怎么用 MySQL

在 FastAPI 项目里，MySQL 常见用途有：

• 存用户
• 存任务记录
• 存会话和消息
• 存文档元数据

最简单的同步示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import pymysql
from fastapi import FastAPI

app = FastAPI()

def get_conn():
    return pymysql.connect(
        host="127.0.0.1",
        user="root",
        password="123456",
        database="demo",
        charset="utf8mb4",
    )

@app.get("/users/{user_id}")
def get_user(user_id: int):
    conn = get_conn()
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                "SELECT id, username FROM users WHERE id = %s",
                (user_id,),
            )
            row = cursor.fetchone()
        return {"data": row}
    finally:
        conn.close()

不过实际项目里通常会再往前走一步：

• 使用连接池
• 抽 Repository / Service 层
• 使用 SQLAlchemy 等更完整的数据库层

23. AI 项目里比较常见的表设计

如果你在做文档问答、RAG、聊天系统，常见会有这些表：

• users
• sessions
• messages
• documents
• document_chunks
• tasks

例如：

• documents：存文档标题、路径、状态、所属用户
• tasks：存异步任务 ID、状态、错误信息
• messages：存对话消息和角色

也就是说：

MySQL 更适合存“事实记录”和“长期业务数据”，不适合拿来做高频缓存层。

24. 常见坑

24.1 用字符串拼 SQL

错误示例：

1

sql = f"SELECT * FROM users WHERE username = '{username}'"

这会有 SQL 注入风险。

正确做法：

1
2
3
4

cursor.execute(
    "SELECT * FROM users WHERE username = %s",
    (username,),
)

24.2 忘记提交事务

执行了 INSERT / UPDATE / DELETE 之后，如果没有 commit()，数据可能不会真正写入。

24.3 UPDATE / DELETE 不带 WHERE

这会影响整张表。

24.4 滥用 SELECT *

开发初期问题不大，但项目里更推荐明确字段，避免：

• 取了不需要的数据
• 表结构变动后影响接口
• 传输和解析成本增加

24.5 不建索引或乱建索引

都不对。

应该根据查询模式来设计索引。

24.6 字符集没统一

推荐统一使用：

• 数据库字符集：utf8mb4
• Python 连接：charset="utf8mb4"

24.7 把 MySQL 当缓存用

这会让数据库承担不适合它的高频压力。

25. 面试常问知识点

25.1 MySQL 的事务 ACID

事务的四个核心特性：

• 原子性：要么都成功，要么都失败
• 一致性：事务前后数据要保持业务规则正确
• 隔离性：并发事务之间互不干扰到一定程度
• 持久性：提交后的数据要可靠保存

实际回答时可以结合转账例子：扣钱和加钱必须放在同一个事务里。

25.2 常见事务隔离级别

从低到高可以这样记：

常见概念：

• 脏读：读到别人还没提交的数据
• 不可重复读：同一事务里两次读同一行，结果不同
• 幻读：同一事务里两次按条件查询，结果集行数不同

25.3 InnoDB 和 MyISAM 的区别

面试里最常回答：

实际项目里一般优先选择 InnoDB。

25.4 索引为什么能加快查询

MySQL InnoDB 的常见索引结构是 B+ 树。

它的好处：

• 树高度低，磁盘 IO 次数少
• 叶子节点有序，适合范围查询
• 查询可以从全表扫描变成按索引定位

但索引不是越多越好，因为写入、更新、删除时也要维护索引。

25.5 聚簇索引和非聚簇索引

InnoDB 里：

• 聚簇索引：主键索引，叶子节点存整行数据
• 非聚簇索引：普通二级索引，叶子节点通常存主键值

所以通过普通索引查到主键后，如果还需要其他字段，可能要再回到主键索引查一次，这叫“回表”。

如果查询字段都在索引里，可以直接从索引拿到结果，这叫“覆盖索引”。

25.6 联合索引和最左前缀原则

例如有联合索引：

1

CREATE INDEX idx_user_status_time ON messages(user_id, status, created_at);

它比较适合：

1
2
3

WHERE user_id = ?
WHERE user_id = ? AND status = ?
WHERE user_id = ? AND status = ? ORDER BY created_at

但如果只查：

1

WHERE status = ?

通常就用不上这个联合索引的完整能力，因为没有从最左边的 user_id 开始。

25.7 哪些情况容易导致索引失效

常见情况：

• 在索引列上做函数计算
• 对索引列做隐式类型转换
• LIKE '%xxx' 这种前缀不确定的模糊查询
• 联合索引不符合最左前缀原则
• 条件选择性太差，优化器判断走全表扫描更划算

真实排查时不要只靠猜，要看 EXPLAIN。

25.8 EXPLAIN 主要看什么

常看这些字段：

• type：访问类型，通常希望至少到 range、ref、const
• key：实际使用了哪个索引
• rows：预估扫描多少行
• Extra：是否有 Using filesort、Using temporary

它的作用是帮助判断 SQL 有没有走索引、扫描范围大不大、排序和临时表成本高不高。

25.9 MVCC 是什么

MVCC 可以理解成“多版本并发控制”。

它让读操作在很多场景下不用阻塞写操作，写操作也不用阻塞普通一致性读，从而提高并发性能。

在 InnoDB 里，MVCC 和 undo log、Read View 等机制有关。面试里如果只是基础回答，能说明它解决的是“并发读写时的一致性和性能问题”就够了。

25.10 慢查询一般怎么优化

常见步骤：

1. 先用慢查询日志或监控找到慢 SQL
2. 用 EXPLAIN 看执行计划
3. 检查 WHERE、JOIN、ORDER BY 是否有合适索引
4. 避免一次查太多字段和太多行
5. 必要时优化表结构、拆分大查询、引入缓存

不要一上来就说“加索引”，要先看 SQL 的真实访问路径。

26. 一套比较实用的学习顺序

建议这样学：

1. 先学会创建库、建表、插入、查询、更新、删除
2. 学会 WHERE、ORDER BY、LIMIT
3. 学会 GROUP BY 和 JOIN
4. 学会索引和事务
5. 学会用 PyMySQL 连接和执行参数化 SQL
6. 最后再接入 FastAPI、Celery、Redis

27. 总结

MySQL 的核心价值在于：

• 存长期结构化数据
• 支持 SQL 查询
• 支持关联查询
• 支持事务
• 适合承载业务主数据

对于 AI 项目，可以把它理解成：

• 用户数据存储层
• 会话和消息持久化层
• 文档元数据层
• 任务记录层

如果把你前面几份笔记连起来看，可以把它们理解成一条链：

• FastAPI / Drogon：接请求
• Redis：缓存 / 状态 / 中间层
• Celery：后台异步任务
• MySQL：长期持久化数据

28. 参考

• MySQL Official Docs: https://dev.mysql.com/doc/
• PyMySQL Docs: https://pymysql.readthedocs.io/
• SQL Syntax Reference: https://dev.mysql.com/doc/refman/8.0/en/sql-statements.html

Redis 学习笔记

1. Redis 是什么

Redis 是一个高性能的内存型键值数据库，常被用来做：

• 缓存
• 会话存储
• 计数器
• 排行榜
• 消息队列
• 分布式锁
• Celery 的 Broker / Result Backend

先把分工记住：MySQL 更适合放长期、结构化、需要稳定查询的数据，Redis 更适合放高频访问、生命周期短、对速度敏感的数据。

2. 为什么项目里经常会有 Redis

很多系统里都会遇到这些问题：

• 某些数据访问特别频繁
• 接口需要很快返回
• 需要临时状态，不想每次都查数据库
• 需要做限流、验证码、会话缓存
• 需要一个轻量消息通道

这时候 Redis 很适合。

因为它的特点是：

• 读写快
• 支持多种数据结构
• 支持过期时间
• 支持原子操作
• 既能做缓存，也能做简单消息系统

3. 在 AI 项目里 Redis 常见用法

在 AI 模型开发场景里，Redis 很常见：

• 缓存热点查询结果
• 缓存用户会话上下文
• 给 Celery 当 Broker
• 给 Celery 当任务结果存储
• 存任务状态、进度、短期中间结果
• 做限流
• 用 Streams 传递流式输出 token

和你前面整理的两份笔记连起来理解：

• FastAPI：接请求
• Redis：缓存 / 排队 / 状态层
• Celery：异步处理后台任务

4. Redis 和 MySQL 的区别

可以这样对比理解：

简单理解：

• 用户资料、订单、文档元数据更适合 MySQL
• 验证码、登录态、缓存结果、限流计数更适合 Redis

5. 安装和启动

如果本地已经装好了 Redis，可以直接启动服务。

常见方式：

1

redis-server

如果只是临时学习，也可以用 Docker：

1

docker run -d --name redis -p 6379:6379 redis:7

测试服务是否正常：

1

redis-cli ping

如果返回：

1

PONG

说明 Redis 正常启动。

6. Python 连接 Redis

Python 项目里最常用的是 redis-py。

安装：

1

pip install redis

最基础的连接方式：

1
2
3
4
5
6
7
8
9
10
11

import redis

r = redis.Redis(
    host="127.0.0.1",
    port=6379,
    db=0,
    decode_responses=True,
)

r.set("name", "tom")
print(r.get("name"))

说明：

• db=0 表示使用第 0 个逻辑库
• decode_responses=True 表示自动把结果解码为字符串

如果不加 decode_responses=True，很多结果会是 bytes。

7. Redis 的基本概念

Redis 里最核心的概念就是：

• key
• value

所有数据都以 key -> value 的形式存储。

例如：

1
2
3

user:1:name -> tom
task:123:status -> SUCCESS
cache:article:100 -> {...}

实际项目里非常重要的一点是：

key 命名要有规则。

常见命名方式：

• user:1001
• session:token:abc123
• task:ingest:001
• cache:query:hash_xxx

这样后续排查和维护会清晰很多。

7.1 逻辑库 db

Redis 默认会提供多个逻辑库，常见写法里的：

• db=0
• redis://127.0.0.1:6379/1
• redis://127.0.0.1:6379/2

这里最后的数字就是逻辑库编号。

它常用于做基础隔离，例如：

• 0：普通缓存
• 1：Celery broker
• 2：Celery backend

不过要注意：

逻辑库只是“轻隔离”，不是严格意义上的多实例隔离。

如果项目越来越复杂，真正需要资源隔离、权限隔离、性能隔离时，通常还是会考虑拆 Redis 实例。

8. 常用通用命令

8.1 设置和获取

1
2

SET name tom
GET name

8.2 判断是否存在

1

EXISTS name

8.3 删除

1

DEL name

8.4 查看过期时间

1

TTL name

8.5 给 key 设置过期

1

EXPIRE name 60

表示 60 秒后过期。

8.6 直接带过期时间写入

1

SET code 123456 EX 300

表示写入验证码，并在 300 秒后自动过期。

8.7 查看库里的 key

学习阶段常见命令：

1

KEYS *

但要注意：

KEYS * 适合本地调试，不适合在线上大库里频繁使用。

更稳妥的方式通常是：

1

SCAN 0 MATCH user:* COUNT 100

SCAN 更适合在数据量较大时逐步遍历。

9. Redis 的核心数据结构

Redis 不只是简单字符串，它支持很多数据结构。这个是 Redis 很有价值的地方。

最常用的有：

• String
• Hash
• List
• Set
• Sorted Set
• Stream

10. String

这是 Redis 最基础、最常用的数据类型。

适合：

• 普通缓存
• 验证码
• token
• JSON 字符串
• 计数器

10.1 基本命令

1
2

SET username alice
GET username

10.2 自增计数

1
2
3

SET page_views 0
INCR page_views
INCRBY page_views 5

这类操作很适合：

• 接口访问次数统计
• 点赞数
• 限流计数

10.3 Python 示例

1
2
3
4
5
6
7
8
9

import redis

r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

r.set("article:1:title", "Redis Intro")
title = r.get("article:1:title")

r.set("counter:views", 0)
r.incr("counter:views")

11. Hash

Hash 就是一个 key 下挂多组 field-value。

适合：

• 用户对象
• 配置项
• 文档元信息
• 任务状态对象

11.1 基本命令

1
2
3

HSET user:1 name tom age 18 city shanghai
HGET user:1 name
HGETALL user:1

11.2 为什么 Hash 很常用

因为很多业务对象天然就是字段结构，比如：

1
2
3
4

user:1
  name -> tom
  age -> 18
  city -> shanghai

相比把整个对象都塞进一个 JSON 字符串里，Hash 在局部更新字段时更方便。

11.3 Python 示例

1
2

r.hset("user:1", mapping={"name": "tom", "age": 18, "city": "shanghai"})
user = r.hgetall("user:1")

12. List

List 是有序列表。

适合：

• 简单消息队列
• 待处理任务列表
• 最近操作记录

12.1 基本命令

1
2
3

LPUSH queue:tasks task1
LPUSH queue:tasks task2
RPOP queue:tasks

直观一点看，就是：

• 左边入队
• 右边出队

12.2 阻塞读取

1

BLPOP queue:tasks 0

表示如果队列为空就阻塞等待。

这可以做一个很简单的消费者模型，但它更适合轻量场景，不适合复杂消费确认机制。

12.3 使用建议

List 可以做简单队列，但如果项目要：

• 消费组
• 确认机制
• 多消费者协作
• 更强的消息语义

通常会考虑：

• Redis Streams
• RabbitMQ
• Kafka

13. Set

Set 是无序且元素唯一的集合。

适合：

• 去重
• 标签集合
• 记录某用户已点赞哪些内容
• 共同好友 / 共同标签等集合计算

13.1 基本命令

1
2
3

SADD tags:article:1 redis python database
SMEMBERS tags:article:1
SISMEMBER tags:article:1 redis

13.2 集合运算

1
2
3
4
5

SADD set1 a b c
SADD set2 b c d
SINTER set1 set2
SUNION set1 set2
SDIFF set1 set2

这对“标签交集、权限集合、去重集合”这类场景非常方便。

14. Sorted Set

Sorted Set 是带分数、可排序的集合。

适合：

• 排行榜
• 按时间排序的数据
• 按热度排序的数据

14.1 基本命令

1
2
3
4

ZADD ranking 100 tom 95 alice 88 bob
ZRANGE ranking 0 -1
ZREVRANGE ranking 0 -1
ZSCORE ranking tom

这里每个成员都有一个 score。

14.2 场景理解

例如文章热度榜：

• article:100 -> 200
• article:101 -> 150
• article:102 -> 320

就可以按 score 排序快速拿到前 N 名。

15. Stream

Streams 是 Redis 较强的一类消息结构。

适合：

• 事件流
• 多消费者消费
• 流式 token 传递
• 日志型消息

在 AI 项目里，它很适合做：

• Worker 持续写出 token
• Gateway / FastAPI 持续消费 token
• 流式返回给前端

15.1 基本命令

写入：

1

XADD chat:stream:1 * token hello

读取：

1

XREAD COUNT 10 STREAMS chat:stream:1 0

15.2 消费组

如果是多个消费者协作，可以用消费组：

1
2

XGROUP CREATE chat:stream:1 group1 0 MKSTREAM
XREADGROUP GROUP group1 consumer1 COUNT 10 STREAMS chat:stream:1 >

15.3 为什么 Streams 比 List 更适合流式场景

因为它支持：

• 消息 ID
• 消费组
• 待确认消息
• 更像日志流的结构

如果后面你做：

• SSE
• token 流输出
• 异步处理链路

Streams 会比简单 List 更稳一些。

16. 过期时间和 TTL

Redis 很核心的一点就是支持 key 过期。

这在缓存场景里非常重要。

16.1 设置过期

1

SET session:token:abc user_1 EX 3600

表示 1 小时后自动过期。

16.2 查看剩余时间

1

TTL session:token:abc

16.3 为什么过期时间很重要

因为很多数据本来就不该永久保存，例如：

• 登录态
• 验证码
• 缓存结果
• 限流计数
• 任务中间状态

如果不设置过期时间，Redis 很容易越堆越大。

17. 持久化

Redis 虽然主要是内存数据库，但也支持持久化。

常见方式：

• RDB
• AOF

17.1 RDB

它更像是按时间点保存一份内存快照。

特点：

• 文件紧凑
• 恢复快
• 更适合定期备份

17.2 AOF

它的思路是把写命令按顺序追加记录下来。

特点：

• 数据恢复通常更完整
• 文件可能更大
• 更强调操作日志式恢复

17.3 怎么理解

如果 Redis 里只是放缓存，丢了还能重建，那持久化要求没那么高。

如果 Redis 里存了：

• 任务状态
• 流式消息
• 某些关键临时业务状态

那就要更认真考虑持久化策略。

18. Pub/Sub

Redis 还支持发布订阅。

基本命令：

1
2

SUBSCRIBE news
PUBLISH news "hello"

适合：

• 简单实时通知
• 临时消息广播

但它有一个明显特点：

如果订阅者不在线，消息一般不会帮你补回来。

所以在很多需要“可回放、可追踪”的场景里，Streams 通常比 Pub/Sub 更合适。

19. 事务和 Lua

Redis 也支持事务和 Lua 脚本。

19.1 事务

基本写法：

1
2
3
4

MULTI
SET key1 value1
INCR counter
EXEC

适合把一组命令放在一起执行。

19.2 Lua 脚本

Lua 常用于：

• 多命令原子操作
• 分布式锁安全释放
• 把多个步骤封装成一个服务端操作

在实际项目里，Lua 经常是为了解决“先判断再删除”这种竞态问题。

20. 分布式锁

Redis 常被用来做分布式锁。

最常见写法是：

1

SET lock:task_1 request_abc NX EX 30

含义：

• NX：只有 key 不存在时才设置成功
• EX 30：30 秒后自动过期

如果返回成功，就表示拿到了锁。

20.1 为什么需要过期时间

因为如果加锁后进程挂了，没有过期时间的话，锁就可能一直不释放。

20.2 为什么不能直接 DEL

因为可能出现这种情况：

1. 线程 A 拿到锁
2. 锁过期
3. 线程 B 拿到同名新锁
4. 线程 A 这时候再执行 DEL
5. 把线程 B 的锁删掉了

所以更安全的做法是：

• 加锁时写入唯一值
• 解锁时先判断 value 是否还是自己的
• 判断和删除要放在一个原子操作里，通常用 Lua

20.3 Python 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import uuid

import redis

r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

lock_key = "lock:task:1"
lock_value = str(uuid.uuid4())

locked = r.set(lock_key, lock_value, nx=True, ex=30)

if locked:
    try:
        print("do something")
    finally:
        unlock_script = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        else
            return 0
        end
        """
        r.eval(unlock_script, 1, lock_key, lock_value)

21. 缓存是 Redis 最常见的用途

21.1 最简单的缓存思路

先查 Redis：

• 有数据就直接返回
• 没数据再查数据库或调用模型服务
• 查到结果后再写回 Redis

这就是最常见的缓存模式。

21.2 Python 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import json

import redis

r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

def get_article(article_id: int):
    cache_key = f"article:{article_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    data = {"id": article_id, "title": "Redis Intro"}
    r.set(cache_key, json.dumps(data), ex=300)
    return data

21.3 缓存设计里要注意什么

• 一定要设计 TTL
• 不要无脑缓存超大对象
• key 命名要稳定
• 更新数据库时要考虑缓存同步

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

这是 Redis 学习里很常见的三个词。

22.1 缓存穿透

请求的数据本来就不存在：

• Redis 没有
• 数据库也没有
• 每次都打到后端

常见处理：

• 参数校验
• 对空结果也做短期缓存
• 布隆过滤器

22.2 缓存击穿

某个热点 key 失效瞬间，大量请求同时打到后端。

常见处理：

• 热点 key 不要同时过期
• 加锁重建缓存
• 提前刷新

22.3 缓存雪崩

大量 key 在同一时间过期，导致后端压力暴涨。

常见处理：

• TTL 加随机值
• 分批过期
• 多级缓存

23. 限流

Redis 很适合做接口限流，因为：

• INCR 是原子操作
• 配合 TTL 很方便

23.1 最简单的固定窗口限流

思路：

1. 某个用户请求一次就 INCR
2. 第一次请求时设置过期时间
3. 超过阈值就拒绝

Python 示例：

1
2
3
4
5
6
7
8
9
10

import redis

r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

def allow_request(user_id: str, limit: int = 10, window: int = 60) -> bool:
    key = f"rate_limit:{user_id}"
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)
    return current <= limit

这个模式在：

• 登录接口
• 短信发送
• AI 推理调用

都很常见。

24. Session / 登录态

Redis 也经常用来存登录态或短期会话。

例如：

1

session:token:abc123 -> user_id=1001

然后设置：

1

EXPIRE session:token:abc123 3600

这样用户 1 小时不活跃就自动失效。

25. FastAPI 里怎么用 Redis

FastAPI 项目里，Redis 常见用途有：

• 缓存接口结果
• 存验证码
• 存 session
• 做限流
• 存短期任务状态

25.1 一个简单缓存例子

这里用 redis.asyncio 更符合 FastAPI 的异步风格。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

import json

from fastapi import FastAPI
from redis.asyncio import Redis

app = FastAPI()

redis_client = Redis(
    host="127.0.0.1",
    port=6379,
    db=0,
    decode_responses=True,
)

@app.get("/articles/{article_id}")
async def get_article(article_id: int):
    cache_key = f"article:{article_id}"
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)

    data = {"id": article_id, "title": "Redis Intro"}
    await redis_client.set(cache_key, json.dumps(data), ex=300)
    return data

补充：

• redis.asyncio 更适合异步接口
• 正式项目里通常会统一封装 Redis 客户端
• 服务关闭时最好把客户端连接关闭掉

25.2 一个简单限流例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from fastapi import FastAPI, HTTPException, Request
from redis.asyncio import Redis

app = FastAPI()
redis_client = Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

@app.get("/infer")
async def infer(request: Request):
    client_ip = request.client.host
    key = f"rate_limit:{client_ip}"

    current = await redis_client.incr(key)
    if current == 1:
        await redis_client.expire(key, 60)

    if current > 20:
        raise HTTPException(status_code=429, detail="too many requests")

    return {"message": "ok"}

25.3 生命周期里管理 Redis 客户端

如果项目规模再大一点，通常会把 Redis 放到 FastAPI 生命周期里初始化和关闭。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

from contextlib import asynccontextmanager

from fastapi import FastAPI
from redis.asyncio import Redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = Redis(
        host="127.0.0.1",
        port=6379,
        db=0,
        decode_responses=True,
    )
    yield
    await app.state.redis.aclose()

app = FastAPI(lifespan=lifespan)

这样做的好处：

• 连接初始化位置统一
• 更方便复用
• 关闭服务时能正常释放资源

26. Celery 为什么经常配 Redis

你在 Celery.md 里已经接触到了这一点。

Redis 在 Celery 里通常可以做两件事：

• Broker：存待执行任务消息
• Result Backend：存任务状态和结果

示例：

1
2
3
4
5
6
7

from celery import Celery

app = Celery(
    "demo",
    broker="redis://127.0.0.1:6379/1",
    backend="redis://127.0.0.1:6379/2",
)

这里通常会把逻辑库分开：

• /1：Broker
• /2：Backend

这样排查和清理都更方便。

26.1 为什么很多人喜欢 Redis + Celery

因为它足够简单：

• 环境轻
• 容易本地启动
• Python 项目接入快

但也要注意：

• 不要无限保存任务结果
• 大结果不要直接堆 Redis
• 任务状态和业务长期数据要区分

27. AI / RAG 项目里 Redis 的几个实用位置

如果你做的是 RAG 或模型服务，Redis 常放在这些位置：

27.1 热点缓存

例如：

• 相同 query 的检索结果缓存
• embedding 结果缓存
• 频繁访问的文档元数据缓存

27.2 会话上下文缓存

例如：

• 最近 N 轮聊天记录
• 临时上下文摘要
• 会话状态

27.3 任务状态层

例如：

• task:123:status
• task:123:progress
• task:123:result_ref

27.4 流式输出通道

例如：

• Worker 生成 token 后写入 chat:stream:{task_id}
• Gateway 或 API 服务持续消费
• 对外通过 SSE 返回

这个设计非常适合“生成过程持续输出”的场景。

28. Redis Streams 在流式输出里的思路

一个简化流程可以这样理解：

1. FastAPI 收到对话请求
2. 把任务提交给 Celery
3. Worker 调模型生成 token
4. 每生成一段 token 就 XADD 到某个 Stream
5. Gateway 或 API 层 XREAD / XREADGROUP
6. 持续把内容推给前端

例如 Stream key：

1

chat:stream:task_1001

这个模式的好处：

• 推理和对外接口解耦
• 可以观察中间过程
• 可以做断线恢复或补读

29. key 设计建议

Redis 真正到了项目里，key 设计非常重要。

建议：

• 带业务前缀
• 带对象类型
• 带唯一 ID
• 必要时带版本

例如：

• user:1001
• session:token:abc123
• cache:article:100
• task:ingest:001:status
• chat:stream:task_1001

不要这样写：

• a
• test
• temp1

这种 key 后期几乎不可维护。

30. Redis 内存管理要有意识

Redis 很快，但它是以内存为核心的，所以内存管理很重要。

需要注意：

• 大 key 会拖慢操作
• 无过期时间的缓存容易无限增长
• Stream、List、Set 如果不清理会不断膨胀
• 不要把超大模型结果直接塞进去

对 AI 项目尤其重要：

• 大文本结果建议存数据库或文件
• Redis 里尽量存引用、状态、索引、小块结果

30.1 排查时少用危险大命令

实际项目里还要注意一些“能用，但别乱用”的命令。

例如：

• KEYS *
• FLUSHDB
• FLUSHALL

其中：

• KEYS * 在大库里可能带来明显扫描压力
• FLUSHDB 会清空当前库
• FLUSHALL 会清空整个 Redis 所有库

本地学习可以用，但线上环境要非常谨慎。

31. 常见坑

31.1 把 Redis 当永久数据库

Redis 可以持久化，但它更常见的定位还是：

• 缓存
• 状态层
• 高速临时存储

长期核心业务数据还是更适合数据库。

31.2 忘记设置 TTL

这是最常见的问题之一。

很多缓存、验证码、限流 key 如果没有 TTL，会一直堆着不清。

31.3 一个 Redis 库里什么都混着放

例如：

• 缓存
• Celery broker
• Celery backend
• 会话
• 任务流

全部都混在同一个逻辑库里，会让排查和维护变得很乱。

至少应该在命名和逻辑库上做一定隔离。

31.4 把超大对象直接塞进 Redis

例如：

• 整篇长文档
• 大块模型输出
• 大量 embedding

这些会明显增加内存压力，也会拖慢网络传输。

31.5 分布式锁写得不安全

只会 SETNX 或只会 DEL，但没有考虑：

• 锁超时
• 唯一 value
• 原子释放

这种写法在并发场景里容易出问题。

31.6 用 Pub/Sub 做必须可靠消费的任务

Pub/Sub 更像广播，不是强消息队列。

如果你要求：

• 消息可追踪
• 消费可确认
• 掉线后能补读

Streams 通常更合适。

32. 面试常问知识点

32.1 Redis 为什么快

常见回答角度：

• 数据主要在内存里，少了磁盘随机 IO
• 核心命令执行模型简单，数据结构专门为内存访问优化
• 单线程执行命令，避免大量线程切换和锁竞争
• 使用 IO 多路复用处理大量连接

注意这里的“单线程”主要指命令执行路径。Redis 在持久化、异步删除、网络 IO 等方面也会用到后台线程。

32.2 Redis 单线程为什么还能扛高并发

因为 Redis 的瓶颈通常不在 CPU，而在网络和内存访问。

它用一个主线程顺序执行命令，可以减少锁竞争；同时用 IO 多路复用同时管理很多连接，所以能处理大量短平快的请求。

但如果出现：

• 大 key
• 慢命令
• Lua 脚本执行太久
• 一次性返回大量数据

仍然会阻塞其他请求。

32.3 常见数据结构怎么选

可以按场景回答：

面试里重点不是背命令，而是能把业务场景和数据结构对应起来。

32.4 缓存穿透、击穿、雪崩怎么区分

• 穿透：查一个根本不存在的数据，每次都打到数据库
• 击穿：一个热点 key 过期，大量请求同时打到数据库
• 雪崩：大量 key 同时过期，数据库瞬间被打满

常见解决：

• 穿透：参数校验、缓存空值、布隆过滤器
• 击穿：互斥锁重建缓存、热点 key 提前刷新
• 雪崩：TTL 加随机值、分批过期、多级缓存

32.5 Redis 的过期删除和内存淘汰

Redis key 到期后，不一定会在那一瞬间立刻删除。

常见机制：

• 惰性删除：访问 key 时发现过期，再删除
• 定期删除：后台定期抽样检查一部分过期 key
• 内存淘汰：内存达到上限时，根据淘汰策略清理 key

所以面试里要区分：

• 过期删除：处理已经设置 TTL 的 key
• 内存淘汰：内存不够时按策略腾空间

32.6 RDB 和 AOF 怎么选

简单理解：

• RDB：定期快照，文件小，恢复快，但可能丢失最近一段时间数据
• AOF：记录写命令，数据更完整，但文件更大，恢复可能更慢

如果 Redis 只是缓存，持久化要求可以低一些。

如果 Redis 里放任务状态、消息流等比较重要的数据，就要认真考虑 AOF、备份和恢复策略。

32.7 Redis 事务和 MySQL 事务有什么不同

Redis 的事务主要是：

• MULTI 开始
• 命令排队
• EXEC 一次性执行

它保证一组命令按顺序连续执行，但不像 MySQL 那样强调复杂的回滚、隔离级别和 MVCC。

如果需要把“读取、判断、修改”做成严格原子操作，通常会用 Lua 脚本。

32.8 分布式锁面试怎么回答

基础写法：

1

SET lock:xxx unique_value NX EX 30

要点：

• NX 保证只有不存在时才能加锁
• EX 防止进程挂掉后锁永远不释放
• value 要唯一，避免误删别人的锁
• 解锁时“判断 value + 删除 key”要原子执行，通常用 Lua

实际项目里还要关注锁超时、业务执行时间、锁续期和异常释放。

32.9 Redis 高可用常见方案

面试常见层次：

• 主从复制：读写分离、数据副本
• Sentinel：监控主节点，故障时自动选主
• Cluster：分片存储，适合更大数据量和更高吞吐

简单项目里一个 Redis 实例就够用；线上核心系统通常要考虑高可用、备份、监控和容量规划。

33. 一套比较实用的学习顺序

建议按这个顺序掌握：

1. 先学会连接 Redis 和 redis-cli
2. 掌握 String、Hash、List、Set、Sorted Set
3. 学会 TTL 和过期机制
4. 学会缓存和限流
5. 学会分布式锁的基本思路
6. 再看 Pub/Sub 和 Streams
7. 最后再把 Redis 接到 FastAPI 和 Celery

34. 总结

Redis 的核心价值可以概括成几句话：

• 它很快，适合高频读写
• 它支持多种数据结构，不只是一个简单缓存
• 它天然适合存“短期状态”和“可过期数据”
• 它很适合做 FastAPI 和 Celery 之间的中间层

在你这套 AI 模型开发学习链路里，可以这样理解它的位置：

• FastAPI：提供接口
• Redis：缓存、状态、消息中间层
• Celery：执行后台异步任务
• MySQL：长期结构化数据存储

如果后面你继续往“RAG 系统、推理服务、异步任务平台”方向走，Redis 几乎一定会成为一个基础组件。

35. 参考

• Redis Docs: https://redis.io/docs/
• Redis Data Types: https://redis.io/docs/latest/develop/data-types/
• Redis Commands: https://redis.io/docs/latest/commands/
• Redis EXPIRE: https://redis.io/docs/latest/commands/expire/
• Redis TTL: https://redis.io/docs/latest/commands/ttl/
• Redis Streams: https://redis.io/docs/latest/develop/data-types/streams/
• redis-py Guide: https://redis.io/docs/latest/develop/clients/redis-py/

LangChain

LangChain 是一个 LLM 应用开发框架。它本身不是模型，也不是向量数据库，而是把模型调用、Prompt、输出解析、工具调用、Embedding、VectorStore、Retriever、Agent 等能力包装成统一组件，方便开发者快速组合 LLM 应用。

可以把 LangChain 理解成一个“编排层”：

1
2
3
4
5

用户输入
  -> PromptTemplate
  -> ChatModel / LLM
  -> OutputParser
  -> 返回结果

如果是 RAG 应用，则会多出文档加载、切分、向量化、检索和上下文组装：

1
2
3
4
5

离线索引：
文档 -> Loader -> TextSplitter -> Embedding -> VectorStore

在线问答：
问题 -> Retriever -> 相关 chunk -> Prompt -> LLM -> Answer

1. LangChain 主要解决什么问题

1. 统一模型调用接口
   不同模型厂商的 API 格式不同，LangChain 把它们统一成 invoke()、stream()、batch() 等接口。

2. 统一 Prompt 和链式编排
   使用 ChatPromptTemplate 管理 system/user 消息模板，再通过管道方式把 prompt、model、parser 串起来。

3. 快速搭建 RAG
   提供 DocumentLoader、TextSplitter、Embeddings、VectorStore、Retriever 等组件，降低搭建检索增强生成应用的样板代码量。

4. 支持工具调用和 Agent
   可以把搜索、数据库查询、计算函数等能力包装成 tool，让模型根据任务决定是否调用工具。

5. 生态集成丰富
   LangChain 对 OpenAI、Anthropic、HuggingFace、Ollama、FAISS、Chroma、Milvus、Qdrant、PGVector 等都有集成，适合快速验证方案。

2. 入门应用：用 LangChain 搭一个本地 RAG 文档问答

这个例子和我当前的 RAG 项目思路一致：文档切分、Embedding、FAISS 检索、Prompt 组装、调用本地 vLLM 的 OpenAI-compatible 接口。

安装依赖：

1

pip install -U langchain langchain-core langchain-community langchain-openai langchain-text-splitters langchain-huggingface faiss-cpu sentence-transformers

准备一个本地知识文件：

1

data/knowledge.txt

示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68