---
id: plan.week14
title: "第 14 周：后端补强——消息队列、gRPC、Redis / MySQL 专题、全部项目部署、CI/CD"
type: week
domain: plan
stack: [go, mysql, redis, mq, grpc-microservices, docker, nginx, linux, cicd, elasticsearch, mongodb, k8s, data-pipeline, nodejs, java-spring, network-os, web-security]
week: [14]
est_hours: 12
order: 14
---

# 第 14 周：后端补强——消息队列、gRPC、Redis / MySQL 专题、全部项目部署、CI/CD

> 第四段的第 1 周。上周 `week13-agent-m4-eval-lowcode.md`，下周 `week15-package-resume.md`。

**为什么有这一周**：项目三完成后，作品里已经有 Redis、Redis Streams、Docker。但招聘里“熟悉 MySQL、Redis、消息队列、微服务”的要求，面试会顺着问下去（索引为什么失效、隔离级别、缓存击穿、消息重复消费……）。这一周用**能运行的小实验**把这些高频题过一遍，再把三个项目部署上线、配好 CI/CD。

**原则**：每个专题做到“能跑一个最小例子 + 能用自己的话回答面试题”，不深挖源码。Go 网关是**可选**内容，时间不够直接跳过（见总纲第 6 节）。

**根据 BOSS直聘 2026-10-23 调研的调整**：Go 只有 2 个岗位是硬性要求，所以 Go 网关继续保持可选；调研中出现的 ES 混合检索、MongoDB、K8s、数据管道、Node.js、Java（各有若干岗位提到）以及计算机基础，放到下面的“第 7 天（调整）”做**了解级**扩展：每项跑通专题里的示例、读完面试题，不做项目。

**本周环境**（2026-10 实际验证）：RabbitMQ 4.0、Redis 7 以上、MySQL 8（我在 MariaDB 11.8 上跑过同样的 SQL，`EXPLAIN` 的主要列相同）、Python 3.12 + aio-pika 9 / redis-py / PyMySQL / grpcio 1.84、Go 1.27。

**参考资料**

- RabbitMQ 教程：https://www.rabbitmq.com/tutorials ；aio-pika：https://docs.aio-pika.com/
- Kafka 介绍：https://kafka.apache.org/intro
- gRPC Python：https://grpc.io/docs/languages/python/ ；Protocol Buffers：https://protobuf.dev/
- Redis 文档：https://redis.io/docs/latest/
- MySQL 8 参考手册（索引、EXPLAIN、InnoDB 事务）：https://dev.mysql.com/doc/refman/8.0/en/

## 本周最终目录

```text
backend-extend/
├── pyproject.toml
├── docker-compose.dev.yml        # rabbitmq、redis、mysql
├── proto/backend_extend/grpcgen/run.proto
├── scripts/gen_proto.py
├── src/backend_extend/
│   ├── __init__.py
│   ├── mq_rabbit.py              # 声明交换机 / 队列 / 死信、发布、幂等消费
│   ├── grpcgen/__init__.py       # 生成的代码放这里（run_pb2*.py 不手改）
│   ├── grpc_service.py           # gRPC 服务：一元调用 + 服务端流
│   ├── redis_patterns.py         # 分布式锁、缓存三大问题、滑动窗口限流
│   └── mysql_lab.py              # 索引 / EXPLAIN、隔离级别、原子扣减、转账
└── tests/
    ├── conftest.py               # 测试前自动生成 gRPC 代码
    ├── test_rabbit.py
    ├── test_grpc.py
    ├── test_redis_patterns.py
    └── test_mysql_lab.py
deploy/                            # 服务器上的总部署
├── docker-compose.yml
├── nginx/conf.d/apps.conf
└── deploy.md
go-gateway/                        # 可选：Go 写的鉴权网关
├── main.go
└── main_test.go
.github/workflows/ci.yml           # 或 .gitea/workflows/ci.yml
```

---

## 第 1 天：消息队列——RabbitMQ 动手 + Kafka 概念

### 学习内容

**1. 为什么要消息队列**

- 说明：消息队列把“发起方”和“处理方”解耦：发起方把消息放进队列就返回，处理方按自己的速度慢慢处理。三个典型用途：**异步**（注册后发欢迎邮件不阻塞接口）、**削峰**（瞬间涌入的请求先排队）、**解耦**（一个事件多个系统各自订阅）。代价是系统变复杂：要考虑消息丢失、重复、顺序、积压。
- 案例：项目三里“创建任务 → 后台执行”现在是数据库表 + 看门狗实现的；如果执行器要部署成多台机器、任务量很大，就可以改成“接口写库并发消息 → 执行器从队列取任务”。

**2. RabbitMQ 的核心概念**

| 概念 | 是什么 | 类比 |
| --- | --- | --- |
| 生产者 / 消费者 | 发消息 / 收消息的程序 | |
| 交换机（exchange） | 接收消息，按规则转发给队列；类型有 direct（精确匹配路由键）、topic（通配符 `task.*`）、fanout（广播） | 邮局分拣 |
| 队列（queue） | 存消息的地方，消费者从这里取 | 信箱 |
| 绑定（binding） | 交换机和队列之间的转发规则 | |
| 确认（ack） | 消费者处理完告诉服务器“可以删了”；没确认就断开，消息会重新投递 | |
| 死信（dead letter） | 被拒绝、过期或超过长度的消息转到另一个交换机，方便排查和重试 | 退信 |
| 预取（prefetch） | 每个消费者最多同时拿多少条未确认的消息，防止一个消费者被压垮 | |

- 说明：**消息不丢**要三处配合：交换机和队列声明为持久（durable）+ 消息设为持久（delivery_mode=2）、生产者开启发布确认（publisher confirms）、消费者**处理成功后再 ack**。
- 说明：**消息会重复**：网络抖动、消费者处理完但 ack 前崩溃，都会导致同一条消息被投递两次。所以消费者必须**幂等**：用消息 ID（或业务唯一键）去重，生产环境去重记录放在 Redis（`SET NX` + 过期）或数据库唯一索引里。

**3. 动手：交换机 + 队列 + 死信 + 幂等消费**

先启动本周的依赖：

```yaml
# 文件：backend-extend/docker-compose.dev.yml
services:
  rabbitmq:
    image: rabbitmq:4-management       # 带管理界面：http://localhost:15672（guest / guest，只能本机登录）
    ports: ["5672:5672", "15672:15672"]
  redis:
    image: redis:7
    ports: ["6379:6379"]
  mysql:
    image: mysql:8
    environment:
      MYSQL_ROOT_PASSWORD: root
      MYSQL_DATABASE: learn
      MYSQL_USER: app
      MYSQL_PASSWORD: app
    ports: ["3306:3306"]
```

```toml
# backend-extend/pyproject.toml（节选）
[project]
name = "backend-extend"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["aio-pika>=9", "redis>=5", "pymysql>=1.1", "grpcio>=1.70"]

[dependency-groups]
dev = ["pytest>=8", "pytest-asyncio>=0.24", "grpcio-tools>=1.70"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
pythonpath = ["src"]
```

```python
# 文件：backend-extend/src/backend_extend/__init__.py
```

```python
# 文件：backend-extend/src/backend_extend/mq_rabbit.py
"""RabbitMQ 最小示例：topic 交换机 + 持久队列 + 死信队列 + 幂等消费。"""
import json
from collections.abc import Awaitable, Callable

import aio_pika
from aio_pika import DeliveryMode, ExchangeType, Message
from aio_pika.abc import AbstractChannel, AbstractExchange, AbstractIncomingMessage, AbstractQueue


async def declare(channel: AbstractChannel, prefix: str) -> tuple[AbstractExchange, AbstractQueue, AbstractQueue]:
    """声明是幂等的：已经存在且参数相同就直接用；参数不同会报错（改参数要先删队列）。"""
    dlx = await channel.declare_exchange(f"{prefix}.dlx", ExchangeType.DIRECT, durable=True)
    dead = await channel.declare_queue(f"{prefix}.dead", durable=True)
    await dead.bind(dlx, routing_key=prefix)

    events = await channel.declare_exchange(f"{prefix}.events", ExchangeType.TOPIC, durable=True)
    tasks = await channel.declare_queue(
        f"{prefix}.tasks", durable=True,
        arguments={"x-dead-letter-exchange": f"{prefix}.dlx", "x-dead-letter-routing-key": prefix})
    await tasks.bind(events, routing_key="task.*")   # task.created、task.retry 都会进来
    return events, tasks, dead


async def publish(exchange: AbstractExchange, routing_key: str, body: dict, message_id: str) -> None:
    msg = Message(json.dumps(body, ensure_ascii=False).encode(), content_type="application/json",
                  delivery_mode=DeliveryMode.PERSISTENT, message_id=message_id)
    await exchange.publish(msg, routing_key=routing_key)   # 通道开了发布确认时，这里会等服务器确认


class IdempotentConsumer:
    """按 message_id 去重。演示用内存集合；生产环境用 Redis SET NX 或数据库唯一索引。"""

    def __init__(self, handler: Callable[[dict], Awaitable[None]]):
        self.handler = handler
        self.seen: set[str] = set()
        self.duplicates = 0
        self.failed = 0

    async def handle(self, message: AbstractIncomingMessage) -> None:
        # 正常结束 → ack；抛出异常 → reject(requeue=False) → 进入死信队列
        async with message.process(requeue=False):
            if message.message_id in self.seen:
                self.duplicates += 1
                return
            await self.handler(json.loads(message.body))
            self.seen.add(message.message_id or "")


async def drain(queue: AbstractQueue, consumer: IdempotentConsumer, max_messages: int, timeout: float = 2.0) -> int:
    """逐条取消息处理（测试和脚本用）。常驻服务用 queue.consume(consumer.handle) 或 queue.iterator()。"""
    n = 0
    while n < max_messages:
        msg = await queue.get(timeout=timeout, fail=False)
        if msg is None:
            break
        try:
            await consumer.handle(msg)
        except Exception:
            consumer.failed += 1
        n += 1
    return n


async def connect(url: str) -> tuple[aio_pika.abc.AbstractRobustConnection, AbstractChannel]:
    conn = await aio_pika.connect_robust(url)          # robust：断线自动重连
    channel = await conn.channel(publisher_confirms=True)
    await channel.set_qos(prefetch_count=10)
    return conn, channel
```

