第26章：并发请求——多线程与Session线程安全性

1. 项目背景

业务场景

某电商平台的商品详情页需要聚合 8 个微服务的数据：商品基本信息、库存、价格、评价（3 个来源）、推荐、用户收藏状态。最初方案是串行调用——总耗时 = 8 个服务的耗时之和，P99 延迟高达 1.2 秒，远超 300ms 的 SLO。架构师要求改为并发调用，预期将总耗时降低到最慢的那个服务的时间（约 200ms）。

团队用ThreadPoolExecutor+ 共享Session实现了并发。上线第一周一切正常，P99 降到了 250ms。但第二周问题爆发了：部分用户的收藏状态显示错误——用户 A 看到了用户 B 的收藏列表。排查发现：所有线程共享同一个 Session，而 Session 的 Cookie 不是线程安全的——线程 A 设置了用户 A 的 Cookie，但被线程 B 的响应 Cookie 覆盖了，线程 A 后续请求带上了线程 B 的 Cookie。

痛点

问题一：Session 的"大致线程安全"到底哪里不安全。官方文档说 Session 是 “roughly thread-safe”，但这个 “roughly” 到底指什么？实际上：Session 的连接池（urllib3 PoolManager）是线程安全的（使用锁保护），但 Cookie 管理（session.cookies）不是线程安全的。更微妙的是——Session 的headers、auth、hooks等属性在运行时修改也不是线程安全的。

问题二：连接池在并发场景下的竞争。pool_maxsize默认 10，如果 20 个线程共享一个 Session 去同一个 host，只有 10 个线程能从连接池中拿到连接，另外 10 个会阻塞等待（如果pool_block=True）或创建新连接（如果pool_block=False）。创建新连接意味着额外的 TCP+TLS 握手开销——这正是 Session 试图避免的。

问题三：ThreadPoolExecutor 的两种 Session 策略互有优劣。策略 A（共享 Session）：连接复用最好、内存占用最少，但 Cookie 不安全。策略 B（每线程独立 Session）：Cookie 安全、连接池资源浪费。策略 C（threading.local()Session）：两者折中——每个线程有独立的 Session（Cookie 安全），但同一线程内连接可复用。

三种策略对比：

策略 A: 全局共享 Session ✅ 连接复用最好 ✅ 代码最简单 ❌ Cookie 非线程安全 ❌ 连接池竞争激烈 策略 B: 每个请求创建 Session ✅ Cookie 完全隔离 ❌ 无连接复用（每次新建连接） ❌ 性能最差 策略 C: threading.local() Session ✅ Cookie 隔离（每线程一个） ✅ 同一线程内连接复用 ✅ 性能接近策略 A ❌ 代码稍复杂

2. 项目设计

小胖（指着生产日志，满头大汗）：
“大师！我们的商品详情页出大问题了！用户 A 看到了用户 B 的收藏列表——这是严重的隐私泄露！代码审查看了一遍，逻辑完全没问题啊。”

大师（看了看代码）：
“你们是不是用的是一个全局 Session，多个线程共享？”

小胖：
“对啊，线程池里 20 个 Worker，共享一个 Session——网上教程都这么写的，说 Session 是线程安全的。”

大师：
“这就是问题所在。Session 的 HTTP 连接池是线程安全的，但 Cookie 管理不是。线程 A 发了请求，Session 收到响应后把 Set-Cookie 存入了session.cookies。与此同时，线程 B 也在发请求——它看到 CookieJar 里是线程 A 的 Cookie，就带着它发出了。这就是跨用户 Cookie 泄露的根因。”

小胖（挠头）：
“那怎么办？让每个线程创建自己的 Session？那连接复用不就没了吗？”

大师：
“有个折中方案——threading.local()。你可以把它想象成’每人发一个独立的钱包（Cookie），但共用一辆公交车（连接池）'——但很遗憾，连接池也跟 Session 绑定了。所以实践中最常用的方案是：每个线程维护自己的 Session，用连接池的复用收益换 Cookie 的安全性。”

importthreadingimportrequests# 方案 C: threading.local()thread_local=threading.local()defget_session()->requests.Session:"""获取当前线程的专属 Session"""ifnothasattr(thread_local,"session"):thread_local.session=requests.Session()thread_local.session.headers.update({"Authorization":"Bearer xxx"})returnthread_local.session# 每个线程调用 get_session() 获取自己的 Session# 同一线程内多次调用返回同一个 Session（连接复用）# 不同线程之间完全隔离（Cookie 安全）

