第一次写并发脚本的人多半是从等得太久开始的。我当时接了个批量下载任务一千多个文件单线程跑要三个多小时改成线程池之后不到十分钟跑完。Python 里的并发话题绕不开线程网上关于它和 GIL 的争论也比比皆是但真正动手做过几个项目就会明白线程不是不能用是要用对地方。这篇文章我从 Python 线程的基本用法讲起重点放在 GIL 到底影响什么、锁怎么用才不会写出死锁、线程之间怎么通信、以及工程中最常用的线程池配置。适合想把脚本速度提上来、又不想一上来就啃异步那套复杂写法的读者爬虫、批量接口调用、数据采集、文件批处理这类等待型任务基本都是线程的用武之地。1. 谈Python线程前先搞懂GIL这个老冤家很多初学者一听到 Python 线程就愁眉苦脸——不就是有 GIL 嘛线程根本没用。这话对了一半但更重要的是理解 GIL 到底锁住了什么、什么时候锁不住。GIL 全称 Global Interpreter Lock是 CPython 解释器层面的一个互斥锁它保证同一时刻只有一个线程在执行 Python 字节码。也就是说用 CPython 默认实现的 Python无论你开多少个线程CPU 核心们在执行 Python 代码时永远只有一个人在干活。那为什么会有 GIL核心原因在内存管理。CPython 的对象用引用计数来管理生命周期每个对象被引用和解除引用都会更新计数。如果多个线程同时操作同一个对象的引用计数就必须保证计数操作是原子的否则一个线程刚减完了计数另一个线程又读了旧值对象可能被提前回收程序直接崩溃。给每个对象都加一把细粒度锁当然也可以但锁的开销巨大、死锁路径复杂。GIL 是一种简单粗暴但极其有效的方案让每个线程每执行一小段时间就切换切换间隙由解释器统一管理这样引用计数天然不会被并发破坏。但要注意GIL 锁的是执行 Python 字节码的资格而不是整个进程。线程做 I/O 等待的时候比如等网络响应、等磁盘读取、等多线程里的 queue.get()这些操作会主动释放 GIL让别的线程上桌干活。这正是 Python 线程真正价值所在的场景——I/O 密集任务。1.1 GIL锁的是执行权不是I/O等待我用一个爬虫场景解释最直观。假设你要下载一万个网页单线程跑是这个流程发起请求 → 等服务器响应这一等可能是几百毫秒→ 收到数据 → 处理 → 再发下一个请求。整个时间段里CPU 大部分时间是空转的就干等着网络。用线程的话线程 A 在等响应时GIL 被释放线程 B 就可以趁机发自己的请求B 也等响应时线程 C 又顶上。这样网络等待的时间被多线程折叠了总耗时从所有响应时间相加变成单个响应时间 × 总数 ÷ 线程数级别提升非常可观。这里有个常见误区很多人以为加了线程print() 之类的日志输出也能并行加速。其实 print、字符串拼接、数值计算这些纯 Python 操作照样要抢 GIL线程多反而因为切换开销变慢。真正被加速的是那些等待占比极高的部分。1.2 实测一个计算任务和一个IO任务的区别我写个简单实验大家可以在自己电脑上复现。先定义一个 CPU 密集的累加函数import threading import time def cpu_heavy(n): total 0 for i in range(n): total i return total # 单线程跑两次 start time.perf_counter() cpu_heavy(30_000_000) cpu_heavy(30_000_000) print(f单线程耗时: {time.perf_counter() - start:.2f}s) # 两个线程各跑一次 threads [ threading.Thread(targetcpu_heavy, args(30_000_000,)), threading.Thread(targetcpu_heavy, args(30_000_000,)), ] start time.perf_counter() for t in threads: t.start() for t in threads: t.join() print(f双线程耗时: {time.perf_counter() - start:.2f}s)在我机器上单线程大约 1.2 秒双线程反而要 1.5 秒以上。原因就是两个线程来回抢 GIL频繁切换带来了额外开销。再换成模拟 I/O 等待def io_sleep(seconds): time.sleep(seconds) # sleep 会释放GIL threads [threading.Thread(targetio_sleep, args(2,)) for _ in range(10)] start time.perf_counter() for t in threads: t.start() for t in threads: t.join() print(f10个sleep任务双线程? - 实际耗时: {time.perf_counter() - start:.2f}s)10 个都 sleep 2 秒的任务单线程要做 20 秒改成 10 个线程并发实测大约 2 秒出头。这就是 GIL 的关键认知它阻止的是 CPU 并行不是 I/O 并发。顺带提一句Python 3.13 开始出现了 no-GIL 的实验性构建即自由线程方向可以让纯 Python 多线程真正并行利用多核。这是未来趋势但眼下绝大多数生产环境跑的仍然是默认 GIL 的 CPython所以按 GIL 存在的模型去写代码依然是当前最稳妥的选择。提示判断一个任务适不适合用线程先问一句它主要卡在计算还是卡在等待。卡在等待线程是好帮手卡在计算线程是无用功甚至帮倒忙。2. 线程的基本用法与那些文档里不写的坑threading 模块的 API 看起来很简单start 一个线程、join 等待结束但真用起来坑不少。我按实际踩过的顺序一个一个说。2.1 最简单的线程写法与start/join的讲究最基础的用法是传给 target 一个函数import threading def work(name): print(f{name} 开始干活) t threading.Thread(targetwork, args(worker-1,)) t.start() # 启动线程 t.join() # 主线程等待该线程结束注意args必须是元组只有一个参数也要写成(worker-1,)少了逗号就变成了字符串。还有一个细节线程对象的start()只能调用一次第二次调用会抛RuntimeError表示线程已经启动过。另外千万不要直接调t.run()——run()是线程函数体手动调用就变成普通函数调用和不开线程没有区别start()才会让操作系统真正调度它。join()是很多人忽略的方法。主线程跑完了如果还有未结束的子线程Python 解释器会等所有非 daemon线程退出后再结束进程。但如果你调用了join(timeout5)主线程最多等它 5 秒超时后继续往下走。这个特性在做批量任务时很有用任务卡住不能永远等着设置超时并记录超时任务后续重试。2.2 daemon线程用了就停不下来的坑daemon守护线程是这个模块里最容易被误解的概念。它不代表后台运行这么简单。看下面这个例子import threading import time def background(): while True: time.sleep(1) print(还在跑...) t threading.Thread(targetbackground) t.start() print(主线程结束)由于background()是死循环主线程 print 完并不会退出进程因为还有一个非 daemon 线程活着它会一直打印。如果不想让程序被这个后台任务拖着不退出可以在创建线程时加daemonTruet threading.Thread(targetbackground, daemonTrue)加了 daemonTrue 之后主线程执行完毕整个进程直接退出不管子线程跑到一半还是死循环统统终止。听起来很方便但坑在于daemon 线程里如果正在写文件、正写一半数据库事务、正带着锁处理共享数据进程一退出这些操作全部中断数据可能损坏。我见过同事有一个日志组件放在 daemon 线程里主程序一退出最后几条日志永远写不进去查了半天才知道是 daemon 还没 flush 就被杀掉了。经验是心跳检测、后台监控、定期清理这类丢了也无所谓的脚本才用 daemonTrue涉及持久化、任务结果收尾的用普通线程并在最后显式 join。2.3 线程函数里如何拿回返回值这是个困惑了很多人的点线程函数是并发执行的没办法像普通函数一样result work()直接拿结果。最原始的办法是把结果塞进一个共享容器比如 listresults [] def work(): results.append(我是结果) t threading.Thread(targetwork) t.start() t.join() print(results[0])但这引入了一个隐患多个线程同时 append 时虽然 CPython 里list.append一般是原子的但你无法保证所有操作都安全而且一旦还要读取中间状态就会面临线程安全问题。更稳的做法是后面要讲的queue.Queue或者直接用concurrent.futures.ThreadPoolExecutor它的submit()会返回一个Future对象future.result()就是线程函数的返回值。所以我的建议是写新代码优先用线程池裸 Thread 多用于需要真正自定义线程生命周期或长时间驻留的后台任务。3. 线程安全锁、原子操作和看不见的脏数据线程之间共享全局变量是 Python 并发最经典的翻车现场。不用实际跑大型项目下面这段代码就足够说明问题import threading counter 0 N 100_000 def increment(): global counter for _ in range(N): counter 1 t1 threading.Thread(targetincrement) t2 threading.Thread(targetincrement) t1.start(); t2.start() t1.join(); t2.join() print(期望值:, 2 * N, 实际值:, counter)我跑过很多次实际值经常在 10 万到 20 万之间浮动很少等于 20 万。按直觉想GIL 不是保证同一时刻只有一个线程执行字节码吗为什么还会丢更新3.1 为什么单纯a 1会丢数据即使GIL存在关键在于counter 1不是一个字节码指令而是好几条字节码的组合。大致过程是读取 counter 当前值、加 1、把新值写回。GIL 保证的是一条字节码执行期间不会被切换但不能保证一个 Python 语句的完整执行过程不被切换。假设线程 A 读到了 counter 100正要加 1 的时候GIL 切换到了线程 BB 也读到 counter 100加 1 后写回 101这时 GIL 切回 AA 继续执行它的加 1把 101 写成 101。两次加 1只加了 1。这种竞态条件在并发编程里非常典型不是 Python 特有只是 GIL 让很多人误以为 Python 里不会有这个问题。所以结论是GIL 只能保护个别原子操作比如 list.append、dict 的单次赋值不能保护多步的读写组合。想要数据不出错必须自己加锁或者用queue.Queue这类内部已加锁的容器。3.2 互斥锁和可重入锁的选用Python 里最简单的锁是threading.Lock。用法很直白lock threading.Lock() def safe_increment(): global counter for _ in range(N): with lock: counter 1用with lock:是推荐写法因为无论如何中间抛不抛异常锁都会正确释放。手动写lock.acquire()/lock.release()一旦漏了 release别的线程永远卡死排查起来非常痛苦。threading.Lock是普通互斥锁同一个线程不能连续acquire两次会死锁。比如一个函数内部加了锁又调用另一个也加了同一把锁的函数。这种场景用threading.RLock可重入锁就没事同一个线程可以重复获取计数器会记录层级每次 release 减一减到 0 才真正释放。我处理递归结构或者封装多层函数时几乎无脑选 RLock。代价是 RLock 稍慢一点但工程上这差距几乎可以忽略。选择锁的粒度也要留意。锁住的代码块太大比如整个循环 all-in-lock那并发等于没有跟单线程串行没区别锁住太小比如只锁一次加一循环外老切换效果也不好。核心原则是锁只保护真正会被并发修改的共享状态不要把无关联的耗时操作也包进去。3.3 死锁怎么复现、怎么排查、怎么预防死锁是锁方案里最头疼的问题。最经典的多锁死锁场景长这样线程 A 先拿锁 1、再拿锁 2线程 B 先拿锁 2、再拿锁 1。两个线程各自拿到第一把锁谁都不肯放手互相等着对方手里的另一把锁于是双双卡死。lock1 threading.Lock() lock2 threading.Lock() def thread_a(): with lock1: time.sleep(0.1) with lock2: pass def thread_b(): with lock2: time.sleep(0.1) with lock1: pass这个程序跑起来就再也结束不了。排查方法是我在实际项目里最常用的套路死锁时程序不会崩只是卡住。先用py-spy dump --pid 进程ID查看每个线程阻塞在哪一行能直接看到线程 A 停在lock2.acquire()附近、线程 B 停在lock1.acquire()附近一目了然。没有 py-spy 的话用faulthandler.dump_traceback_later(10)也能在 10 秒后打印线程堆栈直接定位卡住的代码行。预防死锁的经验有三个一是全局给锁编号所有代码按规定顺序申请比如先锁 1 再锁 2绝不允许反过来二是尽量一个任务只持有一把锁三是用acquire(timeout...)做兜底超时拿不到锁就回滚重试或报错至少不会无限挂起。实际项目里死锁大多是多个锁的获取顺序不一致造成的统一顺序能规避绝大多数问题。注意加锁不是越多越好加锁太多会让并发性能暴跌也会显著提高死锁概率。能用一个队列传递数据就别定义三把锁保护三个列表。4. 线程间通信队列、事件和条件变量线程之间传数据最反直觉的一点是多个线程共享同一个全局变量并手动加锁这个方案虽然可行但极易出错。更好的思路是用现成的线程安全容器其中最推荐queue.Queue。4.1 用Queue替代手工加锁最推荐的传参方式queue.Queue是 Python 标准库里为多线程设计的线程安全队列内部自带锁。它的 get() 在没有数据时会阻塞当前线程直到有数据进来put() 在队列满时也会阻塞。这天然实现了生产者-消费者模式import queue import threading import time def worker(q): while True: item q.get() if item is None: # 哨兵值收到就退出 print(线程退出) break print(f处理: {item}) time.sleep(0.1) task_queue queue.Queue(maxsize20) threads [threading.Thread(targetworker, args(task_queue,)) for _ in range(4)] for t in threads: t.start() for i in range(20): task_queue.put(i) # 给每个线程发一个 None 让它退出 for _ in threads: task_queue.put(None) for t in threads: t.join()这里的None是哨兵值作用是通知线程任务完结、可以退出。设计上发送哨兵值的次数必须等于工作线程数否则有些线程永远阻塞在q.get()。还有一个坑q.get()默认无限阻塞如果任务生产方因为异常没能 put 足够数据所有工作线程都会挂死。此时建议用q.get(timeout5)配合queue.Empty异常处理。你可能会想直接用全局 list 加锁不是也行可以但 list 要自己管理锁、判断空、处理边界代码写多了容易漏。queue.Queue还额外支持task_done()/join()机制可以在主线程等待队列里所有任务都处理完不需要手动统计完成数。4.2 Event和Condition掌控线程的执行节奏除了传数据线程之间还经常要发信号。最常用的信号组件是threading.Event——一个内部布尔标志配合wait()使用。典型场景是主线程准备好资源后通知所有工作线程开始干活。import threading import time start_event threading.Event() def worker(worker_id): print(f线程 {worker_id} 已就绪等待开始...) start_event.wait() # 阻塞直到event被set print(f线程 {worker_id} 开始工作) threads [threading.Thread(targetworker, args(i,)) for i in range(3)] for t in threads: t.start() time.sleep(1) start_event.set() # 放行所有等待线程 for t in threads: t.join()Event 的语义很清晰set()放行clear()重置wait(timeout...)可以带超时。多个线程同时wait()时set()会一次性全部放行很适合做栅栏式启动。threading.Condition是更底层的条件变量配合锁使用支持wait()/notify()/notify_all()可以实现等条件满足再继续的复杂逻辑。说句实在话日常工程中用queue.Queue和Event已经覆盖绝大部分场景了Condition 更多用于阅读第三方源码或性能敏感度极高的场景。知道它存在、了解它用途即可。4.3 threading.local让线程拥有私有空间多线程共享数据要加锁但有些数据偏偏不想共享——比如每个线程独立的数据库连接、HTTP session、日志上下文。这时候用threading.local()创建一个线程私有存储每个线程访问它时得到的都是自己那一份import threading local_data threading.local() def worker(worker_id): local_data.session_id worker_id print(f线程 {worker_id} 的 session_id 是 {local_data.session_id}) threads [threading.Thread(targetworker, args(i,)) for i in range(3)] for t in threads: t.start() for t in threads: t.join()local_data.session_id在不同线程里互不干扰完全不需要加锁。这个设计给我的好处是避免把 session 作为参数在函数栈里传来传去特别适合装饰器、中间件这类场景。但记住一个坑线程池里的线程会复用上一个任务的local_data遗留值可能被下一个任务读到。所以任务收尾时最好显式清理比如local_data.session_id None避免脏数据。5. 线程池控制并发量才是工程的关键前面说的都是裸线程但真实项目里最常用的其实是线程池。大家下载数据、批量调接口时至少八成的并发需求用concurrent.futures.ThreadPoolExecutor就够了。5.1 从裸Thread到线程池工作效率差在哪每处理一个任务就创建一个线程任务多时线程数量失控系统调度开销巨大还容易把文件描述符打满。线程池解决两件事一是复用固定数量的线程避免频繁创建销毁二是通过限制线程数来控制并发量避免把目标服务打挂。最基础的用法from concurrent.futures import ThreadPoolExecutor def download(url): # 模拟下载逻辑 return url urls [fhttps://example.com/page/{i} for i in range(100)] with ThreadPoolExecutor(max_workers8) as executor: results list(executor.map(download, urls))executor.map()很方便它会按传入顺序返回结果适合任务之间无依赖、只需要收集所有结果的场景。with语句会在代码块结束时调用shutdown(waitTrue)等待所有线程执行完再退出比裸线程手动管理 join 干净得多。如果希望任务完成一个处理一个而不是傻等所有结果返回可以用submit()拿到Futurewith ThreadPoolExecutor(max_workers8) as executor: futures [executor.submit(download, url) for url in urls] for future in futures: print(future.result())5.2 max_workers到底设多少公式与实测经验线程池最核心的参数就是max_workers。很多教程给公式CPU 核心数的 2 倍 1这个公式对 CPU 密集并不成立——至少对 Python 线程而言CPU 密集任务堆线程只会更慢前面 GIL 实验已经是证据。我的经验是分场景讨论I/O 密集网络请求、文件读写为主线程数可以远大于核心数。网络请求场景里我把 max_workers 设为 16~64 都试过。一个批量下载 1000 个文件的脚本32 线程比 64 线程好因为线程太多反而在连接建立和排队上消耗资源而且目标服务器也不愿意一次性接收这么多并发。混合场景少量计算 大量等待先评估等待比例。等待占比高的按 I/O 密集配计算占比高的考虑换成进程池。目标服务并发限制如果调用第三方 API对方文档写明了 QPS 或并发上限那么线程池数量必须小于等于该限制否则会被限流或封禁。我常用的思路是初始设成min(32, os.cpu_count() * 5)然后跑小批量测试看平均耗时和错误率逐步增大或减小。凡是涉及外部服务的都要留一定余量不要压满。线程池不是越大越好关键瓶颈往往不在 Python而在下游系统能扛多少并发。5.3 as_completed处理有先有后的结果executor.map()有个不便之处它按输入顺序返回结果前一个任务没完成后面的结果就卡在生成器里拿不到。如果前一个任务是全池子里最慢的那所有更快完成的任务结果都得等它。这时候用as_completed更合理from concurrent.futures import ThreadPoolExecutor, as_completed with ThreadPoolExecutor(max_workers8) as executor: futures {executor.submit(download, url): url for url in urls} for future in as_completed(futures): url futures[future] try: result future.result() print(f完成: {url} - {result}) except Exception as e: print(f失败: {url} - {e})as_completed返回一个迭代器哪个Future先完成就先 yield 哪个。这样你可以尽快处理成功结果、记录失败任务对大批量请求尤其有用。另外一个非常重要的细节future.result()会把线程函数内部抛出的异常原样抛出来。线程池不会吞异常如果不 catch主程序可能在某个未来时刻突然崩溃——这个崩溃看起来还很莫名其妙。所以一定在result()外面包try/except并把任务上下文一起打印便于定位是哪条数据出了问题。6. Python并发三兄弟线程、进程和异步怎么选聊到这里很多读者应该已经意识到线程并不是 Python 并发里唯一的答案。至少要搞明白另外两条路多进程 multiprocessing 和异步 asyncio。不少人一上来就纠结选哪个其实判断标准没那么玄乎。6.1 三类并发的本质差异时间片、进程隔离、事件循环我把它们放在一起对比看起来更清楚维度线程 (threading)进程 (multiprocessing)异步 (asyncio)调度方式操作系统抢占式调度操作系统调度各进程独立单线程事件循环协作式调度内存共享同进程共享全局变量进程间默认不共享需特殊机制传递同线程共享但不能并行是否受GIL影响受纯计算无法并行不受多进程可多核并行受但协程切换不依赖GIL争抢适合场景I/O密集、短任务CPU密集、计算量大海量I/O等待、高并发长连接数据传递难度加锁或queue简单进程间通信较麻烦Queue/Pipe原生await/返回天然简单上手成本低中中高需要适应async/await记忆方法也很简单操作系统在不同线程和进程之间来回切换线程片线程共享同一个解释器和进程内存进程是每份代码各开炉灶而 asyncio 从头到尾就一个线程靠代码自己声明我在这里要等一会儿让别人先去跑来切换。6.2 选型判断看你的任务卡在CPU还是卡在IO卡在哪永远是第一判断标准。任务跑起来CPU 占用一直不到 30%而耗时大多数时间在等网络、等磁盘那就是 I/O 密集线程和 asyncio 都合适。任务一跑 CPU 占用直接冲到 100%比如图片特征提取、复杂计算、本地 OCR 识别那是 CPU 密集在 Python 里就要考虑多进程。举几个我遇到的真实场景批量抓取网页I/O 密集用 ThreadPoolExecutor 或 asyncio线程池实现更快心智负担低。大规模数值计算、量化策略回测CPU 密集用 multiprocessing.Pool让多个核心同时算。本地 OCR 识别一堆图片别用多线程Python 的 OCR 组件通常吃满 CPU线程并发受 GIL 拖累进程池才是正解。AI Agent 调用外部模型接口大多数时间在等模型响应本质是 I/O 密集asyncio 或线程池都不错。此时要注意的是同时等待多少请求而非让多少 CPU 在算并发模型选型反而要聚焦在下游接口限流和超时管理。判断的简单实用技巧把并发线程数加到 2 个跑一遍对比单线程耗时。如果明显提升说明任务里面确有大量等待如果几乎没变化甚至变慢那就是计算在抢 GIL换进程池。这个实验 1 分钟就能做完比我讲一堆理论都直观。6.3 混合并发的一个现实案例爬虫解析实际项目中很少有纯粹的 I/O 或纯 CPU更多是混合。比如爬虫下载网页是 I/O 密集解析 HTML 提取数据是偏 CPU 的活。我的做法是拆分——下载阶段用线程池解析阶段放进进程池。流程上线程池把下载好的 HTML 塞进 queue进程池从 queue 拿内容解析两边并行跑互不拖累。代码大致是这样from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor import queue task_queue queue.Queue(maxsize200) def download(url): # 网络请求返回html return fhtml.../html def parse(html): # 解析操作CPU密集 return len(html) with ThreadPoolExecutor(max_workers16) as t_pool, ProcessPoolExecutor(max_workers4) as p_pool: # 提交下载任务 download_futures [t_pool.submit(download, url) for url in urls] # 逐个解析 for f in download_futures: html f.result() p_pool.submit(parse, html)注意这个写法里进程池的 submit 不需要等待全部线程完成下载的 Future 每完成一个就提交一个解析任务相当于边下边解析。实际跑下来比纯线程池方案还要快不少因为解析那部分 CPU 密集工作不再跟下载任务互相抢 GIL 了。代价是引入多进程后任务数量巨大时内存占用更高所以只建议在解析非常耗 CPU 的阶段做这个拆分。顺带说一句Java 社区最近热炒虚拟线程本质也是在解决线程不适合 I/O 等待的问题。Python 这边长期靠 asyncio 补位未来 no-GIL 构建若能普及线程的适用边界又会有变化。但选型逻辑始终不变先把任务类型看透再决定工具。我个人用下来的体会是线程是 Python 并发里最没必要被神化、也没必要被贬低的那一个。它解决不了 CPU 密集的并行问题但 I/O 密集场景下配合 queue.Queue 和 ThreadPoolExecutor写起来舒服运行稳定几乎能满足日常开发里八成以上的太慢了诉求。最后分享一个小技巧多线程程序卡死的时候不要瞎改代码先pip install py-spy然后py-spy dump --pid 进程PID看看每个线程到底卡在哪一行通常一眼就能看出是死锁、queue 阻塞还是下游接口 timeout。工具这种东西平时备着关键时刻真的能救命。