```python
# 文件：backend-extend/tests/test_rabbit.py
import os
import uuid

import pytest

from backend_extend.mq_rabbit import IdempotentConsumer, connect, declare, drain, publish

AMQP_URL = os.environ.get("AMQP_URL")
pytestmark = pytest.mark.skipif(not AMQP_URL, reason="未设置 AMQP_URL（例如 amqp://guest:guest@localhost/）")


async def test_idempotent_consume_and_dead_letter():
    prefix = f"t{uuid.uuid4().hex[:8]}"
    conn, ch = await connect(AMQP_URL)
    try:
        events, tasks, dead = await declare(ch, prefix)
        handled: list[dict] = []

        async def handler(body: dict) -> None:
            if body.get("fail"):
                raise ValueError("处理失败")
            handled.append(body)

        await publish(events, "task.created", {"id": 1}, message_id="m1")
        await publish(events, "task.created", {"id": 1}, message_id="m1")   # 重复投递
        await publish(events, "task.retry", {"id": 2, "fail": True}, message_id="m2")
        await publish(events, "other.created", {"id": 3}, message_id="m3")   # 路由键不匹配，不进 tasks

        consumer = IdempotentConsumer(handler)
        assert await drain(tasks, consumer, max_messages=10, timeout=1) == 3
        assert handled == [{"id": 1}] and consumer.duplicates == 1 and consumer.failed == 1

        dl = await dead.get(timeout=2, fail=False)
        assert dl is not None and dl.message_id == "m2"   # 失败的消息进了死信队列
        await dl.ack()
    finally:
        for q in (f"{prefix}.tasks", f"{prefix}.dead"):
            await ch.queue_delete(q)
        for ex in (f"{prefix}.events", f"{prefix}.dlx"):
            await ch.exchange_delete(ex)
        await conn.close()
```

**4. Kafka 概念（不动手，能讲清即可）**

| 维度 | RabbitMQ | Kafka |
| --- | --- | --- |
| 模型 | 消息代理：交换机按规则把消息推给队列，消费确认后删除 | 分布式日志：消息追加写入主题（topic）的分区（partition），按保留时间保存，消费者自己记录读到的位移（offset） |
| 擅长 | 业务任务分发、灵活路由、延迟 / 死信、单条确认 | 高吞吐的事件流、日志收集、多个系统重复消费同一份数据、按时间回放 |
| 顺序 | 单个队列 + 单个消费者时有序 | **同一分区内有序**；需要按订单有序时，用订单号做消息 key，保证进同一分区 |
| 扩展消费 | 同一队列多个消费者轮流取 | 消费组（consumer group）：组内每个分区只给一个消费者，分区数决定最大并行度 |
| 重复 | 有，需要幂等 | 有（位移提交前崩溃会重读），需要幂等 |

- 说明：选型的一句话：**业务任务、需要复杂路由和单条确认用 RabbitMQ；海量事件流、需要多方订阅和回放用 Kafka；量小、已经有 Redis 时用 Redis Streams 就够了**（第 3 周做过）。

### 易错点 / 面试常问

- 易错点：`declare_queue` 时参数（比如死信设置）和已经存在的队列不一致会报错 `PRECONDITION_FAILED`；改参数要先删掉旧队列（生产环境用新名字迁移）。
- 易错点：自动确认（`no_ack=True`）模式下，消息一发给消费者就被删除，处理到一半崩溃就丢了；业务消息都用手动确认。
- 易错点：失败消息无限 `requeue=True` 会形成“毒消息”死循环，一直占着消费者；失败要么进死信队列，要么记录重试次数，超过上限再进死信。
- 面试常问：如何保证消息不丢（持久化 + 发布确认 + 处理后再 ack）；如何处理重复消费（幂等：消息 ID / 业务唯一键去重）；如何处理消息积压（先扩容消费者、排查慢的原因，必要时临时转储）；“写数据库 + 发消息”如何保证一致（发件箱模式，第 9 周的实验 B）。

### 要建的文件

- `backend-extend/` 初始化（`uv init`）、`docker-compose.dev.yml`、`mq_rabbit.py`、`tests/test_rabbit.py`
- `notes/mq.md`：上面的对比表 + 4 个面试题用自己的话回答

### 运行方式

```bash
cd backend-extend && docker compose -f docker-compose.dev.yml up -d
uv add aio-pika redis pymysql grpcio && uv add --dev pytest pytest-asyncio grpcio-tools
AMQP_URL=amqp://guest:guest@localhost/ uv run pytest tests/test_rabbit.py -v
```

打开 http://localhost:15672 ，在 Queues 页面看队列和消息数的变化（测试结束会删除自己建的队列）。

### 完成标准

- 测试通过；能在管理界面里找到交换机、队列、绑定；能脱稿回答上面 4 个面试题。

---

## 第 2 天：微服务概念 + gRPC

### 学习内容

**1. 微服务讲到什么程度**

- 说明：微服务是把一个大系统按业务拆成多个可以独立部署的小服务。好处是团队可以独立开发和发布、按需扩容；代价是网络调用会失败、数据分散带来一致性问题、排查要靠链路追踪、部署运维更复杂。**面试的重点是“什么时候不该拆”**：团队小、业务还在变的时候，单体（内部分好模块）更合适。
- 要能说清的词：服务发现和注册（如 Consul、etcd、Nacos）、API 网关（统一鉴权、限流、路由）、配置中心、熔断和降级、超时和重试、链路追踪（如 OpenTelemetry）、分布式事务（发件箱、Saga 补偿）。
- 案例：你的三个项目其实已经是“多个服务”：任务看板（Go）、RAG（Python）、Agent（Python），各自部署，前面用 Nginx 统一入口——这就是一个最小的“网关 + 多服务”结构，第 7 天可以再用 Go 写一个真正的网关。

**2. gRPC 是什么**

- 说明：gRPC 是 Google 开源的远程调用框架：先用 **Protocol Buffers（proto 文件）** 定义接口和消息，再用工具生成各语言的客户端和服务端代码。基于 HTTP/2，二进制编码，比 JSON 更小更快，**接口有强类型约束**，支持流式调用。常用于服务之间的内部调用；对浏览器和第三方一般还是提供 REST / JSON。
- 四种调用方式：一元（一次请求一次响应）、服务端流（一次请求、多次响应，像 SSE）、客户端流、双向流。
- 错误用状态码表示（`INVALID_ARGUMENT`、`NOT_FOUND`、`DEADLINE_EXCEEDED`、`UNAVAILABLE`……），客户端调用要设**超时（deadline）**。
- 案例：把“创建 Agent 任务 + 订阅任务事件”定义成 gRPC 接口：

```protobuf
// 文件：backend-extend/proto/backend_extend/grpcgen/run.proto
syntax = "proto3";

package agentrun.v1;

service RunService {
  rpc CreateRun(CreateRunRequest) returns (Run);              // 一元调用
  rpc WatchRun(WatchRunRequest) returns (stream RunEvent);    // 服务端流
}

message CreateRunRequest {
  string question = 1;
  string idempotency_key = 2;   // 同一个 key 重复调用只创建一次
}

message Run {
  string id = 1;
  string status = 2;
  string question = 3;
}

message WatchRunRequest {
  string run_id = 1;
  int64 after_seq = 2;          // 断线重连时从这个序号之后继续，和 SSE 的 Last-Event-ID 一样
}

message RunEvent {
  int64 seq = 1;
  string type = 2;
  string data = 3;
}
```

- 说明：字段后面的数字是**字段编号**，编码时用它而不是字段名。所以已经上线的字段**不能改编号、不能复用删掉的编号**；新增字段用新编号，老客户端会忽略不认识的字段——这就是 proto 的向后兼容。

生成代码的脚本（proto 文件放在 `proto/backend_extend/grpcgen/` 下，是为了让生成的代码里的导入路径是 `from backend_extend.grpcgen import run_pb2`，可以直接在包里用）：

```python
# 文件：backend-extend/scripts/gen_proto.py
"""生成 gRPC 代码：uv run python scripts/gen_proto.py"""
import pathlib
import sys

from grpc_tools import protoc

ROOT = pathlib.Path(__file__).resolve().parents[1]


def main() -> None:
    proto = ROOT / "proto/backend_extend/grpcgen/run.proto"
    out = ROOT / "src"
    code = protoc.main([
        "grpc_tools.protoc", f"-I{ROOT / 'proto'}",
        f"--python_out={out}", f"--grpc_python_out={out}", f"--pyi_out={out}", str(proto)])
    if code != 0:
        sys.exit(code)
    print("生成完成：src/backend_extend/grpcgen/run_pb2*.py")


if __name__ == "__main__":
    main()
```

```python
# 文件：backend-extend/src/backend_extend/grpcgen/__init__.py
```