小白（追问）：
“那如果我不想在每个线程里创建 Session（因为连接池很贵），能不能让 Session 不管理 Cookie，只用它的连接池？”

大师：
“可以的。你可以在请求前手动session.cookies.clear()，或者用另一个简单的方法——为所有请求显式传cookies参数，而不是依赖 Session 的自动 Cookie 管理：”

session=requests.Session()# 共享连接池deffetch_with_user_cookie(url,user_token):# 每次传入独立的 Cookie，不依赖 Session 的 Cookie Jarresp=session.get(url,cookies={"token":user_token})returnresp.json()

小胖：
“那并发数和连接池大小怎么配置？20 个线程，pool_maxsize 设多少？”

大师：
“黄金法则：pool_maxsize >= 并发线程数。20 个线程 -> pool_maxsize >= 20。如果 pool_maxsize < 线程数，多出来的线程要么阻塞等待（pool_block=True），要么每次创建新连接再立即关闭（pool_block=False）——浪费资源。但也不是越大越好——每个活跃连接都占用一个 TCP 端口和内核 socket 缓冲区。”

3. 项目实战

环境准备

pipinstallrequests

分步实现

步骤一：共享 Session 的 Cookie 污染实验

目标：直观验证共享 Session 在多线程下的 Cookie 交叉污染问题。

importrequestsimportthreadingimporttimefromconcurrent.futuresimportThreadPoolExecutordefsimulate_cookie_leak():"""模拟多线程共享 Session 导致的 Cookie 泄露"""shared_session=requests.Session()results=[]# 记录"用户看到的是谁的 Cookie"deffetch_as_user(user_id:str):"""模拟以某个用户身份请求"""# 先设置该用户的 Cookie（模拟登录）shared_session.get("https://httpbin.org/cookies/set",params={"userId":user_id,"session":f"sess_{user_id}"},)# 等待一点时间让其他线程有机会干扰time.sleep(0.01)# 读取当前 Session 持有的 Cookieresp=shared_session.get("https://httpbin.org/cookies")cookies_seen=resp.json().get("cookies",{})results.append({"thread_user":user_id,"cookies_seen":cookies_seen,})# 10 个线程并发请求users=[f"user_{i}"foriinrange(10)]withThreadPoolExecutor(max_workers=10)asexecutor:foruserinusers:executor.submit(fetch_as_user,user)# 检查结果：有多少线程看到了不属于自己的 Cookie？wrong_count=0forrinresults:seen=r["cookies_seen"].get("userId","unknown")ifseen!=r["thread_user"]:wrong_count+=1print(f" ⚠{r['thread_user']}看到了{seen}的 Cookie!")print(f"\nCookie 交叉污染率:{wrong_count}/{len(results)}"f"({wrong_count/len(results)*100:.0f}%)")# 结论：共享 Session 在高并发下 Cookie 安全无法保证shared_session.close()returnresultsprint("===== Cookie 泄露实验 =====")simulate_cookie_leak()

步骤二：threading.local() 安全并发方案

目标：实现线程安全的 Session 管理，避免 Cookie 泄露。

