asyncio + 线程池：Python 混合并发模型解决 CPU/IO 绑定交错难题

asyncio + 线程池：Python 混合并发模型解决 CPU/IO 绑定交错难题

一、纯异步的困局：当 IO 密集遇上 CPU 密集

Python 的 asyncio 在 IO 密集场景下确实有优势——协程切换成本低，单线程能处理数千并发连接。但现实中的工作负载很少是纯粹的 IO 密集型。以典型的 RAG 系统为例，请求链路通常是：接收请求（IO）→ 向量检索（IO）→ 重排序计算（CPU）→ 大模型推理（IO + CPU）→ 返回结果（IO）。CPU 密集和 IO 密集步骤交替出现，纯异步模型在这里会遇到根本性问题。

问题在于事件循环的阻塞。asyncio 的协程运行在单线程中，事件循环负责调度所有协程。当某个协程执行 CPU 密集计算时（比如对检索结果做重排序），它会独占事件循环线程，导致其他协程无法被调度。结果是：系统在处理一个重排序请求时，其他请求的延迟会明显升高。

有人建议用 asyncio.to_thread 把 CPU 任务丢到线程池。这确实能解决事件循环阻塞问题，但 Python 的 GIL 限制了多线程的 CPU 并行能力。线程池只能让事件循环继续运转，无法真正加速 CPU 计算。要获得真正的 CPU 并行，必须使用多进程。

所以更完整的方案是混合并发模型：asyncio 处理 IO，线程池处理阻塞 IO，进程池处理 CPU 密集计算，三者通过事件循环统一调度。

二、混合并发模型的调度机制与数据流转

混合模型的核心思路是：事件循环作为总调度器，根据任务类型将其分发到不同的执行器（Executor）。下图展示了完整的调度架构：

graph LR
subgraph 事件循环线程
A[协程 A: IO] –> E[事件循环]
B[协程 B: IO] –> E
C[协程 C: IO] –> E
end

E –>|阻塞 IO| F[ThreadPoolExecutor]
E –>|CPU 密集| G[ProcessPoolExecutor]

F –> F1[线程 1: 文件读写]
F –> F2[线程 2: 同步 HTTP]
F –> F3[线程 3: 数据库驱动]

G –> G1[进程 1: 重排序]
G –> G2[进程 2: 向量计算]
G –> G3[进程 3: 编解码]

F1 –> E
F2 –> E
F3 –> E
G1 –> E
G2 –> E
G3 –> E

style E fill:#e1f5fe
style F fill:#fff3e0
style G fill:#fce4ec

几个关键机制值得注意：

三、RAG 系统的混合并发实现

下面是一个 RAG 系统的完整混合并发实现，覆盖向量检索、重排序和大模型推理三个阶段：

import asyncio
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from dataclasses import dataclass
from typing import Optional
import logging
import time

logger = logging.getLogger(__name__)

@dataclass
class RAGRequest:
"""RAG 请求"""
query: str
top_k: int = 10
rerank_top_n: int = 3

@dataclass
class RAGResponse:
"""RAG 响应"""
answer: str
sources: list
latency_ms: dict = None

# ———- CPU 密集任务：放在进程池中执行 ———-

def rerank_documents(query: str, docs: list[dict], top_n: int) -> list[dict]:
"""重排序：CPU 密集计算，在子进程中执行

注意：此函数在子进程中运行，不能引用主进程的对象
"""
import math

def bm25_score(query_terms: list[str], doc_text: str) -> float:
"""简化版 BM25 评分"""
score = 0.0
doc_terms = doc_text.lower().split()
doc_len = len(doc_terms)
avg_len = 100 # 假设平均文档长度
k1, b = 1.5, 0.75

for term in query_terms:
tf = doc_terms.count(term)
idf = math.log(10000 / (1 + tf))
numerator = tf * (k1 + 1)
denominator = tf + k1 * (1 – b + b * doc_len / avg_len)
score += idf * numerator / denominator
return score

query_terms = query.lower().split()
scored = [
{**doc, "rerank_score": bm25_score(query_terms, doc.get("content", ""))}
for doc in docs
]
scored.sort(key=lambda x: x["rerank_score"], reverse=True)
return scored[:top_n]

# ———- 阻塞 IO 任务：放在线程池中执行 ———-

def sync_vector_search(query: str, top_k: int) -> list[dict]:
"""同步向量检索：模拟阻塞的数据库调用

生产中替换为实际的向量数据库 SDK 调用
"""
time.sleep(0.1) # 模拟网络延迟
return [
{"id": f"doc_{i}", "content": f"文档内容 {i}，关于 {query}",
"score": 0.95 – i * 0.05}
for i in range(top_k)
]

def sync_llm_inference(prompt: str) -> str:
"""同步大模型推理：模拟阻塞的 API 调用"""
time.sleep(0.3) # 模型推理延迟
return f"基于检索结果，关于「{prompt}」的回答如下…"

# ———- 异步 RAG 服务 ———-