```python
# 文件：backend-extend/src/backend_extend/grpc_service.py
"""gRPC 服务端（grpc.aio 异步版）。先运行 scripts/gen_proto.py 生成代码。"""
import asyncio
import itertools

import grpc

from backend_extend.grpcgen import run_pb2, run_pb2_grpc


class RunService(run_pb2_grpc.RunServiceServicer):
    def __init__(self) -> None:
        self.runs: dict[str, run_pb2.Run] = {}
        self.events: dict[str, list[run_pb2.RunEvent]] = {}
        self.by_key: dict[str, str] = {}
        self._ids = itertools.count(1)

    async def CreateRun(self, request: run_pb2.CreateRunRequest, context: grpc.aio.ServicerContext) -> run_pb2.Run:
        if not request.question.strip():
            await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "question 不能为空")
        if request.idempotency_key and request.idempotency_key in self.by_key:
            return self.runs[self.by_key[request.idempotency_key]]
        run_id = f"r{next(self._ids)}"
        run = run_pb2.Run(id=run_id, status="queued", question=request.question)
        self.runs[run_id] = run
        if request.idempotency_key:
            self.by_key[request.idempotency_key] = run_id
        # 演示：预先放好几条事件；真实项目里由后台执行器写入
        self.events[run_id] = [run_pb2.RunEvent(seq=i, type=t, data=d) for i, (t, d) in enumerate(
            [("run_started", ""), ("node_finished", "proposer"), ("run_completed", "结论")], start=1)]
        return run

    async def WatchRun(self, request: run_pb2.WatchRunRequest, context: grpc.aio.ServicerContext):
        if request.run_id not in self.runs:
            await context.abort(grpc.StatusCode.NOT_FOUND, f"任务 {request.run_id} 不存在")
        for ev in self.events[request.run_id]:
            if ev.seq > request.after_seq:
                yield ev
                await asyncio.sleep(0)   # 让出事件循环；真实场景是等待新事件


async def start_server(address: str = "127.0.0.1:0") -> tuple[grpc.aio.Server, int]:
    server = grpc.aio.server()
    run_pb2_grpc.add_RunServiceServicer_to_server(RunService(), server)
    port = server.add_insecure_port(address)   # 生产环境用 TLS：add_secure_port
    await server.start()
    return server, port


async def main() -> None:  # 不自动运行：常驻服务
    server, port = await start_server("0.0.0.0:50051")
    print(f"gRPC 服务已启动：{port}")
    await server.wait_for_termination()


if __name__ == "__main__":
    asyncio.run(main())
```

```python
# 文件：backend-extend/tests/conftest.py
import pathlib
import subprocess
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]

# 生成的代码不提交（加入 .gitignore），测试前自动生成一次
if not (ROOT / "src/backend_extend/grpcgen/run_pb2.py").exists():
    subprocess.run([sys.executable, str(ROOT / "scripts/gen_proto.py")], check=True)
```

```python
# 文件：backend-extend/tests/test_grpc.py
import grpc
import pytest

from backend_extend.grpc_service import start_server
from backend_extend.grpcgen import run_pb2, run_pb2_grpc


async def test_unary_stream_and_errors():
    server, port = await start_server()
    try:
        async with grpc.aio.insecure_channel(f"127.0.0.1:{port}") as channel:
            stub = run_pb2_grpc.RunServiceStub(channel)
            run = await stub.CreateRun(run_pb2.CreateRunRequest(question="选技术栈", idempotency_key="k1"), timeout=3)
            again = await stub.CreateRun(run_pb2.CreateRunRequest(question="选技术栈", idempotency_key="k1"), timeout=3)
            assert run.id == again.id and run.status == "queued"

            events = [ev async for ev in stub.WatchRun(run_pb2.WatchRunRequest(run_id=run.id, after_seq=1), timeout=3)]
            assert [e.seq for e in events] == [2, 3] and events[-1].type == "run_completed"

            with pytest.raises(grpc.aio.AioRpcError) as e:
                await stub.CreateRun(run_pb2.CreateRunRequest(question="  "), timeout=3)
            assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT

            with pytest.raises(grpc.aio.AioRpcError) as e:
                [ev async for ev in stub.WatchRun(run_pb2.WatchRunRequest(run_id="nope"), timeout=3)]
            assert e.value.code() == grpc.StatusCode.NOT_FOUND
    finally:
        await server.stop(grace=None)
```

- Go 版（了解）：Go 里用 `protoc` + `protoc-gen-go` + `protoc-gen-go-grpc` 生成代码，服务端实现生成的接口即可。同一个 proto 文件，Go 服务和 Python 服务可以互相调用——这是 gRPC 跨语言的价值。命令和插件版本以官方文档为准：https://grpc.io/docs/languages/go/quickstart/ 。

### 易错点 / 面试常问

- 易错点：生成的 `*_pb2.py` 不要手改，改 proto 后重新生成；生成目录加进 `.gitignore`，或者统一提交并在 CI 里检查是否最新（团队二选一）。
- 易错点：客户端调用不设超时，下游卡住时上游的连接和协程会越积越多；每次调用都要带 `timeout`，并且一路往下传（下游的超时要比上游短）。
- 易错点：只对 `UNAVAILABLE` 这类临时错误重试，`INVALID_ARGUMENT` 重试没有意义；有副作用的接口重试要配合幂等键（本例的 `idempotency_key`）。
- 面试常问：gRPC 和 REST 的区别（proto 强类型 vs JSON、HTTP/2 vs 多为 HTTP/1.1、支持流式、浏览器支持差）；proto 怎么做向后兼容（只加字段不改编号）；微服务的优缺点和拆分时机；服务之间调用失败怎么处理（超时、重试、熔断、降级）。

### 要建的文件

- `proto/.../run.proto`、`scripts/gen_proto.py`、`grpc_service.py`、`tests/conftest.py`、`tests/test_grpc.py`；`.gitignore` 加 `src/backend_extend/grpcgen/run_pb2*`
- `notes/microservices.md`：微服务优缺点、什么时候不该拆、gRPC vs REST

### 运行方式

```bash
uv run python scripts/gen_proto.py
uv run pytest tests/test_grpc.py -v
```

### 完成标准

- 测试通过；能解释 proto 字段编号的作用和流式调用的用途。

---
## 第 3 天：Redis 面试专题——锁、缓存三大问题、限流

### 学习内容

**1. 五种基本数据结构和典型用途（先背下这张表）**

| 类型 | 典型用途 | 常用命令 |
| --- | --- | --- |
| String | 缓存对象（JSON）、计数器、分布式锁 | `SET k v NX PX 3000`、`INCR`、`GET` |
| Hash | 存一个对象的多个字段（用户资料） | `HSET`、`HGETALL` |
| List | 简单队列、最新 N 条 | `LPUSH`、`RPOP`、`LRANGE` |
| Set | 去重、共同好友、标签 | `SADD`、`SISMEMBER`、`SINTER` |
| Sorted Set（ZSet） | 排行榜、延迟队列、滑动窗口限流 | `ZADD`、`ZRANGE`、`ZREMRANGEBYSCORE` |

另外要知道：Stream（消息队列，第 3 周用过）、Bitmap（签到）、HyperLogLog（UV 估算）。

**2. 分布式锁**

- 说明：多个进程要互斥地做一件事（比如同一时间只允许一个进程重建某个缓存）。加锁用 `SET key 随机值 NX PX 过期毫秒`：`NX` 保证只有一个人能设置成功，`PX` 保证持有者崩溃后锁会自动释放。**释放锁时必须确认是自己的锁**（比较随机值），而“比较 + 删除”要用 Lua 脚本一次完成，否则两步之间锁可能已经过期、被别人拿走，你会删掉别人的锁。

**3. 缓存的三大问题**

| 问题 | 现象 | 常用解决办法 |
| --- | --- | --- |
| 穿透 | 查**根本不存在**的数据，缓存里永远没有，每次都打到数据库（常见于恶意请求） | 缓存空值（短过期）；参数校验；布隆过滤器 |
| 击穿 | 一个**热点 key** 过期的瞬间，大量请求同时去查数据库 | 互斥锁：只让一个请求重建缓存，其他请求稍等再读；热点数据逻辑过期、后台刷新 |
| 雪崩 | **大量 key 同时过期**或 Redis 整体不可用，请求全部压到数据库 | 过期时间加随机值；Redis 高可用（主从 + 哨兵 / 集群）；限流降级 |

**4. 限流**

- 说明：固定窗口（`INCR` + `EXPIRE`）实现最简单，但窗口边界可能放过两倍流量；滑动窗口用 ZSet 记录每次请求的时间戳，统计最近 N 秒内的数量，更平滑；令牌桶允许一定突发（第 3 周 Go 项目里用过）。

**5. 代码：把上面三件事写成可测试的函数**

```python
# 文件：backend-extend/src/backend_extend/redis_patterns.py
import asyncio
import json
import random
import secrets
import time
from collections.abc import Awaitable, Callable
from typing import Any

from redis.asyncio import Redis

# 比较 + 删除必须原子执行：只有值等于自己的令牌才删除
RELEASE_LUA = """
if redis.call('get', KEYS[1]) == ARGV[1] then
  return redis.call('del', KEYS[1])
else
  return 0
end
"""
NULL = "__null__"   # 空值标记：防止缓存穿透


async def acquire_lock(r: Redis, key: str, ttl_ms: int) -> str | None:
    token = secrets.token_hex(8)
    return token if await r.set(key, token, nx=True, px=ttl_ms) else None


async def release_lock(r: Redis, key: str, token: str) -> bool:
    return await r.eval(RELEASE_LUA, 1, key, token) == 1


async def get_or_load(r: Redis, key: str, loader: Callable[[], Awaitable[Any | None]], ttl: int = 300,
                      null_ttl: int = 30, jitter: float = 0.1, wait_s: float = 2.0) -> Any | None:
    """旁路缓存：先读缓存，没有再查库并回填。同时处理穿透、击穿、雪崩。"""
    cached = await r.get(key)
    if cached is not None:
        return None if cached == NULL else json.loads(cached)

    lock_key = f"lock:{key}"
    token = await acquire_lock(r, lock_key, ttl_ms=5000)
    if token is None:                       # 击穿：别人正在重建，等一会儿再读缓存
        deadline = time.monotonic() + wait_s
        while time.monotonic() < deadline:
            await asyncio.sleep(0.05)
            cached = await r.get(key)
            if cached is not None:
                return None if cached == NULL else json.loads(cached)
        # 等太久就自己查（兜底），不要让请求一直卡住
    try:
        data = await loader()
        if data is None:
            await r.set(key, NULL, ex=null_ttl)                       # 穿透：空值也缓存，但过期短
        else:
            await r.set(key, json.dumps(data, ensure_ascii=False),
                        ex=int(ttl * (1 + random.uniform(0, jitter))))  # 雪崩：过期时间加随机
        return data
    finally:
        if token:
            await release_lock(r, lock_key, token)


async def allow_request(r: Redis, key: str, limit: int, window_s: float) -> bool:
    """滑动窗口限流：最近 window_s 秒内最多 limit 次。"""
    now = time.time()
    member = f"{now}-{secrets.token_hex(4)}"
    async with r.pipeline(transaction=True) as pipe:      # MULTI / EXEC：几条命令一起执行
        pipe.zremrangebyscore(key, 0, now - window_s)     # 删掉窗口外的记录
        pipe.zadd(key, {member: now})
        pipe.zcard(key)
        pipe.expire(key, int(window_s) + 1)
        _, _, count, _ = await pipe.execute()
    if count > limit:
        await r.zrem(key, member)                         # 被拒绝的请求不占名额
        return False
    return True
```

