_{🧠 图解记忆：Global Interpreter Lock（全局解释器锁）；点击图片可查看原图。}

💡 答案要点

GIL 是什么：

Global Interpreter Lock（全局解释器锁）。Python 的机制，同一时刻只有一个线程执行 Python 字节码。

单线程 Python：
线程1: [获取GIL] → [执行字节码] → [释放GIL] → ...
                       ↓
              其他线程等待

多线程 Python（CPU 密集）：
线程1: [GIL] → [执行] → [释放] → ...    线程2: [等待GIL]...
                       ↓
              CPU 利用率 ≈ 1核（其他核空闲）

AI 应用中的 GIL 场景：

场景	GIL 影响	解决方案
LLM API 调用（I/O 等待）	无影响	asyncio（I/O 密集）
Embedding 模型（CPU 计算）	影响大	多进程
Tokenizer / 后处理	影响大	多进程
数据预处理	影响大	multiprocessing

多进程方案：

展开 Python 代码示例（31 行）

import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from functools import partial

# 场景：批量 embedding（CPU 密集）
def embed_texts_worker(texts: list[str], model_name: str) -> list[list[float]]:
    """Worker 进程函数"""
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer(model_name)
    return model.encode(texts).tolist()

def batch_embed(texts: list[str], model_name: str = "BAAI/bge-large") -> list[list[float]]:
    """多进程批量 embedding"""
    n_workers = mp.cpu_count()
    chunk_size = max(1, len(texts) // n_workers)
    
    # 分块
    chunks = [texts[i:i+chunk_size] for i in range(0, len(texts), chunk_size)]
    
    # 每个块在独立进程中处理
    with ProcessPoolExecutor(max_workers=n_workers) as executor:
        results = list(executor.map(
            partial(embed_texts_worker, model_name=model_name),
            chunks
        ))
    
    # 合并结果
    return [item for chunk in results for item in chunk]

# 使用
embeddings = batch_embed(["文本1", "文本2", "文本3", "文本4"])

多进程 vs 多线程 vs asyncio 对比：

场景                  推荐方案         原因
────────────────────────────────────────────────────
LLM API 调用          asyncio          I/O 等待，GIL 不影响
Embedding 模型推理    ProcessPool      CPU 密集，GIL 阻塞
数据预处理            ThreadPool       I/O + CPU 混合
模型推理（GPU）        单进程 + CUDA    GPU 无 GIL 问题

进程池复用（避免重复加载模型）：

展开 Python 代码示例（47 行）

class EmbeddingWorkerPool:
    """进程池 + 模型复用"""
    
    def __init__(self, model_name: str, n_workers: int = None):
        self.model_name = model_name
        self.n_workers = n_workers or mp.cpu_count()
        
        # 进程池（在主进程创建）
        self.pool = mp.Pool(
            processes=self.n_workers,
            initializer=self._init_worker,
            initargs=(model_name,)
        )
    
    @staticmethod
    def _init_worker(model_name: str):
        """Worker 进程初始化（每个进程只执行一次）"""
        global _model
        from sentence_transformers import SentenceTransformer
        _model = SentenceTransformer(model_name)
        print(f"进程 {mp.current_process().name} 加载模型完成")
    
    def embed(self, texts: list[str]) -> list[list[float]]:
        """使用进程池 embedding"""
        global _model
        return _model.encode(texts).tolist()
    
    def batch_embed(self, texts: list[str], batch_size: int = 32) -> list[list[float]]:
        """带 batch 的批量处理"""
        results = []
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i+batch_size]
            chunk_results = self.pool.map(_embed_batch, batch)
            results.extend(chunk_results)
        return results
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.pool.close()
        self.pool.join()


# 使用
with EmbeddingWorkerPool("BAAI/bge-large") as pool:
    embeddings = pool.batch_embed(["文本1", "文本2", "文本3", "文本4"])

面试话术：

"GIL 的影响取决于代码是否执行 Python 字节码。网络 I/O 常适合 asyncio；纯 Python 的 CPU 密集任务可考虑多进程；但 NumPy、tokenizer、推理库等原生扩展可能释放 GIL，GPU 工作也主要在设备端，因此不能一概而论。先用 profiler 定位 Python CPU、原生算子、数据搬运还是 GPU 瓶颈，再在线程、多进程、异步和批处理之间选择。"

📚 参考：PEP 703 – Making the Global Interpreter Lock Optional in CPython

---