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_blockTrue或创建新连接如果pool_blockFalse。创建新连接意味着额外的 TCPTLS 握手开销——这正是 Session 试图避免的。问题三ThreadPoolExecutor 的两种 Session 策略互有优劣。策略 A共享 Session连接复用最好、内存占用最少但 Cookie 不安全。策略 B每线程独立 SessionCookie 安全、连接池资源浪费。策略 Cthreading.local()Session两者折中——每个线程有独立的 SessionCookie 安全但同一线程内连接可复用。三种策略对比策略 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_localthreading.local()defget_session()-requests.Session:获取当前线程的专属 Sessionifnothasattr(thread_local,session):thread_local.sessionrequests.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 管理”sessionrequests.Session()# 共享连接池deffetch_with_user_cookie(url,user_token):# 每次传入独立的 Cookie不依赖 Session 的 Cookie Jarrespsession.get(url,cookies{token:user_token})returnresp.json()小胖“那并发数和连接池大小怎么配置20 个线程pool_maxsize 设多少”大师“黄金法则pool_maxsize 并发线程数。20 个线程 - pool_maxsize 20。如果 pool_maxsize 线程数多出来的线程要么阻塞等待pool_blockTrue要么每次创建新连接再立即关闭pool_blockFalse——浪费资源。但也不是越大越好——每个活跃连接都占用一个 TCP 端口和内核 socket 缓冲区。”生活比喻技术映射多窗口银行柜台多线程并发请求排队叫号一个号码一个业务threading.local() 隔离 Session所有窗口共用一个大堂经理共享 Session 的 Cookie 非线程安全柜台太少客户排队pool_maxsize 不足大堂经理同时服务多人搞混了Cookie 跨线程泄露3. 项目实战环境准备pipinstallrequests分步实现步骤一共享 Session 的 Cookie 污染实验目标直观验证共享 Session 在多线程下的 Cookie 交叉污染问题。importrequestsimportthreadingimporttimefromconcurrent.futuresimportThreadPoolExecutordefsimulate_cookie_leak():模拟多线程共享 Session 导致的 Cookie 泄露shared_sessionrequests.Session()results[]# 记录用户看到的是谁的 Cookiedeffetch_as_user(user_id:str):模拟以某个用户身份请求# 先设置该用户的 Cookie模拟登录shared_session.get(https://httpbin.org/cookies/set,params{userId:user_id,session:fsess_{user_id}},)# 等待一点时间让其他线程有机会干扰time.sleep(0.01)# 读取当前 Session 持有的 Cookierespshared_session.get(https://httpbin.org/cookies)cookies_seenresp.json().get(cookies,{})results.append({thread_user:user_id,cookies_seen:cookies_seen,})# 10 个线程并发请求users[fuser_{i}foriinrange(10)]withThreadPoolExecutor(max_workers10)asexecutor:foruserinusers:executor.submit(fetch_as_user,user)# 检查结果有多少线程看到了不属于自己的 Cookiewrong_count0forrinresults:seenr[cookies_seen].get(userId,unknown)ifseen!r[thread_user]:wrong_count1print(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_localthreading.local()defget_thread_session()-requests.Session:获取线程专属的 Session线程安全ifnothasattr(_thread_local,session):srequests.Session()# 每个线程的 Session 可以有自己的默认配置s.headers.update({User-Agent:ThreadSafeClient/1.0})_thread_local.sessionsreturn_thread_local.sessiondeffetch_url(url:str)-dict:安全的并发请求函数sessionget_thread_session()starttime.perf_counter()try:respsession.get(url,timeout10)elapsedtime.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:int10)-list:并发获取多个 URLresults[]starttime.perf_counter()withThreadPoolExecutor(max_workersmax_workers)asexecutor:futures{executor.submit(fetch_url,url):urlforurlinurls}forfutureinas_completed(futures):resultfuture.result()results.append(result)total_timetime.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, fP50{statistics.median(elapsed_list):.1f}ms, fmax{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?q1,https://httpbin.org/get?q2,https://httpbin.org/get?q3,https://httpbin.org/headers,https://httpbin.org/ip,https://httpbin.org/user-agent,]concurrent_fetch(urls*2,max_workers6)# 12 个请求6 线程步骤三连接池调优——并发场景下的 pool_maxsize目标验证不同 pool_maxsize 对并发性能的影响。importrequestsfromrequests.adaptersimportHTTPAdapterfromconcurrent.futuresimportThreadPoolExecutor,as_completedimporttimeimportstatisticsdefbenchmark_pool_size(pool_maxsize:int,n_requests:int30,n_workers:int15):测试不同连接池大小对并发性能的影响sessionrequests.Session()adapterHTTPAdapter(pool_maxsizepool_maxsize)session.mount(https://,adapter)times[]deffetch():starttime.perf_counter()try:session.get(https://httpbin.org/get,timeout10)times.append(time.perf_counter()-start)exceptException:passstarttime.perf_counter()withThreadPoolExecutor(max_workersn_workers)asexecutor:futures[executor.submit(fetch)for_inrange(n_requests)]forfinas_completed(futures):f.result()totaltime.perf_counter()-start session.close()avgstatistics.mean(times)*1000iftimeselse0p99sorted(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]:resultbenchmark_pool_size(size,n_requests30,n_workers10)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]})可能遇到的坑及解决方法坑1threading.local()的 Session 不会自动关闭# threading.local() 在线程退出时不会自动调用 close()# 长时间运行可能导致连接泄露# 解决: 在 Worker 线程退出前手动关闭或使用 atexit 注册清理importatexit atexit.register(lambda:getattr(_thread_local,session,None)and_thread_local.session.close())坑2GIL 限制了 requests 线程并发的 CPU 密集型部分# requests 的 JSON 解析、SSL 加解密受 GIL 影响# 但 HTTP IO 等待期间 GIL 被释放所以 IO 密集型场景线程并发仍然有效# 如果 JSON 解析是瓶颈考虑用 orjson 替代标准 json坑3连接池耗尽时pool_blockTrue导致死锁风险# 两个线程各持有一个连接互相等待对方释放 - 死锁# 建议: pool_blockFalse默认让额外线程创建临时连接# 或: 设置合理的 pool_maxsize 预期最大并发线程数测试验证importpytestimportrequestsimportthreadingfromconcurrent.futuresimportThreadPoolExecutorclassTestConcurrentRequests:验证并发请求功能deftest_thread_local_session_isolation(self):验证 threading.local() 为不同线程创建不同 Sessionlocalthreading.local()sessions_seen[]defget_and_record():ifnothasattr(local,s):local.srequests.Session()sessions_seen.append(id(local.s))withThreadPoolExecutor(max_workers5)ase:for_inrange(20):e.submit(get_and_record)# 所有线程都应该看到同一个 Session ID因为只有 5 个线程重复使用# 但不同线程之间的 Session 是不同的unique_sessionslen(set(sessions_seen))print(f\n Session 实例数:{unique_sessions})assert1unique_sessions5deftest_concurrent_requests_succeed(self):urls[https://httpbin.org/get]*5deffetch(url):returnrequests.get(url,timeout10).status_codewithThreadPoolExecutor(max_workers3)ase:resultslist(e.map(fetch,urls))assertall(r200forrinresults)deftest_session_not_thread_safe_for_cookies(self):验证 Session Cookie 在多线程下不安全预期行为srequests.Session()cookie_issues[]defuser_flow(user_id):s.get(https://httpbin.org/cookies/set,params{user:user_id})resps.get(https://httpbin.org/cookies)seenresp.json().get(cookies,{}).get(user,unknown)ifseen!user_id:cookie_issues.append((user_id,seen))withThreadPoolExecutor(max_workers5)ase:foriinrange(20):e.submit(user_flow,fu{i})print(f\n Cookie 不一致次数:{len(cookie_issues)})# 期望有 Cookie 交叉污染证明非线程安全# 如果为 0只是运气好s.close()4. 项目总结核心知识点策略Cookie 安全连接复用代码复杂度全局共享 Session❌ 不安全✅ 最好低每请求创建 Session✅ 安全❌ 无复用低threading.local() Session✅ 安全✅ 线程内复用中每个请求传 cookies 参数✅ 安全✅ 复用中优点 缺点对比维度ThreadPoolExecutorasyncio aiohttp单线程串行并发能力中受限于线程数高协程切换无编程模型同步易理解异步需 async/await同步GIL 影响IO 密集时影响小几乎没有无内存占用每线程 ~8MB 栈每协程 ~KB 级最低适用场景聚合多个下游 API并发调用减少总延迟批量数据采集并发下载 N 个页面API 网关/聚合层高并发路由请求定时任务批处理加快处理速度注意事项Session 的 Cookie 不是线程安全的——每个线程应有独立 Session 或每次传 cookiespool_maxsize 应 并发线程数避免连接池竞争threading.local() 需要手动管理 Session 生命周期closeGIL 对 SSL 加解密和 JSON 解析有影响极高并发场景考虑 asyncio常见踩坑经验案例一共享 Session 导致用户串号。某电商网关 20 个 Worker 线程共享一个 Session。高峰期用户 A 的认证 Token通过 Set-Cookie 返回被写入了 session.cookies同时用户 B 的请求恰好复用这个 Cookie导致 B 看到了 A 的数据。P0 级数据泄露。根因Session Cookie 非线程安全。修复改用 threading.local() 或每次请求显式传入 cookies。案例二pool_maxsize220 线程饥饿。某数据采集程序 20 线程共享一个 Session默认 pool_maxsize10。实际测试发现 10 个线程阻塞等待连接。根因连接池不够大且pool_blockTrue导致线程排队。修复pool_maxsize 调整为 30或 pool_blockFalse 允许创建临时连接。案例三threading.local() Session 未关闭导致 fd 泄漏。某服务长时间运行不重启后发现Too many open files错误。排查发现 threading.local() 创建的 Session 在线程死亡后没有调用 close()连接池中的 socket 文件描述符未释放。根因未管理 Session 生命周期。修复在 Worker 退出时显式session.close()。思考题设计题你需要设计一个并发请求协调器——控制对同一个 host 的并发连接数不超过 pool_maxsize但对不同 host 的请求不互相限制。如何使用信号量Semaphore实现这个功能深度题Python 的 GILGlobal 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 实战修炼与源码剖析