```python
# 文件：backend-extend/tests/test_redis_patterns.py
import asyncio
import os
import uuid

import pytest
from redis.asyncio import Redis

from backend_extend.redis_patterns import acquire_lock, allow_request, get_or_load, release_lock

REDIS_URL = os.environ.get("REDIS_URL")
pytestmark = pytest.mark.skipif(not REDIS_URL, reason="未设置 REDIS_URL（例如 redis://localhost:6379/0）")


@pytest.fixture
async def r():
    client = Redis.from_url(REDIS_URL, decode_responses=True)
    prefix = f"t:{uuid.uuid4().hex[:8]}"
    yield client, prefix
    keys = [k async for k in client.scan_iter(f"*{prefix}*")]
    if keys:
        await client.delete(*keys)
    await client.aclose()


async def test_lock_only_owner_can_release(r):
    client, p = r
    t1 = await acquire_lock(client, f"{p}:lock", 2000)
    assert t1 and await acquire_lock(client, f"{p}:lock", 2000) is None
    assert not await release_lock(client, f"{p}:lock", "别人的令牌")
    assert await release_lock(client, f"{p}:lock", t1)
    assert await acquire_lock(client, f"{p}:lock", 2000)


async def test_breakdown_only_one_loader(r):
    client, p = r
    calls = 0

    async def loader():
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.2)      # 模拟慢查询
        return {"id": 1, "name": "看板"}

    results = await asyncio.gather(*[get_or_load(client, f"{p}:board:1", loader) for _ in range(20)])
    assert calls == 1 and all(x == {"id": 1, "name": "看板"} for x in results)
    assert 300 <= await client.ttl(f"{p}:board:1") <= 330      # 过期时间带了随机值


async def test_penetration_caches_null(r):
    client, p = r
    calls = 0

    async def loader():
        nonlocal calls
        calls += 1
        return None

    assert await get_or_load(client, f"{p}:board:404", loader) is None
    assert await get_or_load(client, f"{p}:board:404", loader) is None
    assert calls == 1


async def test_sliding_window(r):
    client, p = r
    got = [await allow_request(client, f"{p}:rl:u1", limit=3, window_s=1) for _ in range(5)]
    assert got == [True, True, True, False, False]
    await asyncio.sleep(1.1)
    assert await allow_request(client, f"{p}:rl:u1", limit=3, window_s=1)
```

### 易错点 / 面试常问

- 易错点：加锁写成 `SETNX` + `EXPIRE` 两条命令不是原子的，中间崩溃就会留下永不过期的锁；用一条 `SET ... NX PX`。
- 易错点：锁的过期时间比业务执行时间短，锁会提前释放，两个进程同时执行；要么估算足够长的时间，要么持有期间定期续期（Java 的 Redisson 叫“看门狗”）。
- 易错点：生产环境不要用 `KEYS *`（会阻塞 Redis），用 `SCAN` 分批遍历；大 key（一个 Hash 几百万字段）删除和读取都会卡，要拆分。
- 面试常问：
  - 缓存和数据库怎么保持一致（先更新数据库再删缓存；删除失败时重试或用消息队列 / 订阅 binlog 补偿；第 3 周讲过为什么是删而不是更新）。
  - Redis 为什么快（内存操作、单线程执行命令避免锁竞争、I/O 多路复用）；Redis 6 以后的多线程只用在网络读写上。
  - 持久化 RDB 和 AOF 的区别（快照 vs 追加日志，恢复速度和数据完整性的取舍）。
  - 过期删除策略（惰性删除 + 定期删除）和内存淘汰策略（如 `allkeys-lru`）。
  - Redis 分布式锁在主从切换时可能失效（主节点没同步就宕机），强一致需求用数据库行锁或 etcd / ZooKeeper；Redlock 有争议，了解即可。

### 要建的文件

- `redis_patterns.py`、`tests/test_redis_patterns.py`；`notes/redis.md`：数据结构表 + 三大问题表 + 5 个面试题的回答

### 运行方式

```bash
REDIS_URL=redis://localhost:6379/0 uv run pytest tests/test_redis_patterns.py -v
```

### 完成标准

- 测试通过（重点看 `calls == 1`：20 个并发请求只查了一次“数据库”）；能脱稿讲清缓存三大问题。

---

## 第 4 天：MySQL 面试专题——索引、EXPLAIN、事务和隔离级别

### 学习内容

**1. 索引**

- 说明：InnoDB 的索引是 **B+ 树**。主键索引（聚簇索引）的叶子节点存整行数据；二级索引的叶子节点存的是**主键值**，查到之后还要回主键索引取整行，这叫**回表**。如果要查的列都在索引里，就不用回表，叫**覆盖索引**（`EXPLAIN` 的 Extra 显示 `Using index`）。
- 联合索引 `(user_id, status, created_at)` 按“先 user_id、再 status、再 created_at”排序，所以只有从最左边的列开始用才有效——**最左前缀原则**。只按 `status` 查用不上这个索引。
- 索引失效的常见情况：不满足最左前缀；对索引列用函数或计算（`WHERE user_id + 0 = 5`、`DATE(created_at) = ...`）；隐式类型转换（字符串列和数字比较）；`LIKE '%abc'` 前面带通配符；`OR` 连接了没有索引的列。

**2. EXPLAIN 看什么**

| 列 | 看什么 |
| --- | --- |
| `type` | 访问方式，从好到差大致是 `const` > `eq_ref` > `ref` > `range` > `index`（扫整个索引）> `ALL`（全表扫描） |
| `key` | 实际用了哪个索引，`NULL` 表示没用 |
| `rows` | 估计要扫描的行数 |
| `Extra` | `Using index`（覆盖索引）、`Using filesort`（额外排序，要注意）、`Using temporary`（用了临时表） |

**3. 事务和隔离级别**

- 说明：事务的 ACID：原子性（全做或全不做）、一致性、隔离性（并发事务互不干扰的程度）、持久性（提交后不丢）。
- 并发问题：**脏读**（读到别人没提交的数据）、**不可重复读**（同一事务里两次读同一行，结果不同，因为别人提交了修改）、**幻读**（两次按条件查询，行数不同，因为别人插入了新行）。

| 隔离级别 | 脏读 | 不可重复读 | 幻读 | 说明 |
| --- | --- | --- | --- | --- |
| 读未提交 READ UNCOMMITTED | 可能 | 可能 | 可能 | 基本不用 |
| 读已提交 READ COMMITTED | 不会 | 可能 | 可能 | PostgreSQL 默认 |
| 可重复读 REPEATABLE READ | 不会 | 不会 | 普通读不会（快照）；加锁读靠间隙锁防止 | **MySQL InnoDB 默认** |
| 串行化 SERIALIZABLE | 不会 | 不会 | 不会 | 性能最差 |

- 说明：**MVCC**（多版本并发控制）：每行保留多个版本，普通 `SELECT` 读的是事务开始时（可重复读）或每条语句开始时（读已提交）的快照，读不加锁、不阻塞写。`SELECT ... FOR UPDATE` 是“当前读”，读最新数据并加锁。

**4. 代码：用两个连接亲眼看到隔离级别的区别**

```python
# 文件：backend-extend/src/backend_extend/mysql_lab.py
"""MySQL 实验：索引 / EXPLAIN、隔离级别、原子扣减、转账。连接串：mysql://app:app@127.0.0.1:3306/learn"""
import random
from urllib.parse import urlparse

import pymysql
from pymysql.cursors import DictCursor


def connect(url: str) -> pymysql.Connection:
    u = urlparse(url)
    return pymysql.connect(host=u.hostname, port=u.port or 3306, user=u.username, password=u.password,
                           database=u.path.lstrip("/"), charset="utf8mb4", cursorclass=DictCursor,
                           autocommit=False)


def setup_orders(conn: pymysql.Connection, n: int = 3000) -> None:
    with conn.cursor() as cur:
        cur.execute("DROP TABLE IF EXISTS lab_orders")
        cur.execute("""
            CREATE TABLE lab_orders (
              id BIGINT PRIMARY KEY AUTO_INCREMENT,
              user_id INT NOT NULL,
              status VARCHAR(16) NOT NULL,
              amount INT NOT NULL,
              created_at DATETIME NOT NULL,
              KEY idx_user_status_time (user_id, status, created_at)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""")
        rnd = random.Random(42)
        rows = [(rnd.randint(1, 300), rnd.choice(["paid", "unpaid", "refund"]), rnd.randint(1, 999),
                 f"2026-{rnd.randint(1, 12):02d}-{rnd.randint(1, 28):02d} 10:00:00") for _ in range(n)]
        cur.executemany("INSERT INTO lab_orders (user_id, status, amount, created_at) VALUES (%s, %s, %s, %s)", rows)
        cur.execute("ANALYZE TABLE lab_orders")   # 更新统计信息，让优化器的估算更准
        cur.fetchall()
    conn.commit()


def explain(conn: pymysql.Connection, sql: str, args: tuple = ()) -> dict:
    with conn.cursor() as cur:
        cur.execute("EXPLAIN " + sql, args)
        row = cur.fetchone()
    return {"type": row["type"], "key": row["key"], "rows": row["rows"], "Extra": row["Extra"] or ""}


def setup_accounts(conn: pymysql.Connection, balances: dict[int, int]) -> None:
    with conn.cursor() as cur:
        cur.execute("DROP TABLE IF EXISTS lab_accounts")
        cur.execute("CREATE TABLE lab_accounts (id INT PRIMARY KEY, balance INT NOT NULL) ENGINE=InnoDB")
        cur.executemany("INSERT INTO lab_accounts VALUES (%s, %s)", list(balances.items()))
    conn.commit()


def balance(conn: pymysql.Connection, account_id: int) -> int:
    with conn.cursor() as cur:
        cur.execute("SELECT balance FROM lab_accounts WHERE id = %s", (account_id,))
        return cur.fetchone()["balance"]


def read_twice(a: pymysql.Connection, b: pymysql.Connection, isolation: str) -> tuple[int, int]:
    """事务 A 读两次余额，中间事务 B 修改并提交。返回 A 两次读到的值。"""
    with a.cursor() as cur:
        cur.execute(f"SET SESSION TRANSACTION ISOLATION LEVEL {isolation}")
    a.begin()
    first = balance(a, 1)
    with b.cursor() as cur:
        cur.execute("UPDATE lab_accounts SET balance = balance + 100 WHERE id = 1")
    b.commit()
    second = balance(a, 1)
    a.commit()
    return first, second


def deduct(conn: pymysql.Connection, account_id: int, amount: int) -> bool:
    """原子扣减：条件写在 UPDATE 里，并发也不会扣成负数。看影响行数判断是否成功。"""
    with conn.cursor() as cur:
        n = cur.execute("UPDATE lab_accounts SET balance = balance - %s WHERE id = %s AND balance >= %s",
                        (amount, account_id, amount))
    conn.commit()
    return n == 1


def transfer(conn: pymysql.Connection, src: int, dst: int, amount: int) -> None:
    """转账：一个事务里扣一方、加一方。按 id 顺序加锁，避免两笔反向转账互相等待（死锁）。"""
    try:
        with conn.cursor() as cur:
            cur.execute("SELECT id, balance FROM lab_accounts WHERE id IN (%s, %s) ORDER BY id FOR UPDATE", (src, dst))
            rows = {r["id"]: r["balance"] for r in cur.fetchall()}
            if rows.get(src, 0) < amount:
                raise ValueError("余额不足")
            cur.execute("UPDATE lab_accounts SET balance = balance - %s WHERE id = %s", (amount, src))
            cur.execute("UPDATE lab_accounts SET balance = balance + %s WHERE id = %s", (amount, dst))
        conn.commit()
    except Exception:
        conn.rollback()
        raise

```