class RAGService:
"""混合并发 RAG 服务

– 向量检索：线程池（阻塞 IO）
– 重排序：进程池（CPU 密集）
– 大模型推理：线程池（阻塞 IO）
"""

def __init__(
self,
thread_pool_size: int = 8,
process_pool_size: int = 4,
):
self.thread_pool = ThreadPoolExecutor(
max_workers=thread_pool_size,
thread_name_prefix="rag-io",
)
self.process_pool = ProcessPoolExecutor(
max_workers=process_pool_size,
)
self.loop = asyncio.get_event_loop()

async def search(self, request: RAGRequest) -> RAGResponse:
"""执行完整的 RAG 流程"""
latencies = {}
t0 = time.monotonic()

# 第一步：向量检索（线程池，避免阻塞事件循环）
t1 = time.monotonic()
docs = await self.loop.run_in_executor(
self.thread_pool,
sync_vector_search,
request.query,
request.top_k,
)
latencies["retrieval_ms"] = (time.monotonic() – t1) * 1000

# 第二步：重排序（进程池，绕过 GIL 实现 CPU 并行）
t2 = time.monotonic()
reranked = await self.loop.run_in_executor(
self.process_pool,
rerank_documents,
request.query,
docs,
request.rerank_top_n,
)
latencies["rerank_ms"] = (time.monotonic() – t2) * 1000

# 第三步：大模型推理（线程池，阻塞 IO）
context = "\\n".join(d["content"] for d in reranked)
prompt = f"上下文：\\n{context}\\n\\n问题：{request.query}"

t3 = time.monotonic()
answer = await self.loop.run_in_executor(
self.thread_pool,
sync_llm_inference,
prompt,
)
latencies["inference_ms"] = (time.monotonic() – t3) * 1000

latencies["total_ms"] = (time.monotonic() – t0) * 1000

return RAGResponse(
answer=answer,
sources=[{"id": d["id"], "score": d.get("rerank_score", 0)}
for d in reranked],
latency_ms=latencies,
)

async def batch_search(
self, requests: list[RAGRequest]
) -> list[RAGResponse]:
"""批量处理：多个请求并发执行"""
tasks = [self.search(req) for req in requests]
return await asyncio.gather(*tasks)

def shutdown(self):
"""优雅关闭线程池和进程池"""
self.thread_pool.shutdown(wait=False)
self.process_pool.shutdown(wait=False)

# ———- 使用示例 ———-

async def main():
service = RAGService(
thread_pool_size=8,
process_pool_size=4,
)

# 单请求
req = RAGRequest(query="Python 异步编程最佳实践", top_k=10)
resp = await service.search(req)
print(f"回答: {resp.answer}")
print(f"延迟: {resp.latency_ms}")

# 批量请求
batch = [
RAGRequest(query="向量检索原理"),
RAGRequest(query="RAG 系统优化"),
RAGRequest(query="多进程并发"),
]
results = await service.batch_search(batch)
for r in results:
print(f"总延迟: {r.latency_ms['total_ms']:.1f}ms")

service.shutdown()

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

四、混合模型的工程代价与选型边界

混合并发模型解决了纯异步的 CPU 阻塞问题，但也带来了新的复杂度。

进程池的启动开销。 ProcessPoolExecutor 在首次提交任务时会 fork 子进程，启动时间约 100-300ms。如果请求频率低、CPU 任务小，启动开销可能比任务本身还大。对于短生命周期任务，可以考虑预启动进程池。

序列化是性能瓶颈。 进程间通信需要 pickle 序列化。如果传递的文档列表包含大量文本，序列化和反序列化的耗时可能占总耗时的 30% 以上。优化策略：只传文档 ID，子进程自行从共享内存或 Redis 读取数据。

错误传播更复杂。 子进程中的异常会被序列化后重新抛出，但有些异常（如 SystemExit、自定义 C 扩展异常）无法被正确序列化。生产中建议在子进程函数内部捕获所有异常，返回统一的结果对象而非抛出异常。

资源竞争。 线程池和进程池共享系统资源（CPU、内存、文件描述符）。如果线程池占满了 CPU，进程池的任务就无法及时调度。需要根据实际负载调整池大小，并设置优先级策略。

选型建议： 当 CPU 密集任务的单次执行时间 > 50ms，且请求 QPS > 10 时，混合模型的优势才明显。对于低频请求或轻量计算，asyncio.to_thread 足矣，不必引入进程池的复杂度。

五、总结

Python 的混合并发模型是解决 CPU/IO 交错负载的务实方案。asyncio 处理 IO 协程调度，ThreadPoolExecutor 处理阻塞 IO，ProcessPoolExecutor 处理 CPU 密集计算，三者通过 loop.run_in_executor() 统一编排。

核心工程要点：进程池只传小对象以减少序列化开销，子进程函数内部做异常兜底，线程池和进程池的大小需要根据实际负载动态调整。混合模型适合 CPU 任务耗时较长、请求并发较高的场景，轻量场景下简单的 asyncio.to_thread 更经济。

改写总结：