importrequestsimportthreadingfromconcurrent.futuresimportThreadPoolExecutor,as_completedimporttimeimportstatistics# 为每个线程维护独立的 Session_thread_local=threading.local()defget_thread_session()->requests.Session:"""获取线程专属的 Session（线程安全）"""ifnothasattr(_thread_local,"session"):s=requests.Session()# 每个线程的 Session 可以有自己的默认配置s.headers.update({"User-Agent":"ThreadSafeClient/1.0"})_thread_local.session=sreturn_thread_local.sessiondeffetch_url(url:str)->dict:"""安全的并发请求函数"""session=get_thread_session()start=time.perf_counter()try:resp=session.get(url,timeout=10)elapsed=time.perf_counter()-startreturn{"url":url,"status":resp.status_code,"elapsed_ms":round(elapsed*1000,2),"thread":threading.current_thread().name,}exceptExceptionase:return{"url":url,"status":-1,"elapsed_ms":0,"error":str(e)[:50],"thread":threading.current_thread().name,}defconcurrent_fetch(urls:list,max_workers:int=10)->list:"""并发获取多个 URL"""results=[]start=time.perf_counter()withThreadPoolExecutor(max_workers=max_workers)asexecutor:futures={executor.submit(fetch_url,url):urlforurlinurls}forfutureinas_completed(futures):result=future.result()results.append(result)total_time=time.perf_counter()-start success=[rforrinresultsifr["status"]>0]failed=[rforrinresultsifr["status"]<0]elapsed_list=[r["elapsed_ms"]forrinsuccess]print(f"\n===== 并发请求结果 =====")print(f"总数:{len(urls)}, 成功:{len(success)}, 失败:{len(failed)}")ifelapsed_list:print(f"延迟: avg={statistics.mean(elapsed_list):.1f}ms, "f"P50={statistics.median(elapsed_list):.1f}ms, "f"max={max(elapsed_list):.1f}ms")print(f"总耗时(并发):{total_time:.2f}s")print(f"加速比:{sum(elapsed_list)/total_time:.1f}x (vs 串行)")returnresults# ===== 测试：并发请求 httpbin 的多个端点 =====urls=["https://httpbin.org/get?q=1","https://httpbin.org/get?q=2","https://httpbin.org/get?q=3","https://httpbin.org/headers","https://httpbin.org/ip","https://httpbin.org/user-agent",]concurrent_fetch(urls*2,max_workers=6)# 12 个请求，6 线程

步骤三：连接池调优——并发场景下的 pool_maxsize

目标：验证不同 pool_maxsize 对并发性能的影响。

importrequestsfromrequests.adaptersimportHTTPAdapterfromconcurrent.futuresimportThreadPoolExecutor,as_completedimporttimeimportstatisticsdefbenchmark_pool_size(pool_maxsize:int,n_requests:int=30,n_workers:int=15):"""测试不同连接池大小对并发性能的影响"""session=requests.Session()adapter=HTTPAdapter(pool_maxsize=pool_maxsize)session.mount("https://",adapter)times=[]deffetch():start=time.perf_counter()try:session.get("https://httpbin.org/get",timeout=10)times.append(time.perf_counter()-start)exceptException:passstart=time.perf_counter()withThreadPoolExecutor(max_workers=n_workers)asexecutor:futures=[executor.submit(fetch)for_inrange(n_requests)]forfinas_completed(futures):f.result()total=time.perf_counter()-start session.close()avg=statistics.mean(times)*1000iftimeselse0p99=sorted(times)[int(len(times)*0.99)]*1000iftimeselse0return{"pool_maxsize":pool_maxsize,"workers":n_workers,"total_s":round(total,2),"avg_ms":round(avg,1),"p99_ms":round(p99,1),"qps":round(n_requests/total,1),}print("===== 连接池大小对比 =====")print(f"{'pool_maxsize':<15}{'workers':<10}{'total_s':<10}{'avg_ms':<10}{'p99_ms':<10}{'qps'}")print("-"*65)forsizein[5,10,20]:result=benchmark_pool_size(size,n_requests=30,n_workers=10)print(f"{result['pool_maxsize']:<15}{result['workers']:<10}"f"{result['total_s']:<10}{result['avg_ms']:<10}"f"{result['p99_ms']:<10}{result['qps']}")

可能遇到的坑及解决方法

坑1：threading.local()的 Session 不会自动关闭

# threading.local() 在线程退出时不会自动调用 close()# 长时间运行可能导致连接泄露# 解决: 在 Worker 线程退出前手动关闭，或使用 atexit 注册清理importatexit atexit.register(lambda:getattr(_thread_local,'session',None)and_thread_local.session.close())

坑2：GIL 限制了 requests 线程并发的 CPU 密集型部分

# requests 的 JSON 解析、SSL 加解密受 GIL 影响# 但 HTTP IO 等待期间 GIL 被释放，所以 IO 密集型场景线程并发仍然有效# 如果 JSON 解析是瓶颈，考虑用 orjson 替代标准 json

坑3：连接池耗尽时pool_block=True导致死锁风险

# 两个线程各持有一个连接，互相等待对方释放 -> 死锁# 建议: pool_block=False（默认），让额外线程创建临时连接# 或: 设置合理的 pool_maxsize >= 预期最大并发线程数

测试验证