```python
# 文件：backend-extend/tests/test_mysql_lab.py
import os
import threading

import pytest

from backend_extend.mysql_lab import (balance, connect, deduct, explain, read_twice, setup_accounts,
                                      setup_orders, transfer)

MYSQL_URL = os.environ.get("MYSQL_URL")
pytestmark = pytest.mark.skipif(not MYSQL_URL, reason="未设置 MYSQL_URL（例如 mysql://app:app@127.0.0.1:3306/learn）")


@pytest.fixture
def conn():
    c = connect(MYSQL_URL)
    yield c
    c.close()


def test_index_usage(conn):
    setup_orders(conn)
    assert explain(conn, "SELECT * FROM lab_orders WHERE user_id = %s", (5,))["key"] == "idx_user_status_time"
    # 最左前缀：跳过 user_id 只按 status 查，用不上联合索引
    assert explain(conn, "SELECT * FROM lab_orders WHERE status = %s", ("paid",))["type"] == "ALL"
    # 对索引列做计算，索引失效
    assert explain(conn, "SELECT * FROM lab_orders WHERE user_id + 0 = %s", (5,))["type"] == "ALL"
    # 覆盖索引：要的列都在索引里，不用回表
    covered = explain(conn, "SELECT user_id, status, created_at FROM lab_orders WHERE user_id = %s", (5,))
    assert "Using index" in covered["Extra"]
    # 等值条件 + 按索引下一列排序：不需要额外排序
    ordered = explain(conn, "SELECT * FROM lab_orders WHERE user_id = %s AND status = %s ORDER BY created_at", (5, "paid"))
    assert ordered["key"] == "idx_user_status_time" and "filesort" not in ordered["Extra"]


def test_isolation_levels():
    a, b = connect(MYSQL_URL), connect(MYSQL_URL)
    try:
        setup_accounts(a, {1: 100})
        first, second = read_twice(a, b, "READ COMMITTED")
        assert second == first + 100          # 不可重复读：同一事务里两次结果不同
        first, second = read_twice(a, b, "REPEATABLE READ")
        assert second == first                # 可重复读：读的是事务开始时的快照
        assert balance(a, 1) == first + 100   # 新事务才看得到最新值
    finally:
        a.close()
        b.close()


def test_atomic_deduct_under_concurrency(conn):
    setup_accounts(conn, {1: 50})
    results: list[bool] = []

    def worker():
        c = connect(MYSQL_URL)
        try:
            results.append(deduct(c, 1, 10))
        finally:
            c.close()

    threads = [threading.Thread(target=worker) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert results.count(True) == 5 and balance(conn, 1) == 0   # 只有 5 次成功，不会扣成负数


def test_transfer_rolls_back(conn):
    setup_accounts(conn, {1: 100, 2: 0})
    transfer(conn, 1, 2, 30)
    with pytest.raises(ValueError):
        transfer(conn, 2, 1, 999)
    assert (balance(conn, 1), balance(conn, 2)) == (70, 30)
    conn.commit()
```

### 易错点 / 面试常问

- 易错点：PyMySQL 默认不自动提交，忘了 `commit()` 数据就不会保存；而在“可重复读”下，长时间不结束的事务会一直读旧快照，看起来像“数据没更新”。
- 易错点：“先查余额、再在代码里减、再写回”在并发下会丢失更新；用 `UPDATE ... SET balance = balance - ? WHERE balance >= ?` 这样的原子语句，或者 `SELECT ... FOR UPDATE` 加锁。
- 易错点：索引不是越多越好：每个索引都会让写入变慢、占空间；区分度很低的列（如性别）单独建索引意义不大。
- 面试常问：
  - 为什么 InnoDB 用 B+ 树（树矮、磁盘读取次数少；叶子节点有序相连，适合范围查询）。
  - 聚簇索引和二级索引、回表、覆盖索引、最左前缀、索引下推（了解）。
  - 一条慢 SQL 怎么排查（慢查询日志 → `EXPLAIN` → 看 `type` / `key` / `rows` / `Extra` → 调整索引或改写 SQL）。
  - MySQL 默认隔离级别是什么，怎么用 MVCC 实现可重复读；什么是间隙锁。
  - 死锁怎么产生、怎么避免（固定加锁顺序、事务尽量短）；InnoDB 会自动检测死锁并回滚其中一个事务，应用要能重试。
  - 和 PostgreSQL 的差异（第 2 周的速查表：默认隔离级别、自增写法、`RETURNING`、JSON 支持等）。

### 要建的文件

- `mysql_lab.py`、`tests/test_mysql_lab.py`；`notes/mysql.md`：索引失效清单、EXPLAIN 列表、隔离级别表、6 个面试题的回答

### 运行方式

```bash
MYSQL_URL=mysql://app:app@127.0.0.1:3306/learn uv run pytest tests/test_mysql_lab.py -v
docker compose -f docker-compose.dev.yml exec mysql mysql -uapp -papp learn   # 手动试 EXPLAIN
```

### 完成标准

- 测试全部通过；能对着 `EXPLAIN` 的输出解释“为什么这个查询没用上索引”。

---
## 第 5 天：把三个项目部署到一台服务器

### 学习内容

**1. 服务器基础（Linux 必会命令）**

- 说明：招聘里的“熟悉 Linux”在应用开发岗位上，通常指能自己登录服务器、部署服务、看日志、排查端口和磁盘问题。下面这些命令每个都要亲手用过。

| 场景 | 命令 |
| --- | --- |
| 登录和密钥 | `ssh-keygen -t ed25519`、`ssh-copy-id user@服务器`，之后在 `/etc/ssh/sshd_config` 关闭密码登录 |
| 用户和权限 | `adduser deploy`、`usermod -aG docker deploy`（用普通用户部署，不用 root）、`chmod`、`chown` |
| 防火墙 | `ufw allow 22,80,443/tcp && ufw enable`（只开必须的端口；数据库、Redis 不对外开放） |
| 进程和端口 | `ss -lntp`（谁在监听哪个端口）、`top` / `htop`、`ps aux | grep uvicorn` |
| 磁盘和内存 | `df -h`、`du -sh *`、`free -h`；Docker 占满磁盘时 `docker system df`、`docker image prune` |
| 日志 | `docker compose logs -f --tail=100 服务名`、`journalctl -u docker` |
| 文件 | `tail -f`、`less`、`grep -rn 关键词 目录`、`scp` / `rsync` |

**2. 部署结构：一台服务器、一个 Compose、一个 Nginx**

- 说明：三个项目各自已经有 Dockerfile（第 3、6、13 周）。服务器上建一个 `deploy/` 目录，用一个总的 Compose 把它们和公共依赖（PostgreSQL、Redis）一起启动；Nginx 是唯一对外的入口，按域名转发到不同服务。每个前端用自己的子域名，这样前端代码里的 `/api` 不用改。

```text
浏览器 ──HTTPS──> Nginx(:443)
                   ├─ board.example.com  → 看板前端静态文件 + /api → task-board:8080
                   ├─ rag.example.com    → 问答前端静态文件 + /api → rag-qa:8000
                   └─ agent.example.com  → agent-mini:8100（自带页面）
内部网络：postgres、redis（不映射端口到公网）
```

