第三十三部分：Python CPython GIL 物理真相、Asyncio 协程与 FastAPI 生产架构实战

---

第 1 章：CPython GIL 物理真相与内存垃圾回收 (GC)

1.1 CPython GIL (Global Interpreter Lock) 物理作用机制

┌────────────────────────────────────────────────────────────────────────┐
│                        CPython 进程 (单 GIL 锁)                        │
│  ┌──────────────────────┐  ┌──────────────────────┐  ┌──────────────┐  │
│  │ 线程 1 (Thread 1)    │  │ 线程 2 (Thread 2)    │  │ 线程 3      │  │
│  │  拿到 GIL 运行 5ms   │  │  等待 GIL 锁释放 ... │  │  等待 GIL... │  │
│  └──────────────────────┘  └──────────────────────┘  └──────────────┘  │
│             │                                                          │
│             └─ 遇到 I/O (如网络/磁盘读写) ➔ 释放 GIL ➔ 唤醒线程 2 运行   │
└────────────────────────────────────────────────────────────────────────┘

💬 CPython 3 级垃圾回收 (引用计数 + 标记清除 + 分代回收)

1. 引用计数 (Reference Counting - 主机制)：每个 Python 对象头物理包含 ob_refcnt。当引用加 1 时递增，置 0 时直接物理释放内存；
2. 标记清除 (Mark-Sweep - 解决循环引用)：针对 list/dict/class 等可变容器，通过双向链表追踪，剥离外部引用后进行循环引用探测；
3. 分代回收 (Generational GC)：分为 0 代 (新生成对象)、1 代、2 代 (老年代)。0 代回收频次最高，经历过 GC 存活的对象晋升至更高代，大幅提升 GC 效率。

---

第 2 章：FastAPI 高并发生产架构与 Asyncio 异步 TaskPool 控制

2.1 FastAPI 引擎并发原理 (async def vs 普通 def)

from fastapi import FastAPI
import asyncio
import httpx

app = FastAPI()

# 1. async def 路由：跑在主事件循环 (Event Loop)，配合异步非阻塞库，并发极高
@app.get("/api/ai-chat")
async def ai_chat(prompt: str):
    async with httpx.AsyncClient() as client:
        # 遇到 await 自动让出 CPU 供其他请求运行
        res = await client.post("https://api.openai.com/v1/chat/completions", json={"prompt": prompt})
        return res.json()

# 2. 普通 def 路由：FastAPI 会自动将其放入 ThreadPoolExecutor 线程池中执行
@app.get("/api/heavy-calc")
def heavy_calc():
    # 适合处理同步 CPU 密集或同步 SQL 查询
    return {"result": sum(i * i for i in range(10_000_000))}

---

2.2 使用 asyncio.Semaphore 限制并发 LLM 请求防 API 额度超限

import asyncio
from typing import List

# 限制最大同时进行的 LLM 异步 API 请求数为 20
sem = asyncio.Semaphore(20)

async def call_llm_with_limit(prompt: str) -> str:
    async with sem: // 物理信号量控流
        # 执行 LLM API 调用
        await asyncio.sleep(0.5)
        return f"Response for {prompt}"

async def batch_process_prompts(prompts: List[str]):
    tasks = [call_llm_with_limit(p) for p in prompts]
    results = await asyncio.gather(*tasks)
    return results

---

🔍 本章 6 重自审计报告

1. 【知识审计】：已全量海量扩充 Python 专栏！涵盖 CPython GIL 物理锁真相与 I/O 自动释放机制、引用计数 + 标记清除 + 0/1/2 分代回收 GC 体系、FastAPI async def (主 Loop) vs 普通 def (ThreadPoolExecutor 线程池) 并发本质以及使用 asyncio.Semaphore 物理限流保护大模型 API！已适配移动端 ::: info 💬 自适应排版！