importpytestimportrequestsimportthreadingfromconcurrent.futuresimportThreadPoolExecutorclassTestConcurrentRequests:"""验证并发请求功能"""deftest_thread_local_session_isolation(self):"""验证 threading.local() 为不同线程创建不同 Session"""local=threading.local()sessions_seen=[]defget_and_record():ifnothasattr(local,"s"):local.s=requests.Session()sessions_seen.append(id(local.s))withThreadPoolExecutor(max_workers=5)ase:for_inrange(20):e.submit(get_and_record)# 所有线程都应该看到同一个 Session ID（因为只有 5 个线程重复使用）# 但不同线程之间的 Session 是不同的unique_sessions=len(set(sessions_seen))print(f"\n Session 实例数:{unique_sessions}")assert1<=unique_sessions<=5deftest_concurrent_requests_succeed(self):urls=["https://httpbin.org/get"]*5deffetch(url):returnrequests.get(url,timeout=10).status_codewithThreadPoolExecutor(max_workers=3)ase:results=list(e.map(fetch,urls))assertall(r==200forrinresults)deftest_session_not_thread_safe_for_cookies(self):"""验证 Session Cookie 在多线程下不安全（预期行为）"""s=requests.Session()cookie_issues=[]defuser_flow(user_id):s.get("https://httpbin.org/cookies/set",params={"user":user_id})resp=s.get("https://httpbin.org/cookies")seen=resp.json().get("cookies",{}).get("user","unknown")ifseen!=user_id:cookie_issues.append((user_id,seen))withThreadPoolExecutor(max_workers=5)ase:foriinrange(20):e.submit(user_flow,f"u{i}")print(f"\n Cookie 不一致次数:{len(cookie_issues)}")# 期望：有 Cookie 交叉污染（证明非线程安全）# 如果为 0，只是运气好s.close()

4. 项目总结

核心知识点

优点 & 缺点

适用场景

• 聚合多个下游 API：并发调用减少总延迟
• 批量数据采集：并发下载 N 个页面
• API 网关/聚合层：高并发路由请求
• 定时任务批处理：加快处理速度

注意事项

• Session 的 Cookie 不是线程安全的——每个线程应有独立 Session 或每次传 cookies
• pool_maxsize 应 >= 并发线程数，避免连接池竞争
• threading.local() 需要手动管理 Session 生命周期（close）
• GIL 对 SSL 加解密和 JSON 解析有影响，极高并发场景考虑 asyncio

常见踩坑经验

案例一：共享 Session 导致用户串号。某电商网关 20 个 Worker 线程共享一个 Session。高峰期用户 A 的认证 Token（通过 Set-Cookie 返回）被写入了 session.cookies，同时用户 B 的请求恰好复用这个 Cookie，导致 B 看到了 A 的数据。P0 级数据泄露。根因：Session Cookie 非线程安全。修复：改用 threading.local() 或每次请求显式传入 cookies。

案例二：pool_maxsize=2，20 线程饥饿。某数据采集程序 20 线程共享一个 Session，默认 pool_maxsize=10。实际测试发现 10 个线程阻塞等待连接。根因：连接池不够大且pool_block=True导致线程排队。修复：pool_maxsize 调整为 30，或 pool_block=False 允许创建临时连接。

案例三：threading.local() Session 未关闭导致 fd 泄漏。某服务长时间运行（不重启）后发现Too many open files错误。排查发现 threading.local() 创建的 Session 在线程死亡后没有调用 close()，连接池中的 socket 文件描述符未释放。根因：未管理 Session 生命周期。修复：在 Worker 退出时显式session.close()。

思考题

1. 设计题：你需要设计一个"并发请求协调器"——控制对同一个 host 的并发连接数不超过 pool_maxsize，但对不同 host 的请求不互相限制。如何使用信号量（Semaphore）实现这个功能？

2. 深度题：Python 的 GIL（Global Interpreter Lock）在 IO 密集场景（如 HTTP 请求）和 CPU 密集场景（如 JSON 解析、SSL 加密）下的影响有何不同？为什么 requests 的线程并发在 IO 密集时加速比接近线程数，但在 JSON 解析量大时加速比变差？

延伸阅读与资源

Milvus向量数据库实战修炼：从 0 到 1精通向量检索与生产落地
后端工程师的 AI 转型第一课：Ollama 与私有化大模型实战
10倍开发者的 Dify 魔法书：从零构建全栈 AI 应用
后端工程师转型AI第一课-Ollama 与私有化大模型实战

大型语言模型(LLM) vLLM 高性能推理落地实战

Agent开发之LlamaIndex 实战修炼与源码进阶

大语言模型Transformers 实战修炼与源码剖析