```yaml
# 文件：deploy/docker-compose.yml
services:
  postgres:
    image: postgres:17
    environment:
      POSTGRES_USER: ${PG_USER}
      POSTGRES_PASSWORD: ${PG_PASSWORD}
      POSTGRES_DB: taskboard
    volumes: [pg-data:/var/lib/postgresql/data]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${PG_USER}"]
      interval: 5s
      retries: 10
    restart: unless-stopped
  redis:
    image: redis:7
    command: ["redis-server", "--appendonly", "yes"]
    volumes: [redis-data:/data]
    restart: unless-stopped

  task-board:
    image: ${REGISTRY}/task-board:${TAG:-latest}
    env_file: ./env/task-board.env
    depends_on:
      postgres: { condition: service_healthy }
      redis: { condition: service_started }
    restart: unless-stopped
  task-board-worker:
    image: ${REGISTRY}/task-board:${TAG:-latest}
    command: ["/app/worker"]
    env_file: ./env/task-board.env
    depends_on: [task-board]
    restart: unless-stopped

  rag-qa:
    image: ${REGISTRY}/rag-qa:${TAG:-latest}
    env_file: ./env/rag-qa.env
    volumes: [rag-data:/app/data]
    depends_on: [redis]
    restart: unless-stopped

  agent-mini:
    image: ${REGISTRY}/agent-mini:${TAG:-latest}
    env_file: ./env/agent-mini.env
    volumes: [agent-data:/app/data]
    restart: unless-stopped

  nginx:
    image: nginx:1.27
    ports: ["80:80", "443:443"]
    volumes:
      - ./nginx/conf.d:/etc/nginx/conf.d:ro
      - ./www:/usr/share/nginx/html:ro           # 两个前端的构建产物：www/board、www/rag
      - /etc/letsencrypt:/etc/letsencrypt:ro     # HTTPS 证书
    depends_on: [task-board, rag-qa, agent-mini]
    restart: unless-stopped

volumes:
  pg-data:
  redis-data:
  rag-data:
  agent-data:
```

```nginx
# 文件：deploy/nginx/conf.d/apps.conf（只列出看板和 Agent，问答站点照着看板写）
server {
    listen 80;
    server_name board.example.com rag.example.com agent.example.com;
    return 301 https://$host$request_uri;          # 统一跳转 HTTPS
}

server {
    listen 443 ssl;
    server_name board.example.com;
    ssl_certificate     /etc/letsencrypt/live/board.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/board.example.com/privkey.pem;

    root /usr/share/nginx/html/board;
    location / {
        try_files $uri $uri/ /index.html;          # 单页应用：刷新子路由不 404
    }
    location /assets/ {
        expires 30d;                               # 文件名带哈希，可以长缓存
    }
    location /api/ {
        proxy_pass http://task-board:8080;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

server {
    listen 443 ssl;
    server_name agent.example.com;
    ssl_certificate     /etc/letsencrypt/live/agent.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/agent.example.com/privkey.pem;

    location / {
        proxy_pass http://agent-mini:8100;
        proxy_http_version 1.1;
        proxy_set_header Connection "";
        proxy_set_header Host $host;
        proxy_buffering off;                       # SSE：不缓冲，事件实时推给浏览器
        proxy_read_timeout 1h;                     # SSE：长连接不要 60 秒就断
    }
}
```

**3. HTTPS 证书**

- 说明：用 Let's Encrypt 的免费证书，`certbot` 申请，90 天有效期，要配置自动续期。最简单的做法是先在宿主机用 `certbot certonly --standalone -d board.example.com ...` 申请（申请时要临时停掉占用 80 端口的 Nginx），证书在 `/etc/letsencrypt/live/` 下，挂进 Nginx 容器。具体命令以官方文档为准：https://certbot.eff.org/ 。没有域名时，先用 IP + HTTP 部署，并录好演示视频。

**4. 部署步骤（写进 `deploy/deploy.md`）**

```bash
# 服务器上（以 deploy 用户）
mkdir -p ~/deploy/env ~/deploy/www && cd ~/deploy
# 1. 上传 docker-compose.yml、nginx/，在 env/ 下写好三个 .env（只在服务器上，不进仓库）
# 2. 上传两个前端的构建产物到 www/board、www/rag
# 3. 拉镜像并启动
docker compose pull && docker compose up -d
# 4. 跑看板的数据库迁移（命令见第 3 周 deploy.md）
# 5. 检查
docker compose ps && docker compose logs --tail=50 agent-mini
curl -I https://board.example.com
```

### 易错点 / 面试常问

- 易错点：数据库和 Redis 不要映射端口到公网（Compose 里不写 `ports`），服务之间走内部网络。
- 易错点：`.env` 文件里有密钥，权限设为 `chmod 600`，不进仓库、不打进镜像。
- 易错点：Nginx 的 `proxy_pass` 后面带不带结尾的 `/` 行为不同：`proxy_pass http://app:8080;` 原样转发路径，`proxy_pass http://app:8080/;` 会去掉 `location` 匹配到的前缀。
- 易错点：容器重建后数据丢失——数据库、Chroma、Agent 的 SQLite 都必须挂数据卷，并定期备份（`pg_dump`、复制数据卷目录）。
- 面试常问：一次部署的完整流程；服务挂了怎么发现（健康检查、`restart: unless-stopped`、监控告警）；怎么做到发布时不中断（多实例滚动更新；单机至少做到“先拉镜像再重启”，缩短中断时间）。

### 要建的文件

- `deploy/docker-compose.yml`、`deploy/nginx/conf.d/apps.conf`、`deploy/deploy.md`（步骤、备份方法、回滚方法）

### 运行方式

- 有服务器：按上面的步骤部署，三个网址都能访问。
- 没有服务器：在本机用 `docker compose up -d` 跑起来，`/etc/hosts` 里把三个域名指向 `127.0.0.1`，先去掉 HTTPS 部分验证。

### 完成标准

- 三个项目都能通过浏览器访问；Agent 页面的流式输出是实时的（不是最后一次性出来）；`deploy.md` 写清了备份和回滚。

---

## 第 6 天：CI/CD——测试、构建镜像、自动部署

### 学习内容

- 说明：
  - **CI（持续集成）**：每次推送自动跑检查和测试，保证主分支随时可用。
  - **CD（持续部署 / 交付）**：测试通过后自动构建镜像、推到镜像仓库，再登录服务器更新服务。
  - 你的仓库在 `git.gozeros.com`。如果它是 Gitea / Forgejo，用 **Gitea Actions**（语法基本兼容 GitHub Actions，文件放 `.gitea/workflows/`，需要管理员开启并注册 Runner）；如果把作品同步到 GitHub，用 GitHub Actions（`.github/workflows/`）。先确认平台支持哪一种，具体以平台文档为准：https://docs.gitea.com/usage/actions/overview 、https://docs.github.com/actions 。
  - **密钥**（镜像仓库密码、服务器 SSH 私钥）放在平台的 Secrets 设置里，工作流里用 `${{ secrets.名字 }}` 引用，不写进文件。
- 案例：一个工作流，三个项目各自测试，`main` 分支通过后构建并部署。

```yaml
# 文件：.github/workflows/ci.yml（Gitea Actions 放 .gitea/workflows/ci.yml）
name: ci
on:
  push:
    branches: [main]
  pull_request:

jobs:
  go:
    runs-on: ubuntu-latest
    defaults: { run: { working-directory: task-board } }
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-go@v5
        with: { go-version: stable }
      - run: go vet ./... && go test ./... -race

  python:
    runs-on: ubuntu-latest
    strategy:
      matrix: { project: [rag-qa, agent-mini, backend-extend] }
    defaults: { run: { working-directory: "${{ matrix.project }}" } }
    steps:
      - uses: actions/checkout@v4
      - uses: astral-sh/setup-uv@v6          # 版本号以官方文档为准
      - run: uv sync --frozen
      - run: uv run ruff check . && uv run pytest -q   # 没有外部服务的测试会自动跳过

  web:
    runs-on: ubuntu-latest
    defaults: { run: { working-directory: web } }
    steps:
      - uses: actions/checkout@v4
      - uses: pnpm/action-setup@v4
      - uses: actions/setup-node@v4
        with: { node-version: 24, cache: pnpm, cache-dependency-path: web/pnpm-lock.yaml }
      - run: pnpm install --frozen-lockfile && pnpm lint && pnpm -r typecheck && pnpm test && pnpm -r build

  deploy:
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    needs: [go, python, web]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: 构建并推送镜像
        run: |
          echo "${{ secrets.REGISTRY_PASSWORD }}" | docker login "${{ vars.REGISTRY }}" -u "${{ secrets.REGISTRY_USER }}" --password-stdin
          for app in task-board rag-qa agent-mini; do
            docker build -t "${{ vars.REGISTRY }}/$app:${{ github.sha }}" "$app"
            docker push "${{ vars.REGISTRY }}/$app:${{ github.sha }}"
          done
      - name: 登录服务器更新
        run: |
          install -m 600 /dev/null key && echo "${{ secrets.DEPLOY_SSH_KEY }}" > key
          ssh -i key -o StrictHostKeyChecking=accept-new "${{ secrets.DEPLOY_HOST }}" \
            "cd ~/deploy && TAG=${{ github.sha }} docker compose pull && TAG=${{ github.sha }} docker compose up -d"
          rm -f key
```

- 说明：镜像标签用提交号（`github.sha`），不要只用 `latest`：出问题时可以明确地切回上一个版本。**回滚**：在服务器上 `TAG=上一个提交号 docker compose up -d`。注意：上面的 `TAG` 只在这一条命令里生效，服务器上手动 `docker compose up -d` 会回到 `latest`；更稳的做法是把当前标签写进 `~/deploy/.env`（Compose 会自动读取）。
- 说明：没有镜像仓库时，也可以在服务器上直接 `git pull && docker compose build`，但构建会占用服务器资源，适合个人项目。

### 易错点 / 面试常问

- 易错点：工作流日志是公开（或团队可见）的，不要 `echo` 密钥；平台会自动遮盖 Secrets，但拼接、编码后的值不会被遮盖。
- 易错点：CI 里用 `--frozen` / `--frozen-lockfile` 安装依赖，锁文件和项目配置不一致时直接失败，避免“我本地能跑”。
- 易错点：测试依赖外部服务（数据库、Redis、RabbitMQ）时，要么在 CI 里用 `services:` 启动它们，要么让这些测试在没有连接串时自动跳过（本计划的写法）。
- 面试常问：CI/CD 的流程；怎么回滚；数据库迁移在发布流程里放在哪一步（先执行向后兼容的迁移，再发新代码；破坏性的改动分两次发布）。

### 要建的文件

- `.github/workflows/ci.yml` 或 `.gitea/workflows/ci.yml`；`deploy/deploy.md` 补充“自动部署和回滚”

### 运行方式

- 推送一次提交，在平台的 Actions 页面看到四个任务；故意让一个测试失败，确认 `deploy` 不会执行。

### 完成标准

- 推送到 `main` 后自动测试、构建、部署；能说出回滚的具体命令。

