Python 多线程的价值不在真并行算 CPU而在等 I/O 的时候别人能干活。CPython 有 GIL同一时刻只有一个线程跑 Python 字节码所以纯 Python 循环/压缩/加密/数值计算丢多线程基本白搭但网络请求、磁盘读写、DB 查询、sleep、等待第三方 API 时会释放 GIL于是多个线程并发等 I/O吞吐就上来了。官方 threading 文档也写得明白线程尤其适合 I/O bound文件/网络多 CPU 真并行该 multiprocessing/ProcessPoolExecutor3.13 free-threaded build 可禁 GIL 但不默认。下面按怎么用 → 用哪儿 → threading vs asyncio 实测 → 真实项目案例一路讲透。一、Python 多线程到底在干啥一个进程里多个线程共享同一块内存地址空间。可以理解为一个程序里冒出多个执行流它们看得见同一份全局变量、同一个对象但同一时刻在 CPython 里只有一个线程跑 Python 字节码。importthreadingimporttimedefworker(name):print(f{name}start, tid{threading.get_ident()})time.sleep(2)# 模拟网络/IO 等待会释放 GILprint(f{name}done)threads[]foriinrange(4):tthreading.Thread(targetworker,args(ft{i},))threads.append(t)t.start()fortinthreads:t.join()print(all done)Thread(target...)把函数丢给新线程start()启动join()等它回来。官方文档示例也是这个套路收集 threads、start、最后 join。生产里更该用 ThreadPoolExecutor手写Thread queue能学原理但项目里更常用concurrent.futures.ThreadPoolExecutor——管线程池、提交任务、拿 Future、等完成、异常收集。concurrent.futures提供异步执行可调用对象的高层接口ThreadPoolExecutor 用线程池ProcessPoolExecutor 用进程池同一套 Executor 抽象。importtimefromconcurrent.futuresimportThreadPoolExecutor,as_completeddeffetch(url):time.sleep(0.5)# 假装 requests.get(url, timeout5)returnf{url}okurls[fhttps://example.com/{i}foriinrange(20)]withThreadPoolExecutor(max_workers8,thread_name_prefixfetcher)aspool:futures{pool.submit(fetch,u):uforuinurls}forfutinas_completed(futures):urlfutures[fut]try:print(url,fet.result())exceptExceptionase:print(url,err,e)submit()返回 Futureas_completed()谁先好谁先处理——这对爬虫/批量接口特别重要别因为第 1 个任务卡 30 秒就把后面 5ms 就完成的结果堵在后面。二、线程安全共享内存是把双刃剑线程共享内存是好事不用 pickle 传数据也是坏事竞态条件。下面这个看着对实际错counter0definc():globalcounterfor_inrange(10000):vcounter v1counterv threads[threading.Thread(targetinc)for_inrange(4)]fortinthreads:t.start()fortinthreads:t.join()print(counter)# 经常不到 40000多个线程同时读/写counter读到的旧值互相覆盖。解法threading.Lock()lockthreading.Lock()definc_safe():globalcounterfor_inrange(10000):withlock:vcounter v1counterv官方 threading 就是这么干的Lock/RLock/Condition/Event/Semaphore/Barrier 管同步queue.Queue是线程安全的天生适合生产者-消费者。几个常用同步原语原语用途Lock/RLock保护共享变量/临界区Queue线程安全任务队列生产者-消费者Event一个线程喊准备好了/退出吧Condition复杂状态通知生产者-消费者升级版Semaphore限流比如最多 10 个 DB 连接Timer/local()延迟任务 / 线程局部变量threading.local()很有用——同一个变量名每个线程各存一份Session、trace_id、当前用户上下文、DB 连接都能塞进去requests Session 每个线程一个比疯狂 new Session 好。三、使用场景什么时候上多线程适合网络 I/O爬虫请求 URL、调外部 API/LLM API、访问 HTTP/RPC/对象存储、批量发 webhook磁盘/文件 I/O批量读日志、复制文件、读 CSV/图片/音频、写多文件数据库/消息等待SQLAlchemy/psycopg2/pymysql 执行慢 SQL 时线程等 socket 返回消费 Kafka/RabbitMQ 等 I/OGUI/后台任务CLI 后台刷进度、桌面程序后台下载、Web 服务里给同步库加个小线程池包同步阻塞库有大量同步库requests、selenium、旧 SDK不想全改 asyncThreadPoolExecutor 是最小改动方案这里有个实测依据有人测 500 个并发 API 调用asyncio 和 threading吞吐几乎一样都等网络但 500 协程内存 0.8MB500 线程 14.2MB——I/O 天花板一样开销 asyncio 更小。不适合 / 要换思路纯 Python CPU 密集图像像素循环、大 JSON 纯 Python 解析、密码学纯 Python、斐波那契/素数/压缩——GIL 在多线程≈不加速甚至更慢。官方说想用多核用 multiprocessing/ProcessPoolExecutor实测线程池 100 个 fib 和串行差不多ProcessPoolExecutor 才 4 倍。无限 max_workers线程不是免费栈/调度/上下文切换/连接池耗尽都要钱。爬虫别开 1000 workers常 5/10/20/32/CPU×2 起步再压测。CPU 重的同步函数在 asyncio 里裸跑asyncio 单线程协作time.sleep(1)或重 CPU 会卡全事件循环得run_in_executor丢线程池或 ProcessPoolExecutor 甩 CPU。四、threading vs asynciothreading / ThreadPoolExecutor OS 内核抢占式多线程操作系统切线程阻塞requests.get()/time.sleep()/conn.execute()时线程睡、GIL 释放、别人上。好处——同步代码/requests/Selenium/旧 SDK 直接能用坏处——线程内存大、锁复杂、debug 难、万级并发撑不住。asyncio 单线程协作式协程一个线程遇到await主动让出事件循环。好处——万级连接/低内存/高 I/O 吞吐aiohttp/async DB driver 时比线程池快还省内存坏处——代码要 async 全家桶忘了 await/在事件循环跑重 CPU 就全卡requests 不能直接 await。官方 threading 也点明asyncio 是不用多个 OS 线程也能任务级并发的替代路径实际 benchmark10k HTTP 请求 asyncio 12.4s、threading 23.1sasyncio 内存还低——但前提是 asyncio aiohttp 这种真异步栈。五、性能测试下面三个 micro-benchmark 把纯 I/O / 纯 CPU / 混合三种脸孔全照出来。Benchmark 1I/O 密集——asyncio 赢在内存和调度# bench_io_sync.pyimporttimeimportrequests URLhttps://httpbin.org/delay/0.1N100deffetch(i):rrequests.get(URL,timeout5)# print(f{i} {r.status_code})if__name____main__:t0time.perf_counter()foriinrange(N):fetch(i)print(sync,time.perf_counter()-t0)# bench_io_thread.pyimporttimefromconcurrent.futuresimportThreadPoolExecutorimportrequests URLhttps://httpbin.org/delay/0.1N100deffetch(i):rrequests.get(URL,timeout5)returnr.status_codeif__name____main__:t0time.perf_counter()withThreadPoolExecutor(max_workers20)aspool:list(pool.map(fetch,range(N)))print(thread,time.perf_counter()-t0)# bench_io_async.pyimporttimeimportasyncioimportaiohttp URLhttps://httpbin.org/delay/0.1N100asyncdeffetch(session,i):asyncwithsession.get(URL,timeoutaiohttp.ClientTimeout(total5))asr:returnr.statusasyncdefmain():asyncwithaiohttp.ClientSession()assession:tasks[fetch(session,i)foriinrange(N)]awaitasyncio.gather(*tasks)if__name____main__:t0time.perf_counter()asyncio.run(main())print(async,time.perf_counter()-t0)预期sync ≈ N×0.1s/很小并发 ≈ 10s 级thread 20 workers ≈ 10s/20 ≈ 0.51s 级asyncaiohttp 最快/最省内存。asyncio 赢在没 OS 线程栈、没 GIL 调度焦虑、一个事件循环调度万协程threading(requests) 赢在代码不用改 async、库生态同步。实测 asyncio 10k 请求 12.4s、threading 23.1s、内存差距巨大。Benchmark 2CPU 密集——两者都跪ProcessPool 赢# bench_cpu.pyimporttimefromconcurrent.futuresimportThreadPoolExecutor,ProcessPoolExecutordefcpu_task(n):x0foriinrange(n):xi*ireturnx N20WORK5_000_000if__name____main__:t0time.perf_counter()foriinrange(N):cpu_task(WORK)print(sync,time.perf_counter()-t0)t0time.perf_counter()withThreadPoolExecutor(max_workers4)asp:list(p.map(cpu_task,[WORK]*N))print(thread,time.perf_counter()-t0)t0time.perf_counter()withProcessPoolExecutor(max_workers4)asp:list(p.map(cpu_task,[WORK]*N))print(process,time.perf_counter()-t0)预期sync 和 thread 差不多thread 可能还慢点GIL 抢/上下文切process 接近 4 核 4×。这就是 GILCPU 活不怎么释 GIL多 OS 线程也一个时间跑 Python 字节码。官方推荐 CPU 多核用 ProcessPoolExecutor实测 fib/sha/纯算线程帮不上。Benchmark 3混合——async ProcessPool / thread ProcessPool真实后端常这样先 HTTP 拿 JSON再 JSON loads再算特征/签名/统计。async 里 CPU 重会卡事件循环threading 里 CPU 重也抢 GIL正确是async/thread 管 I/O 编排ProcessPoolExecutor 管重算# 思想async 管 I/O重 CPU 甩 ProcessPoolimportasynciofromconcurrent.futuresimportProcessPoolExecutordefheavy_cpu(data):# 假设很重returnsum(x*xforxindata)asyncdefmain():loopasyncio.get_running_loop()withProcessPoolExecutor()aspp:rawb{vals:[1,2,3]}# 假数据datalist(range(5_000_000))rawaitloop.run_in_executor(pp,heavy_cpu,data)print(r)asyncio.run(main())CPU 进事件循环 → asyncio 慢threading 略好OS 抢占multiprocessing true parallel 最稳。六、选型速查表场景推荐已有 requests/Selenium/旧 SDK/同步 DB driverThreadPoolExecutor新项目、aiohttp/asyncpg/aiomysql、万连接聊天 WSasyncio纯 Python 大计算ProcessPoolExecutor批量图片下载PIL 压缩下载线程池/进程池PIL 若 C 释 GIL 好仍 Python 循环仍小心Web 框架同步 handler 调慢 SQL/HTTP小 ThreadPoolExecutor长期高并发考虑 async frameworkGUI 后台任务threading Queue/Signal超多小 I/O task10k 连接asyncio 内存赢threading 能扛但内存大一句话拍板同步阻塞库一大堆 → ThreadPoolExecutor能从 socket 到 DB 全 async → asyncioCPU 重 → ProcessPoolExecutorasync I/O CPU 重 → asyncio run_in_executor/ProcessPoolExecutor。七、真实项目案例批量网页采集 结构化入库系统需求公司要爬 10 万个商品页/论文页/房源页拿 HTML → 解析标题/价格/摘要 → 存 PostgreSQL/CSV/S3 → 失败重试/限速/监控/优雅退出。为什么这事多线程天然对90% 时间在 DNS TCP server wait read socket write DB wait纯 Python 解析占比小requests 阻塞但 I/O 时释 GILThreadPoolExecutor 一把梭。生产级架构URL Producer │ ├── fetcher threads (download html) │ ↓ put raw_html into parse_queue ├── parser threads (bs4/正则/抽取) │ ↓ put record into db_queue └── writer threads (batch INSERT / CSV rotate / S3 upload) metrics: success/fail/429/timeout/qsize stop_event: CtrlC → graceful shutdown rate_limiter: Semaphore / token bucket per_thread Session: threading.local()比一个函数 fetchparsesave 全塞线程池强在哪职责拆开下载慢归下载、解析吃 CPU 归解析、写库归写库队列削峰某站 429 时 fetcher 堵、parser/writer 还能清库存整体背压可控。关键代码骨架importsysimporttimeimportjsonimportthreadingfromqueueimportQueue,Emptyfromconcurrent.futuresimportThreadPoolExecutor,as_completedimportrequests stop_eventthreading.Event()metrics{submitted:0,success:0,fail:0,bytes:0,}metrics_lockthreading.Lock()# 每个线程一个 Session复用 TCP 连接避免全局 Session 并发争用_session_localthreading.local()defget_session():ifnothasattr(_session_local,s):srequests.Session()s.headers.update({User-Agent:AcmeBot/1.0})_session_local.ssreturn_session_local.sdeffetch_url(url,timeout8):sget_session()resps.get(url,timeouttimeout)resp.raise_for_status()returnurl,resp.text,len(resp.content)这里threading.local()给每线程 Session——requests Session 复用连接很重要但多线程共享一个大 Session 会引连接池/状态争用实务 per-thread Session 更稳。限流Semaphore 当粗糙令牌桶importrandomclassRateLimiter:def__init__(self,max_parallel10,base_wait0.2):self.semthreading.Semaphore(max_parallel)self.base_waitbase_waitdefacquire(self):self.sem.acquire()time.sleep(self.base_wait*random.uniform(0.8,1.2))defrelease(self):self.sem.release()实际更狠得上令牌桶/漏桶/域名级限速/robots.txt/代理轮换/指数退避——爬虫第一死因不是不够快是太快被 403/429/IP封/法务找。主采集器ThreadPoolExecutor as_completed stop_eventdefcrawl(urls,max_workers12,timeout8,max_retry3):withThreadPoolExecutor(max_workersmax_workers,thread_name_prefixcrawler)aspool:# 控制已提交但未终任务别无限膨胀active0MAX_INFLIGHTmax_workers*3futures{}defsubmit_one(u,retry0):nonlocalactive futpool.submit(fetch_url,u,timeout)futures[fut](u,retry)active1foruinurls:ifstop_event.is_set():break# 简单背压别无限 submitwhileactiveMAX_INFLIGHT:time.sleep(0.01)ifstop_event.is_set():breaksubmit_one(u)forfutinas_completed(futures):url,retryfutures[fut]active-1try:url,html,sizefut.result()# 这里接 parser / db_queue.put withmetrics_lock:metrics[success]1metrics[bytes]sizeprint(fOK{url}{size})exceptExceptionase:ifretrymax_retryandnotstop_event.is_set():print(fRETRY{url}{retry1}{e})submit_one(url,retry1)else:withmetrics_lock:metrics[fail]1print(fFAIL{url}{e})ifstop_event.is_set():# 不再新提交等已提交完或进程退出passcrawl([fhttps://example.com/{i}foriinrange(100)],max_workers10)这段有 4 个工程细节as_completed先完先处理MAX_INFLIGHT防队列爆/内存涨max_retrystop_event做优雅退出metrics_lock保护共享计数。concurrent.futures.Future和asyncio.Future不是一回事官方提醒别混。graceful shutdownCtrlC 别烂尾importsignaldefhandle_sigint(sig,frame):print(\nShutdown requested, finish in-flight...)stop_event.set()signal.signal(signal.SIGINT,handle_sigint)signal.signal(signal.SIGTERM,handle_sigint)生产再加print metrics、flush queues、csv writer close、DB conn commit、prometheus push gateway、log error urls、下次断点续传 URL dedup(redis/md5/PG 主键)。八、这个项目 max_workers 设多少别信线程越多越快。经验值场景建议轻量 HTTP API / LLM 小包1032大文件下载48带宽/磁盘先满Selenium/Playwright 浏览器自动化24/机器内存DB 写 PG MySQLDB 连接池大小别超 max_connections同站爬虫612 域名限速别 DDoS 人家CPU 解析重workers↓ ProcessPoolExecutor 给解析/指纹/压缩concurrent.futures早期max_workersNone对 ThreadPoolExecutor 默认cpu * 5——承认它常用于 I/O overlap 而非 CPU但实际还得按目标站/DB/内存/连接池调。九、常见坑大全requests 能线程安全能用但 Session/Adapter 要懂Per-thread Session 最稳共享池要测。max_workers 1000 很蠢线程栈/调度/连接池/文件描述符/GC 全炸asyncio 万连接轻、1000 OS 线程就重。list(pool.map)顺序等要谁先好先处理用as_completed。Future 互等互锁A Future 等 B、B 等 A、线程池空 → deadlock1 worker 里result()另一 Future → deadlock官方 concurrent.futures 写了。asyncio 里裸requests.get()是犯罪它阻塞事件循环得asyncio.to_thread/run_in_executor或换 aiohttp。Thread dump JSON/CSV/IPC 也要锁print 不一定 atom写同一 csv writer 多条线程要 Lock或 queue → 单 writer。CPU 3.13 free-threading 别兴奋太早3.13 free-threaded build 可真并行但非默认生态 C 扩展未必全 thread-safe生产先测。十、最终选型口诀新项目、全异步栈、万连接 IM/WS/网关/微服务 client → asyncio旧代码/requests/Selenium/SDK/短平快脚本 → ThreadPoolExecutorCPU 纯算/加密/压缩/科学算 → ProcessPoolExecutorasync event loop 里重 CPU → run_in_executor/ProcessPoolExecutor。落到这个爬虫案例短期上线ThreadPoolExecutor per-thread Session as_completed retry Semaphore Queue pipeline长期平台化、自研 async adapter、aiohttp/async DB/万任务 → 迁 asyncio解析模型推理重 → asyncio 编排 ProcessPoolExecutor 跑 torch/onnx/numpy pipeline。