---

## 第 7 天（可选）：用 Go 写一个鉴权网关

> 时间不够就跳过这一天（总纲第 6 节）。做了的话，简历上 Go 的部分会更有说服力：“用 Go 写了统一鉴权网关，透传 SSE”。

### 学习内容

- 说明：**API 网关**是所有请求的统一入口，负责鉴权、限流、请求 ID、日志、路由转发，后面的业务服务就不用各自实现。这里用 Go 标准库的 `httputil.ReverseProxy` 写一个最小网关放在 `agent-mini` 前面：
  - 校验 JWT（复用第 2 周的写法），把用户 ID 放进 `X-User-ID` 请求头转发给后端，并**删掉客户端自己带的同名请求头**（防伪造）；
  - 给每个请求一个 `X-Request-ID`，日志里能把网关和后端的记录串起来；
  - SSE 要能实时透传。`ReverseProxy` 对 `Content-Type: text/event-stream` 的响应会**自动立即刷新**，所以不用额外配置；测试里专门验证这一点。
- 案例：

```go
// 文件：go-gateway/main.go
package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"encoding/json"
	"errors"
	"log/slog"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/golang-jwt/jwt/v5"
)

func writeJSON(w http.ResponseWriter, code int, v any) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.WriteHeader(code)
	_ = json.NewEncoder(w).Encode(v)
}

// requestID 给每个请求一个 ID；客户端带了就沿用
func requestID(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		id := r.Header.Get("X-Request-ID")
		if id == "" {
			b := make([]byte, 8)
			_, _ = rand.Read(b)
			id = hex.EncodeToString(b)
			r.Header.Set("X-Request-ID", id)
		}
		w.Header().Set("X-Request-ID", id)
		start := time.Now()
		next.ServeHTTP(w, r)
		slog.Info("request", "id", id, "method", r.Method, "path", r.URL.Path, "cost", time.Since(start))
	})
}

// auth 校验 JWT，把用户 ID 写进 X-User-ID
func auth(secret []byte, next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		r.Header.Del("X-User-ID") // 先删掉：客户端自己带的不可信
		raw, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer ")
		if !ok {
			writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "缺少令牌"})
			return
		}
		tok, err := jwt.Parse(raw, func(*jwt.Token) (any, error) { return secret, nil },
			jwt.WithValidMethods([]string{"HS256"}), jwt.WithExpirationRequired())
		if err != nil {
			writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "令牌无效或已过期"})
			return
		}
		sub, err := tok.Claims.GetSubject()
		if err != nil || sub == "" {
			writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "令牌缺少用户"})
			return
		}
		r.Header.Set("X-User-ID", sub)
		next.ServeHTTP(w, r)
	})
}

// NewGateway：/api/* 鉴权后去掉 /api 前缀转发给 upstream；/healthz 不鉴权
func NewGateway(upstream *url.URL, secret []byte) http.Handler {
	proxy := &httputil.ReverseProxy{
		Rewrite: func(pr *httputil.ProxyRequest) {
			pr.SetURL(upstream)
			pr.SetXForwarded()
		},
		ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
			slog.Error("upstream", "err", err, "id", r.Header.Get("X-Request-ID"))
			writeJSON(w, http.StatusBadGateway, map[string]string{"error": "后端服务不可用"})
		},
	}
	mux := http.NewServeMux()
	mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
		writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
	})
	mux.Handle("/api/", http.StripPrefix("/api", auth(secret, proxy)))
	return requestID(mux)
}

func main() {
	secret := os.Getenv("JWT_SECRET")
	if len(secret) < 32 {
		slog.Error("JWT_SECRET 至少 32 个字符")
		os.Exit(1)
	}
	upstream, err := url.Parse(envOr("UPSTREAM", "http://127.0.0.1:8100"))
	if err != nil {
		slog.Error("UPSTREAM 无效", "err", err)
		os.Exit(1)
	}
	srv := &http.Server{
		Addr:              envOr("ADDR", ":8080"),
		Handler:           NewGateway(upstream, []byte(secret)),
		ReadHeaderTimeout: 5 * time.Second,
		// 不设 WriteTimeout：它会掐断 SSE 长连接
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	go func() {
		slog.Info("gateway listening", "addr", srv.Addr, "upstream", upstream.String())
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			slog.Error("listen", "err", err)
			os.Exit(1)
		}
	}()
	<-ctx.Done()
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	_ = srv.Shutdown(shutdownCtx)
}

func envOr(k, def string) string {
	if v := os.Getenv(k); v != "" {
		return v
	}
	return def
}
```

```go
// 文件：go-gateway/main_test.go
package main

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net/http"
	"net/http/httptest"
	"net/url"
	"testing"
	"time"

	"github.com/golang-jwt/jwt/v5"
)

var secret = []byte("test-secret-test-secret-test-secret!")

func token(t *testing.T, sub string, exp time.Duration) string {
	t.Helper()
	s, err := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
		"sub": sub, "exp": time.Now().Add(exp).Unix(),
	}).SignedString(secret)
	if err != nil {
		t.Fatal(err)
	}
	return s
}

func newGateway(t *testing.T, upstream http.Handler) *httptest.Server {
	t.Helper()
	up := httptest.NewServer(upstream)
	t.Cleanup(up.Close)
	u, _ := url.Parse(up.URL)
	gw := httptest.NewServer(NewGateway(u, secret))
	t.Cleanup(gw.Close)
	return gw
}

func get(t *testing.T, gw *httptest.Server, path, tok string, extra map[string]string) *http.Response {
	t.Helper()
	req, _ := http.NewRequest(http.MethodGet, gw.URL+path, nil)
	if tok != "" {
		req.Header.Set("Authorization", "Bearer "+tok)
	}
	for k, v := range extra {
		req.Header.Set(k, v)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	return resp
}

func TestAuthAndForward(t *testing.T) {
	gw := newGateway(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		writeJSON(w, http.StatusOK, map[string]string{
			"path": r.URL.Path, "user": r.Header.Get("X-User-ID"), "rid": r.Header.Get("X-Request-ID"),
		})
	}))

	cases := []struct {
		name string
		tok  string
		want int
	}{
		{"没有令牌", "", http.StatusUnauthorized},
		{"令牌过期", token(t, "u1", -time.Minute), http.StatusUnauthorized},
		{"正常", token(t, "u1", time.Hour), http.StatusOK},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			resp := get(t, gw, "/api/runs", c.tok, map[string]string{"X-User-ID": "admin"})
			defer resp.Body.Close()
			if resp.StatusCode != c.want {
				t.Fatalf("status = %d, want %d", resp.StatusCode, c.want)
			}
			if c.want != http.StatusOK {
				return
			}
			var body map[string]string
			_ = json.NewDecoder(resp.Body).Decode(&body)
			if body["path"] != "/runs" || body["user"] != "u1" || body["rid"] == "" {
				t.Fatalf("upstream saw %+v", body) // 伪造的 X-User-ID=admin 必须被替换
			}
		})
	}

	resp := get(t, gw, "/healthz", "", nil)
	resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		t.Fatalf("healthz = %d", resp.StatusCode)
	}
}

func TestSSEIsStreamedImmediately(t *testing.T) {
	release := make(chan struct{})
	gw := newGateway(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "text/event-stream")
		fmt.Fprint(w, "id: 1\ndata: first\n\n")
		w.(http.Flusher).Flush()
		<-release // 第二条事件要等测试读到第一条之后才发
		fmt.Fprint(w, "id: 2\ndata: second\n\n")
	}))
	t.Cleanup(func() { close(release) }) // 先于 Close 执行，避免上游处理函数卡住

	resp := get(t, gw, "/api/runs/r1/events", token(t, "u1", time.Hour), nil)
	defer resp.Body.Close()
	line := make(chan string, 1)
	go func() {
		l, _ := bufio.NewReader(resp.Body).ReadString('\n')
		line <- l
	}()
	select {
	case l := <-line:
		if l != "id: 1\n" {
			t.Fatalf("got %q", l)
		}
	case <-time.After(2 * time.Second):
		t.Fatal("第一条事件没有及时到达：被网关缓冲了")
	}
}

func TestUpstreamDown(t *testing.T) {
	u, _ := url.Parse("http://127.0.0.1:1") // 没有服务监听的端口
	gw := httptest.NewServer(NewGateway(u, secret))
	defer gw.Close()
	resp := get(t, gw, "/api/runs", token(t, "u1", time.Hour), nil)
	resp.Body.Close()
	if resp.StatusCode != http.StatusBadGateway {
		t.Fatalf("status = %d", resp.StatusCode)
	}
}
```

- 说明：`agent-mini` 那边读取 `X-User-ID` 作为用户（代替请求体里的 `user_id`），并且只监听内部网络——否则别人绕过网关直接访问后端，就能伪造请求头。

### 易错点 / 面试常问

- 易错点：网关后面的服务如果也对公网开放，“网关鉴权”就形同虚设；后端只在内部网络可访问。
- 易错点：`http.Server` 设置 `WriteTimeout` 会把 SSE 长连接在超时后掐断；需要超时保护时用 `ReadHeaderTimeout`，并在业务层控制。
- 易错点：`t.Cleanup` 按“后注册先执行”的顺序运行，测试里关闭 `release` 通道要先于关闭上游服务，否则 `Close` 会一直等待卡住的请求。
- 面试常问：网关负责什么、不该负责什么（鉴权、限流、路由、日志可以；业务逻辑不要放进网关）；限流放在网关还是服务里（全局限流放网关，按业务的精细限流放服务）；Go 的 `ReverseProxy` 怎么处理流式响应。

### 要建的文件

- `go-gateway/main.go`、`go-gateway/main_test.go`、`go-gateway/Dockerfile`（照第 3 周的多阶段构建）；`deploy/docker-compose.yml` 里加上网关，`agent-mini` 改成只暴露给网关

### 运行方式

```bash
cd go-gateway && go mod init go-gateway && go get github.com/golang-jwt/jwt/v5
go vet ./... && go test ./... -race -v
JWT_SECRET=$(openssl rand -hex 32) UPSTREAM=http://127.0.0.1:8100 go run .
```

### 完成标准

- 测试通过；通过网关访问 Agent 页面时流式输出依然实时。

---

<!-- week-topics:start -->

## 本周相关专题

> 本节由 `tools/gen_week_topics.py` 根据专题文件的 `week` 字段自动生成，请勿手改；想调整就修改对应专题文件的 frontmatter 后重新生成。

- **AI 编程练习（每周）**：本周至少一个功能用 AI 编程工具（Cursor / Claude Code / Codex 等）完成，按 [AI 编程：工作流](../topics/ai/ai-coding/01-workflow-rules-skills.md) 写清需求和规则，按 [验证与证据](../topics/ai/ai-coding/02-verify-evidence.md) 保存测试结果和会话记录（脱敏后），第 15 周整理成“AI 编程 + 验证方法”的应聘案例。

| 技术栈 | 课程 / 场景 / 面试题 |
| --- | --- |
| [模型侧基础（了解级）](../topics/ai/llm-serving/index.md) | 场景：[公司级大模型网关](../topics/ai/llm-serving/scenarios/multi-provider-gateway.md) |
| [数据管道基础](../topics/backend/data-pipeline/index.md) | 课程：[数据管道基础](../topics/backend/data-pipeline/01-batch-stream-reliability.md)；场景：[第三方平台订单同步](../topics/backend/data-pipeline/scenarios/third-party-sync.md)；[面试题](../topics/backend/data-pipeline/interview.md) |
| [Elasticsearch](../topics/backend/elasticsearch/index.md) | 课程：[Elasticsearch](../topics/backend/elasticsearch/01-search-basics-hybrid.md)；场景：[知识库混合检索](../topics/backend/elasticsearch/scenarios/kb-search-sync.md)；[面试题](../topics/backend/elasticsearch/interview.md) |
| [gRPC 与微服务](../topics/backend/grpc-microservices/index.md) | 课程：[gRPC 和 Protobuf](../topics/backend/grpc-microservices/01-grpc-proto.md)；课程：[服务治理](../topics/backend/grpc-microservices/02-governance.md)；场景：[单体拆分](../topics/backend/grpc-microservices/scenarios/monolith-split.md)；[面试题](../topics/backend/grpc-microservices/interview.md) |
| [Java / Spring Boot（了解级）](../topics/backend/java-spring/index.md) | 课程：[Java / Spring Boot（了解级）](../topics/backend/java-spring/01-java-spring-awareness.md)；场景：[Python Agent 调用公司已有的 Java 业务系统](../topics/backend/java-spring/scenarios/call-java-service.md)；[面试题](../topics/backend/java-spring/interview.md) |
| [MongoDB](../topics/backend/mongodb/index.md) | 课程：[MongoDB 基础](../topics/backend/mongodb/01-document-model.md)；场景：[AI 客服会话存储](../topics/backend/mongodb/scenarios/conversation-store.md)；[面试题](../topics/backend/mongodb/interview.md) |
| [消息队列（RabbitMQ / Kafka）](../topics/backend/mq/index.md) | 课程：[RabbitMQ](../topics/backend/mq/01-rabbitmq.md)；课程：[Kafka](../topics/backend/mq/02-kafka.md)；课程：[可靠性](../topics/backend/mq/03-reliability.md)；场景：[订单超时自动取消](../topics/backend/mq/scenarios/delayed-order-cancel.md)；[面试题](../topics/backend/mq/interview.md) |
| [MySQL](../topics/backend/mysql/index.md) | 课程：[事务、隔离级别、MVCC 和锁（行锁、间隙锁、死锁）](../topics/backend/mysql/02-tx-locks-mvcc.md)；课程：[运维与扩展](../topics/backend/mysql/03-ops-scale.md)；场景：[大表深分页和历史数据](../topics/backend/mysql/scenarios/deep-pagination.md)；[面试题](../topics/backend/mysql/interview.md) |
| [Node.js（了解级）](../topics/backend/nodejs/index.md) | 课程：[Node.js 后端（了解级）](../topics/backend/nodejs/01-express-nest-basics.md)；场景：[用 Node.js 给 AI 应用做 BFF](../topics/backend/nodejs/scenarios/ai-bff.md)；[面试题](../topics/backend/nodejs/interview.md) |
| [Redis](../topics/backend/redis/index.md) | 课程：[分布式锁和限流](../topics/backend/redis/03-lock-ratelimit.md)；场景：[抢购库存预扣](../topics/backend/redis/scenarios/stock-reservation.md)；[面试题](../topics/backend/redis/interview.md) |
| [网络与操作系统基础](../topics/cs/network-os/index.md) | 课程：[网络基础](../topics/cs/network-os/01-http-tcp.md)；课程：[操作系统基础](../topics/cs/network-os/02-os-basics.md)；场景：[线上连接问题排查](../topics/cs/network-os/scenarios/connection-troubleshooting.md)；[面试题](../topics/cs/network-os/interview.md) |
| [Web 安全基础](../topics/cs/web-security/index.md) | 课程：[Web 安全基础](../topics/cs/web-security/01-xss-csrf-sqli-cors-ssrf.md)；场景：[AI 客服上线前的安全评审](../topics/cs/web-security/scenarios/ai-app-security.md)；[面试题](../topics/cs/web-security/interview.md) |
| [CI/CD](../topics/devops/cicd/index.md) | 课程：[发布与回滚](../topics/devops/cicd/02-release-rollback.md)；场景：[多语言仓库的流水线](../topics/devops/cicd/scenarios/multi-lang-pipeline.md)；[面试题](../topics/devops/cicd/interview.md) |
| [Docker 与 Compose](../topics/devops/docker/index.md) | 场景：[一条命令拉起开发环境](../topics/devops/docker/scenarios/compose-dev-env.md)；[面试题](../topics/devops/docker/interview.md) |
| [Kubernetes 基础](../topics/devops/k8s/index.md) | 课程：[Kubernetes 基础](../topics/devops/k8s/01-core-objects.md)；场景：[AI Agent 服务上 K8s](../topics/devops/k8s/scenarios/agent-rollout.md)；[面试题](../topics/devops/k8s/interview.md) |
| [Linux 运维基础](../topics/devops/linux/index.md) | 课程：[常用命令](../topics/devops/linux/01-commands.md)；课程：[线上排查](../topics/devops/linux/02-troubleshooting.md)；场景：[线上故障排查](../topics/devops/linux/scenarios/incident-troubleshooting.md)；[面试题](../topics/devops/linux/interview.md) |
| [Nginx](../topics/devops/nginx/index.md) | 课程：[HTTPS、SSE / WebSocket、限流和灰度](../topics/devops/nginx/02-https-stream-limit.md)；场景：[AI 应用网关](../topics/devops/nginx/scenarios/ai-gateway-sse-canary.md)；[面试题](../topics/devops/nginx/interview.md) |

**跨技术栈场景**：[内容审核](../scenarios/content-moderation.md)、[数据同步](../scenarios/data-sync.md)、[秒杀 / 高并发抢购](../scenarios/flash-sale.md)、[多租户 SaaS](../scenarios/multi-tenant-saas.md)、[报表生成 Agent](../scenarios/report-agent.md)、[工单系统](../scenarios/ticket-system.md)

<!-- week-topics:end -->

## 第 7 天（调整）：了解级扩展——ES、MongoDB、K8s、数据管道、Node.js、Java、计算机基础

> 依据 BOSS直聘 2026-10-23 调研新增。时间有限时按顺序做，前 3 项优先（与 RAG / Agent 项目关系最紧）；Go 网关（上面的可选第 7 天）放到时间富余时再做。

| 顺序 | 内容 | 用时 | 做什么 | 专题 |
| --- | --- | --- | --- | --- |
| 1 | ES 混合检索 | 1.5 小时 | 跑通 `es_hybrid.py`（BM25 + kNN + RRF + 别名切换），对比第 6 周自己写的 BM25 | [Elasticsearch](../topics/backend/elasticsearch/index.md) |
| 2 | 数据管道 | 1 小时 | 读懂并运行 `incremental-sync`（水位线、限流、重试、幂等、质量检查），想清楚项目二的文档同步怎么改造 | [数据管道](../topics/backend/data-pipeline/index.md) |
| 3 | K8s 基础 | 1 小时 | 读懂 `agent-api` 清单，用 kubeconform 校验；有条件用 kind / minikube 部署一次 | [Kubernetes](../topics/devops/k8s/index.md) |
| 4 | MongoDB | 45 分钟 | 单节点副本集 + `mongo_demo.py`（建模、ESR 索引、事务、变更流） | [MongoDB](../topics/backend/mongodb/index.md) |
| 5 | Node.js / Java | 45 分钟 | Express BFF 示例跑测试；`java OrderDemo.java` 跑一次，能读懂 Spring Boot 分层 | [Node.js](../topics/backend/nodejs/index.md)、[Java](../topics/backend/java-spring/index.md) |
| 6 | 计算机基础 | 1 小时 | 网络（粘包、多路复用示例）和 Web 安全（`secure-api` 13 个攻击测试）；算法放到第 16 周 | [网络与操作系统](../topics/cs/network-os/index.md)、[Web 安全](../topics/cs/web-security/index.md) |

### 完成标准

- 上表每项的示例都在本机运行过（或写明为什么没有运行，例如没有 K8s 集群），每个专题的面试题能用自己的话回答“基础”部分。
- `notes/extend.md` 记录每项“我能讲清楚的 3 句话”。

## 本周验收

- [ ] RabbitMQ、gRPC、Redis、MySQL 四组测试在本机全部通过（连接对应服务，不是跳过）
- [ ] `notes/mq.md`、`notes/microservices.md`、`notes/redis.md`、`notes/mysql.md` 写完，每个面试题都能脱稿回答
- [ ] 三个项目部署完成（或本机 Compose 全部跑通），`deploy/deploy.md` 写清备份和回滚
- [ ] CI 能自动跑测试；（有服务器时）推送后自动部署
- [ ] 第 7 天（调整）的了解级扩展完成，`notes/extend.md` 写完
- [ ] （可选）Go 网关测试通过并接入部署
- [ ] 在总纲进度表勾掉第 14 周