如果你刚接触 Python 多线程一定被各种说法绕晕过有人说它是鸡肋因为 GIL有人说它很好用爬虫全靠它。这两个说法其实都成立关键看你在什么场景里用它。我最早写爬虫的时候也被搞懵一边是教程说 Python 多线程没用一边是实际工程里 ThreadPoolExecutor 用得飞起。后来把线程、进程、协程的适用边界弄清楚以后才真正理解这句话该怎么读。这篇文章我不打算给你讲一堆教科书概念而是从 GIL 开始把 threading、queue、线程池、协程的选型思路完整捋一遍。中间会穿插我在真实项目里踩过的坑以及可以直接抄走的代码片段。适合那些已经会写 Python 基础语法、但一碰到并发就手心冒汗的读者。1. 从 GIL 开始Python 多线程到底卡在哪1.1 线程和进程同一个厨房里的厨师先理清两个最基本的概念。进程是操作系统分配资源的最小单位有自己独立的内存空间线程是进程内部的一条执行路径多个线程共享同一个进程的堆、全局变量和文件描述符。你可以把进程想成一个完整的厨房线程就是厨房里的厨师。进程之间互相看不见对方的灶台而线程之间共享同一个厨房里的所有食材和工具。这个类比很关键。因为共享线程之间交换数据不需要像进程那样走管道、共享内存或文件成本低写起来也直观。但也因为共享多个线程同时修改同一个变量时就会打架于是需要锁。Python 的多线程争议一部分来自这种共享内存模型本身就难写另一部分则来自 CPython 解释器的 GIL。还有一个容易被忽略的点线程是由操作系统调度的。所以你在线程里做time.sleep、网络请求、文件读写时线程会让出 CPU等待事件完成。这个“让出”的动作是理解多线程为什么能加速 IO 任务的关键。1.2 GIL 这把锁到底锁住了什么GIL 的全称是 Global Interpreter Lock全局解释器锁。它是 CPython 解释器里的一个互斥锁作用是保证同一时刻只有一个线程能执行 Python 字节码。为什么会有这个奇怪的设计因为 CPython 的内存管理依赖引用计数如果允许多个线程同时操作对象的引用计数计数会出现竞态条件导致对象被提前释放或永不释放。最简单粗暴的办法就是加一把全局锁让解释器核心状态始终是线程安全的。这带来的后果很直接在 CPython 里同一时刻只有一个线程能把 Python 代码跑在 CPU 上。也就是说写两个线程去循环累加变量不会因为用了多核而变快反而可能因为线程切来切去而更慢。有人因此说“Python 多线程是假的”语气夸张但结论没说错——前提是你在做 CPU 密集型计算。不过要注意GIL 不是永远持有不放手。在线程遇到阻塞型系统调用时比如读文件、网络收发、睡眠它会主动释放 GIL让其他线程有机会执行字节码。这正是多线程在 IO 场景下能发挥作用的原因。Python 3.13 之后推出了不带 GIL 的构建选项但绝大多数生产环境仍是传统 CPython所以现在学多线程依然绕不开 GIL 这个背景。1.3 多线程加速的边界IO 密集 vs CPU 密集先明确两个词。IO 密集型任务是指大部分时间在等待外部资源比如请求网页、读数据库、读写文件、等待用户输入。CPU 密集型任务是指大部分时间在计算比如循环叠加、图像滤镜、加密解密、大规模数值运算。多线程擅长前者。举个例子单线程依次发起 10 个网络请求每个请求等待 2 秒总耗时约 20 秒。如果用多线程在线程 A 等第一个请求响应时线程 B 已经把第二个请求发出去了。理想情况下10 个线程同时等待总耗时接近单个请求耗时加上一点点调度开销约 2 秒多。这个过程里 GIL 并没有拖后腿因为等待响应的线程已经释放了 GIL其他线程可以自由创建连接、解析响应。多线程不擅长后者。如果任务是把 1000 万个数加起来两个线程不光不能并行计算还要反复争抢 GIL 和切换上下文实测常常比单线程还慢。对于 CPU 密集任务正确做法是用多进程每个进程有独立解释器和独立的 GIL可以真正利用多核。顺便说一句很多初学者把多线程当成“并发唯一解”一提到慢就开线程。这个习惯不好。看到性能瓶颈先判断是等外部资源还是算得慢再决定工具。表意不清的情况下可以先跑一个最简单的单线程脚本计时再换多线程计时用数据说话。2. threading 模块实操从建线程到线程池2.1 Thread 的入坑姿势Python 标准库的 threading 模块是基础。最朴素的方式就是创建一个 Thread 对象传入 target 函数和参数然后 start。下面是一个模拟下载文件的例子import threading import time def download(url): print(f开始下载{url}) time.sleep(1) print(f下载完成{url}) t1 threading.Thread(targetdownload, args(http://example.com/1,)) t2 threading.Thread(targetdownload, args(http://example.com/2,)) t1.start() t2.start() t1.join() t2.join() print(全部完成)start 只是通知操作系统创建线程并开始执行主线程不会停下来等它。join 的意思是“主线程在这里等子线程结束”。如果忘了 join主线程打印“全部完成”时子线程可能还在跑。生产环境里我习惯给每个线程显式命名t1 threading.Thread(targetdownload, namedownload-1, args(...))打印日志时带上线程名排查问题会省很多力气。另外还有一个 daemon 参数设为 True 表示守护线程主线程退出后子线程会直接终止。守护线程适合后台心跳任务但不适合保存数据这种必须跑完的工作因为随时可能被“掐死”。2.2 锁不是万能的但共享数据离不开锁多线程最经典的坑是共享变量竞争。看这个例子counter 0 def inc(): global counter for _ in range(100000): counter 1 t1 threading.Thread(targetinc) t2 threading.Thread(targetinc) t1.start() t2.start() t1.join() t2.join() print(counter)你可能会以为结果一定是 200000但实际跑几次你会发现结果经常偏小。原因是counter 1并不是一步完成它要分三步读取当前值、加一、写回。线程 A 读到 100线程 B 也读到 100A 写回 101B 也写回 101两个线程各加一次结果却只加了 1。这就是“丢失更新”。解决办法是加锁import threading counter 0 lock threading.Lock() def inc(): global counter for _ in range(100000): with lock: counter 1with lock在进入时自动 acquire退出时自动 release比手动加锁安全得多不容易出现锁住了却忘记释放的惨案。Lock 是普通锁只能在持有锁的线程里释放RLock 是可重入锁同一线程可多次 acquire 而不会死锁适合递归或嵌套代码但使用场景有限别把 RLock 当普通锁到处用。锁的粒度很讲究。如果你把整个循环都包进去with lock: for _ in range(100000): counter 1虽然结果正确但多线程彻底退化成单线程完全失去并发意义。正确思路是锁只保护“读改写”那一小段临界区外围尽量并行。2.3 线程间通信queue 的正确打开方式共享变量加锁不是处理线程通信的首选方案更推荐使用queue.Queue。Queue 内部已经实现了锁和条件变量线程安全读端和写端解耦代码更干净。下面是一个极简的生产者消费者模型import queue import threading import time task_queue queue.Queue(maxsize10) def producer(): for i in range(20): task_queue.put(ftask-{i}) time.sleep(0.1) task_queue.put(None) def worker(): while True: task task_queue.get() if task is None: task_queue.task_done() break print(f处理 {task}) time.sleep(0.2) task_queue.task_done() threads [threading.Thread(targetworker, namefworker-{i}) for i in range(3)] for t in threads: t.start() producer() task_queue.join() for t in threads: t.join()get默认会一直阻塞直到队列里有新任务。消费者线程用while True不断拿任务拿到 None 再退出这是一种常见的结束信号。注意task_done()每处理完一个任务都要调用一次主线程的queue.join()会一直等到队列中所有任务都被标记完成。热搜里常有人问“queue 不堵塞”怎么写。Queue.get()本身就是阻塞的想让它不阻塞可以用get_nowait()但队列空时会抛queue.Empty异常必须 try 包一层。另一种方式是get(timeout3)超时后抛异常。实际工程里我不建议为了“非阻塞”而写空转轮询while True: try: task q.get_nowait() except queue.Empty: break handle(task)这种逻辑适合一次性清空队列但如果队列长期为空这个循环会疯狂空转吃满 CPU。真要边生产边消费还是让消费者线程用默认的阻塞get更省资源。2.4 用线程池替代手写线程concurrent.futures手动管理 Thread 对象适合任务数量固定、生命周期明确的场景。但如果任务数量很多频繁创建和销毁线程的开销会变得明显这时应该用线程池。《concurrent.futures.ThreadPoolExecutor 是标准库给出的答案。from concurrent.futures import ThreadPoolExecutor, as_completed import requests def fetch(url): resp requests.get(url, timeout5) return url, resp.status_code urls [ http://example.com/api/items/1, http://example.com/api/items/2, http://example.com/api/items/3, ] with ThreadPoolExecutor(max_workers8) as executor: future_map {executor.submit(fetch, url): url for url in urls} for future in as_completed(future_map): url future_map[future] try: _, status future.result() except Exception as exc: print(url, 失败, exc) else: print(url, 状态码, status)with块会在退出时等待所有线程任务结束相当于统一调用了 shutdown(waitTrue)。submit返回一个 Future 对象.result()会等待该任务完成如果任务抛出异常result 会再次抛出。用as_completed的好处是哪个任务先完成就先处理哪个不用等最慢的那个。线程池适合一批互相独立的任务比如批量检查 URL 状态、批量下载文件、批量查询数据库。你只需要关心并发数不需要关心线程创建、回收和异常处理代码维护成本直线下降。3. 多线程实战排错手册踩过的坑和解决思路3.1 打印乱序和共享变量“消失”的更新打印乱序是最容易发现、最容易误判的问题。多个线程同时在 print你会看到输出内容互相穿插比如一行英文中间插进另一行。这是因为 print 分成多次输出线程切换可能发生在中间。如果你只是想看日志用 logging 模块更合适如果想手动打印别分多次 print尽量拼成一个字符串一次性输出。共享变量“消失”的问题在 2.2 节已经演示过。这里再补充一个真实经验某数据采集器需要统计每个线程成功处理了多少条数据开发同学直接在函数外部放了几个全局变量用count 1更新。结果程序运行后统计数量永远对不上。定位时把线程数改成 1一切正常改成 8数据就丢。最后把计数逻辑改成每个线程维护自己的局部变量再汇总到队列问题才解决。用队列传递结果比共享全局变量安全得多。3.2 死锁锁顺序不一致带来的僵局死锁是最让人头疼的问题因为没有报错程序只是卡住不动。最常见的原因是两个线程各自持有一把锁又都想拿对方手里的锁。场景是这样的线程 A 先锁住锁 1然后想锁 2线程 B 先锁住锁 2然后想锁 1。A 不释放锁 1B 不释放锁 2两边都等对方释放于是永久僵住。避免死锁有几个土办法所有线程都按相同顺序获取锁比如一律先锁 1 再锁 2破坏循环等待。尽量不要嵌套锁。能用队列传递数据就别设计需要同时持有两把锁的临界区。在锁的acquire上设置超时拿不到就放弃。不过 Lock 的acquire(timeout...)返回布尔值需要自己处理失败分支。如果程序已经卡住可以按 CtrlC 让 Python 打印当前线程堆栈看看每个线程阻塞在哪个位置。这个操作很笨但有效能快速确认是否死锁。3.3 线程数量该开多少线程开少了跑得慢开多了也不一定快。每个线程都有自己的栈空间和内核资源线程数量过多时操作系统上下文切换开销会吞掉并发收益。我见过有人给线程池设 500 个 worker去请求一个只允许 20 个并发的外部接口结果大量连接超时整体比 20 个线程还慢。经验上IO 密集任务的合理并发量不能只看本机瓶颈还要看下游服务的承受能力。做法很简单从 1 个线程开始成倍往上加同时观察任务总耗时和错误率。如果加到 16 个线程后耗时不再下降错误率开始上升那这个值就是当前场景的最优并发量。不要迷信公式用数据说话。CPU 密集任务用多线程基本没救线程数设成核心数也是白搭直接考虑多进程。混合型任务则要拆解把计算部分交给进程池把 IO 等待部分交给线程池或协程。3.4 线程嵌套线程为什么我不建议热搜里有“python 线程嵌套线程”这个关键词让我想起一个合作项目里的事。当时某开发者为了让每个 worker 能同时处理多个子任务在 worker 内部又创建了新的线程结果任务一来线程数量按指数增长最后程序因为资源耗尽退出连日志都没留下几行。线程嵌套线程的难处在于生命周期控制。外层线程退出内层线程未必退出内层线程报错外层线程无法捕获你以为在池子里控制数量实际线程数完全失控。调试时看到几十个线程互相纠缠头都大。如果你确实遇到外层任务还要并发子任务的场景应当把子任务重新投递到一个独立的线程池而不是在外层线程里再开线程。这样数量可控退出逻辑也清晰。更彻底的做法是重新设计任务粒度让每个任务足够独立不要人为制造父子依赖。4. 场景实战多线程数据采集器从零搭起来4.1 需求与命令行入口为了别太抽象我模拟一个需求给定一个文本文件每行一个公开接口的 URL目标是并发请求这些接口把返回的 JSON 数据结构化后写入输出文件。这是一个很典型的 IO 密集型任务适合用多线程。入口我用 argparse 写这样命令行的--workers控制并发数--input指定输入文件--output指定输出路径方便在脚本里重复调整参数。这也是你在热搜里看到 argparse 的常见用途。import argparse import json import queue import threading import time import requests def parse_args(): p argparse.ArgumentParser(description多线程数据采集器) p.add_argument(--input, requiredTrue, help每行一个URL的输入文件) p.add_argument(--output, requiredTrue, help输出JSON文件路径) p.add_argument(--workers, typeint, default8, help并发线程数) return p.parse_args()输入文件可能是几千行单线程跑一遍可能要几分钟多线程可以把总耗时压到十几秒。这个提升在真实项目里很常见。4.2 生产者-消费者实现采集器的核心用队列做生产消费。主线程负责读文件把 URL 放入队列消费者线程从队列里取 URL请求、解析、保存结果最后主线程统一落盘。def worker(q, out_queue): while True: url q.get() if url is None: q.task_done() break try: resp requests.get(url, timeout5, headers{User-Agent: Mozilla/5.0}) resp.raise_for_status() data resp.json() out_queue.put({url: url, data: data}) except Exception as exc: out_queue.put({url: url, error: str(exc)}) finally: q.task_done() def main(): args parse_args() q queue.Queue(maxsizeargs.workers * 2) out_queue queue.Queue() with open(args.input, encodingutf-8) as f: urls [line.strip() for line in f if line.strip()] threads [] for i in range(args.workers): t threading.Thread(targetworker, args(q, out_queue), namefworker-{i}) t.start() threads.append(t) for url in urls: q.put(url) for _ in range(args.workers): q.put(None) q.join() results [] while True: try: results.append(out_queue.get_nowait()) except queue.Empty: break with open(args.output, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) if __name__ __main__: main()这里有几个细节。队列容量设置成args.workers * 2是为了防止一次性把所有 URL 都加载进队列内存不至于被几千个字符串挤爆。结束信号放了args.workers个 None每个 worker 消费一个保证所有线程都能收到退出信号。out_queue.get_nowait()在最后统一导出结果避免多个线程同时写文件的竞争。4.3 并发控制、超时和重试真实网络环境远没有代码看起来这么顺。一个接口偶尔超时另一个接口返回 500都需要处理。这里我把请求部分封装成独立函数加上指数退避重试HEADERS {User-Agent: Mozilla/5.0} def fetch_json(url, max_retries3): for attempt in range(max_retries): try: resp requests.get(url, timeout5, headersHEADERS) resp.raise_for_status() return resp.json() except Exception: if attempt max_retries - 1: raise time.sleep(0.5 * (2 ** attempt))timeout5是必须的。很多线程卡死的根源不是并发数量而是某个请求一直不返回线程就永远等在那里。加了超时最多 5 秒就抛异常重试机制才有机会介入。指数退避的核心理念是出错后先等短时间再等更长时间避免在服务端还没恢复时疯狂重试把对方打垮。0.5、1、2 秒这样的阶梯足够温和。4.4 把结果安全写盘多线程写同一个文件是最容易出问题的操作之一。如果每个线程拿到结果后直接open(...).write(...)你很可能会看到文件内容互相穿插、JSON 被截断、或者写入顺序完全随机。更稳妥的做法是像 4.2 节那样线程只负责把结果放进输出队列主线程等所有任务结束后统一写盘。单一写入者可以避免锁竞争和文件句柄混乱而且数据落盘顺序可控。如果需要边采边落盘也可以专门开一个写盘线程从输出队列取数据一条条追加到文件。但要注意追加模式下的缓冲问题避免程序崩溃丢数据。总之多线程共享状态越少事故越少。5. 多线程之外协程和多进程到底该怎么选5.1 一张表理清选择思路聊到这里你已经知道多线程不是唯一答案了。做技术选型时可以从任务类型倒推方案。任务类型推荐方案核心原因CPU 密集大量计算、循环、加密多进程避开 GIL利用多核IO 密集请求、文件、数据库多线程 / 协程等待外部资源时让出 CPU高并发网络 IO成千上万个连接协程线程切换开销大协程更轻量混合型先计算后请求再计算多进程 多线程/协程拆解任务各取所长这张表不是绝对规则而是帮助你在拿到一个模糊需求时不至于上来就写 ThreadPoolExecutor。先把任务拆开看瓶颈是计算还是 IO再决定方案。5.2 线程到协程的思维转变协程是比线程更轻量的“用户态调度”。一个线程里可以跑成千上万个协程协程之间通过 await 让出控制权。对于大量网络 IO 的场景协程比多线程更节省资源。简单感受一下 asyncio 的写法import asyncio async def fetch(session, url): print(开始请求, url) await asyncio.sleep(1) # 模拟IO等待 print(完成请求, url) return url async def main(): tasks [fetch(None, fhttp://example.com/{i}) for i in range(5)] results await asyncio.gather(*tasks) print(results) asyncio.run(main())async def定义协程函数await表示让出 CPU。当程序执行到await asyncio.sleep(1)时事件循环会去调度其他协程所以 5 个任务几乎同时进入等待同时完成。这段代码在执行效率上能达到多线程的效果而线程数只有 1。但协程有一个学习门槛你不能在异步函数里随意调用阻塞库。比如requests.get是同步阻塞的直接放进 async 函数里会卡住整个事件循环。如果项目里已经用了大量同步库多线程反而是最小改动方案。5.3 多线程和协程结合的小技巧实际工程里不会那么纯粹经常碰到“异步框架里调一个同步 SDK”的情况。比如你写了一个 asyncio 服务但某个功能必须调用一个老旧的同步客户端库直接调用会卡住事件循环。这时可以用线程池把它“外包”出去import asyncio import requests async def main(): # Python 3.9 resp await asyncio.to_thread(requests.get, http://example.com, timeout5) print(resp.status_code)或者用更低层的方式import asyncio from concurrent.futures import ThreadPoolExecutor async def main(): loop asyncio.get_running_loop() with ThreadPoolExecutor(max_workers4) as pool: result await loop.run_in_executor(pool, requests.get, http://example.com, timeout5)这种做法让你既能享受异步框架的高并发调度又能兼容同步阻塞库。线程池的默认大小可能不够最好根据自己的 IO 并发量显式指定。它是连接“异步世界”和“同步世界”的一座桥。5.4 量化回测、批量任务中的真实取舍前面说了很多技术细节最后落到热搜里的两个关键词“量化交易策略”和“结构化数据”。在量化回测中为了测试不同参数组合常常要跑几十组策略。每组策略内部有大量因子计算属于 CPU 密集用多线程会被 GIL 按住所以常见做法是multiprocessing.Pool并行跑不同参数的回测每个子进程独立计算再把结果汇集到一起。实时行情订阅则是另一个方向。行情推送是持续不断的网络 IO适合用异步或多线程接收再把收到的数据写入数据库或消息队列。数据清洗和结构化转换pandas 底层很多操作是 C 库实现并不完全受 GIL 控制所以某些情况下多线程也有价值但最稳妥的方案还是把数据按时间段分块交给多进程并行处理最后合并结果。关键不是背结论而是先压测再选择。一个任务拿到手先想想它是“等得久”还是“算得久”然后在小规模数据上分别用线程、进程、协程跑一遍谁快用谁。多线程是工具箱里的一把好锤子但别把所有问题都看成钉子。6. 最后分享一点我的体会写并发代码这几年我最大的体会是先把单线程版本跑通再考虑加速。很多人一上来就堆线程结果逻辑错误和并发错误混在一起调试难度翻倍。正确的步骤是先写一个简单、正确、能出结果的版本然后找到真正的性能瓶颈再决定要不要上多线程、多进程还是协程。另一个小技巧是多线程排错先从日志入手。给每个线程取一个可读的名字所有日志统一走 logging带上线程名打印出来。一旦程序卡住或者结果不对先看日志别瞎猜。用队列传递数据比共享变量加锁更容易写对线程池管理生命周期比手动创建线程更省心。避开嵌套线程并发代码会好写很多。最后再补一句实在话多线程提升的是系统吞吐不是单个任务的延迟。你的场景如果是“数量多、等待多”多线程是真香如果是“单个任务计算量大”老老实实去研究多进程和协程。选对工具比努力调参数重要得多。 
长河君 · 2026/10/9 23:30:55 · 阅读 1.2k Python多线程到底怎么用？从GIL到协程的并发选型实战
如果你刚接触 Python 多线程一定被各种说法绕晕过有人说它是鸡肋因为 GIL有人说它很好用爬虫全靠它。这两个说法其实都成立关键看你在什么场景里用它。我最早写爬虫的时候也被搞懵一边是教程说 Python 多线程没用一边是实际工程里 ThreadPoolExecutor 用得飞起。后来把线程、进程、协程的适用边界弄清楚以后才真正理解这句话该怎么读。这篇文章我不打算给你讲一堆教科书概念而是从 GIL 开始把 threading、queue、线程池、协程的选型思路完整捋一遍。中间会穿插我在真实项目里踩过的坑以及可以直接抄走的代码片段。适合那些已经会写 Python 基础语法、但一碰到并发就手心冒汗的读者。1. 从 GIL 开始Python 多线程到底卡在哪1.1 线程和进程同一个厨房里的厨师先理清两个最基本的概念。进程是操作系统分配资源的最小单位有自己独立的内存空间线程是进程内部的一条执行路径多个线程共享同一个进程的堆、全局变量和文件描述符。你可以把进程想成一个完整的厨房线程就是厨房里的厨师。进程之间互相看不见对方的灶台而线程之间共享同一个厨房里的所有食材和工具。这个类比很关键。因为共享线程之间交换数据不需要像进程那样走管道、共享内存或文件成本低写起来也直观。但也因为共享多个线程同时修改同一个变量时就会打架于是需要锁。Python 的多线程争议一部分来自这种共享内存模型本身就难写另一部分则来自 CPython 解释器的 GIL。还有一个容易被忽略的点线程是由操作系统调度的。所以你在线程里做time.sleep、网络请求、文件读写时线程会让出 CPU等待事件完成。这个“让出”的动作是理解多线程为什么能加速 IO 任务的关键。1.2 GIL 这把锁到底锁住了什么GIL 的全称是 Global Interpreter Lock全局解释器锁。它是 CPython 解释器里的一个互斥锁作用是保证同一时刻只有一个线程能执行 Python 字节码。为什么会有这个奇怪的设计因为 CPython 的内存管理依赖引用计数如果允许多个线程同时操作对象的引用计数计数会出现竞态条件导致对象被提前释放或永不释放。最简单粗暴的办法就是加一把全局锁让解释器核心状态始终是线程安全的。这带来的后果很直接在 CPython 里同一时刻只有一个线程能把 Python 代码跑在 CPU 上。也就是说写两个线程去循环累加变量不会因为用了多核而变快反而可能因为线程切来切去而更慢。有人因此说“Python 多线程是假的”语气夸张但结论没说错——前提是你在做 CPU 密集型计算。不过要注意GIL 不是永远持有不放手。在线程遇到阻塞型系统调用时比如读文件、网络收发、睡眠它会主动释放 GIL让其他线程有机会执行字节码。这正是多线程在 IO 场景下能发挥作用的原因。Python 3.13 之后推出了不带 GIL 的构建选项但绝大多数生产环境仍是传统 CPython所以现在学多线程依然绕不开 GIL 这个背景。1.3 多线程加速的边界IO 密集 vs CPU 密集先明确两个词。IO 密集型任务是指大部分时间在等待外部资源比如请求网页、读数据库、读写文件、等待用户输入。CPU 密集型任务是指大部分时间在计算比如循环叠加、图像滤镜、加密解密、大规模数值运算。多线程擅长前者。举个例子单线程依次发起 10 个网络请求每个请求等待 2 秒总耗时约 20 秒。如果用多线程在线程 A 等第一个请求响应时线程 B 已经把第二个请求发出去了。理想情况下10 个线程同时等待总耗时接近单个请求耗时加上一点点调度开销约 2 秒多。这个过程里 GIL 并没有拖后腿因为等待响应的线程已经释放了 GIL其他线程可以自由创建连接、解析响应。多线程不擅长后者。如果任务是把 1000 万个数加起来两个线程不光不能并行计算还要反复争抢 GIL 和切换上下文实测常常比单线程还慢。对于 CPU 密集任务正确做法是用多进程每个进程有独立解释器和独立的 GIL可以真正利用多核。顺便说一句很多初学者把多线程当成“并发唯一解”一提到慢就开线程。这个习惯不好。看到性能瓶颈先判断是等外部资源还是算得慢再决定工具。表意不清的情况下可以先跑一个最简单的单线程脚本计时再换多线程计时用数据说话。2. threading 模块实操从建线程到线程池2.1 Thread 的入坑姿势Python 标准库的 threading 模块是基础。最朴素的方式就是创建一个 Thread 对象传入 target 函数和参数然后 start。下面是一个模拟下载文件的例子import threading import time def download(url): print(f开始下载{url}) time.sleep(1) print(f下载完成{url}) t1 threading.Thread(targetdownload, args(http://example.com/1,)) t2 threading.Thread(targetdownload, args(http://example.com/2,)) t1.start() t2.start() t1.join() t2.join() print(全部完成)start 只是通知操作系统创建线程并开始执行主线程不会停下来等它。join 的意思是“主线程在这里等子线程结束”。如果忘了 join主线程打印“全部完成”时子线程可能还在跑。生产环境里我习惯给每个线程显式命名t1 threading.Thread(targetdownload, namedownload-1, args(...))打印日志时带上线程名排查问题会省很多力气。另外还有一个 daemon 参数设为 True 表示守护线程主线程退出后子线程会直接终止。守护线程适合后台心跳任务但不适合保存数据这种必须跑完的工作因为随时可能被“掐死”。2.2 锁不是万能的但共享数据离不开锁多线程最经典的坑是共享变量竞争。看这个例子counter 0 def inc(): global counter for _ in range(100000): counter 1 t1 threading.Thread(targetinc) t2 threading.Thread(targetinc) t1.start() t2.start() t1.join() t2.join() print(counter)你可能会以为结果一定是 200000但实际跑几次你会发现结果经常偏小。原因是counter 1并不是一步完成它要分三步读取当前值、加一、写回。线程 A 读到 100线程 B 也读到 100A 写回 101B 也写回 101两个线程各加一次结果却只加了 1。这就是“丢失更新”。解决办法是加锁import threading counter 0 lock threading.Lock() def inc(): global counter for _ in range(100000): with lock: counter 1with lock在进入时自动 acquire退出时自动 release比手动加锁安全得多不容易出现锁住了却忘记释放的惨案。Lock 是普通锁只能在持有锁的线程里释放RLock 是可重入锁同一线程可多次 acquire 而不会死锁适合递归或嵌套代码但使用场景有限别把 RLock 当普通锁到处用。锁的粒度很讲究。如果你把整个循环都包进去with lock: for _ in range(100000): counter 1虽然结果正确但多线程彻底退化成单线程完全失去并发意义。正确思路是锁只保护“读改写”那一小段临界区外围尽量并行。2.3 线程间通信queue 的正确打开方式共享变量加锁不是处理线程通信的首选方案更推荐使用queue.Queue。Queue 内部已经实现了锁和条件变量线程安全读端和写端解耦代码更干净。下面是一个极简的生产者消费者模型import queue import threading import time task_queue queue.Queue(maxsize10) def producer(): for i in range(20): task_queue.put(ftask-{i}) time.sleep(0.1) task_queue.put(None) def worker(): while True: task task_queue.get() if task is None: task_queue.task_done() break print(f处理 {task}) time.sleep(0.2) task_queue.task_done() threads [threading.Thread(targetworker, namefworker-{i}) for i in range(3)] for t in threads: t.start() producer() task_queue.join() for t in threads: t.join()get默认会一直阻塞直到队列里有新任务。消费者线程用while True不断拿任务拿到 None 再退出这是一种常见的结束信号。注意task_done()每处理完一个任务都要调用一次主线程的queue.join()会一直等到队列中所有任务都被标记完成。热搜里常有人问“queue 不堵塞”怎么写。Queue.get()本身就是阻塞的想让它不阻塞可以用get_nowait()但队列空时会抛queue.Empty异常必须 try 包一层。另一种方式是get(timeout3)超时后抛异常。实际工程里我不建议为了“非阻塞”而写空转轮询while True: try: task q.get_nowait() except queue.Empty: break handle(task)这种逻辑适合一次性清空队列但如果队列长期为空这个循环会疯狂空转吃满 CPU。真要边生产边消费还是让消费者线程用默认的阻塞get更省资源。2.4 用线程池替代手写线程concurrent.futures手动管理 Thread 对象适合任务数量固定、生命周期明确的场景。但如果任务数量很多频繁创建和销毁线程的开销会变得明显这时应该用线程池。《concurrent.futures.ThreadPoolExecutor 是标准库给出的答案。from concurrent.futures import ThreadPoolExecutor, as_completed import requests def fetch(url): resp requests.get(url, timeout5) return url, resp.status_code urls [ http://example.com/api/items/1, http://example.com/api/items/2, http://example.com/api/items/3, ] with ThreadPoolExecutor(max_workers8) as executor: future_map {executor.submit(fetch, url): url for url in urls} for future in as_completed(future_map): url future_map[future] try: _, status future.result() except Exception as exc: print(url, 失败, exc) else: print(url, 状态码, status)with块会在退出时等待所有线程任务结束相当于统一调用了 shutdown(waitTrue)。submit返回一个 Future 对象.result()会等待该任务完成如果任务抛出异常result 会再次抛出。用as_completed的好处是哪个任务先完成就先处理哪个不用等最慢的那个。线程池适合一批互相独立的任务比如批量检查 URL 状态、批量下载文件、批量查询数据库。你只需要关心并发数不需要关心线程创建、回收和异常处理代码维护成本直线下降。3. 多线程实战排错手册踩过的坑和解决思路3.1 打印乱序和共享变量“消失”的更新打印乱序是最容易发现、最容易误判的问题。多个线程同时在 print你会看到输出内容互相穿插比如一行英文中间插进另一行。这是因为 print 分成多次输出线程切换可能发生在中间。如果你只是想看日志用 logging 模块更合适如果想手动打印别分多次 print尽量拼成一个字符串一次性输出。共享变量“消失”的问题在 2.2 节已经演示过。这里再补充一个真实经验某数据采集器需要统计每个线程成功处理了多少条数据开发同学直接在函数外部放了几个全局变量用count 1更新。结果程序运行后统计数量永远对不上。定位时把线程数改成 1一切正常改成 8数据就丢。最后把计数逻辑改成每个线程维护自己的局部变量再汇总到队列问题才解决。用队列传递结果比共享全局变量安全得多。3.2 死锁锁顺序不一致带来的僵局死锁是最让人头疼的问题因为没有报错程序只是卡住不动。最常见的原因是两个线程各自持有一把锁又都想拿对方手里的锁。场景是这样的线程 A 先锁住锁 1然后想锁 2线程 B 先锁住锁 2然后想锁 1。A 不释放锁 1B 不释放锁 2两边都等对方释放于是永久僵住。避免死锁有几个土办法所有线程都按相同顺序获取锁比如一律先锁 1 再锁 2破坏循环等待。尽量不要嵌套锁。能用队列传递数据就别设计需要同时持有两把锁的临界区。在锁的acquire上设置超时拿不到就放弃。不过 Lock 的acquire(timeout...)返回布尔值需要自己处理失败分支。如果程序已经卡住可以按 CtrlC 让 Python 打印当前线程堆栈看看每个线程阻塞在哪个位置。这个操作很笨但有效能快速确认是否死锁。3.3 线程数量该开多少线程开少了跑得慢开多了也不一定快。每个线程都有自己的栈空间和内核资源线程数量过多时操作系统上下文切换开销会吞掉并发收益。我见过有人给线程池设 500 个 worker去请求一个只允许 20 个并发的外部接口结果大量连接超时整体比 20 个线程还慢。经验上IO 密集任务的合理并发量不能只看本机瓶颈还要看下游服务的承受能力。做法很简单从 1 个线程开始成倍往上加同时观察任务总耗时和错误率。如果加到 16 个线程后耗时不再下降错误率开始上升那这个值就是当前场景的最优并发量。不要迷信公式用数据说话。CPU 密集任务用多线程基本没救线程数设成核心数也是白搭直接考虑多进程。混合型任务则要拆解把计算部分交给进程池把 IO 等待部分交给线程池或协程。3.4 线程嵌套线程为什么我不建议热搜里有“python 线程嵌套线程”这个关键词让我想起一个合作项目里的事。当时某开发者为了让每个 worker 能同时处理多个子任务在 worker 内部又创建了新的线程结果任务一来线程数量按指数增长最后程序因为资源耗尽退出连日志都没留下几行。线程嵌套线程的难处在于生命周期控制。外层线程退出内层线程未必退出内层线程报错外层线程无法捕获你以为在池子里控制数量实际线程数完全失控。调试时看到几十个线程互相纠缠头都大。如果你确实遇到外层任务还要并发子任务的场景应当把子任务重新投递到一个独立的线程池而不是在外层线程里再开线程。这样数量可控退出逻辑也清晰。更彻底的做法是重新设计任务粒度让每个任务足够独立不要人为制造父子依赖。4. 场景实战多线程数据采集器从零搭起来4.1 需求与命令行入口为了别太抽象我模拟一个需求给定一个文本文件每行一个公开接口的 URL目标是并发请求这些接口把返回的 JSON 数据结构化后写入输出文件。这是一个很典型的 IO 密集型任务适合用多线程。入口我用 argparse 写这样命令行的--workers控制并发数--input指定输入文件--output指定输出路径方便在脚本里重复调整参数。这也是你在热搜里看到 argparse 的常见用途。import argparse import json import queue import threading import time import requests def parse_args(): p argparse.ArgumentParser(description多线程数据采集器) p.add_argument(--input, requiredTrue, help每行一个URL的输入文件) p.add_argument(--output, requiredTrue, help输出JSON文件路径) p.add_argument(--workers, typeint, default8, help并发线程数) return p.parse_args()输入文件可能是几千行单线程跑一遍可能要几分钟多线程可以把总耗时压到十几秒。这个提升在真实项目里很常见。4.2 生产者-消费者实现采集器的核心用队列做生产消费。主线程负责读文件把 URL 放入队列消费者线程从队列里取 URL请求、解析、保存结果最后主线程统一落盘。def worker(q, out_queue): while True: url q.get() if url is None: q.task_done() break try: resp requests.get(url, timeout5, headers{User-Agent: Mozilla/5.0}) resp.raise_for_status() data resp.json() out_queue.put({url: url, data: data}) except Exception as exc: out_queue.put({url: url, error: str(exc)}) finally: q.task_done() def main(): args parse_args() q queue.Queue(maxsizeargs.workers * 2) out_queue queue.Queue() with open(args.input, encodingutf-8) as f: urls [line.strip() for line in f if line.strip()] threads [] for i in range(args.workers): t threading.Thread(targetworker, args(q, out_queue), namefworker-{i}) t.start() threads.append(t) for url in urls: q.put(url) for _ in range(args.workers): q.put(None) q.join() results [] while True: try: results.append(out_queue.get_nowait()) except queue.Empty: break with open(args.output, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) if __name__ __main__: main()这里有几个细节。队列容量设置成args.workers * 2是为了防止一次性把所有 URL 都加载进队列内存不至于被几千个字符串挤爆。结束信号放了args.workers个 None每个 worker 消费一个保证所有线程都能收到退出信号。out_queue.get_nowait()在最后统一导出结果避免多个线程同时写文件的竞争。4.3 并发控制、超时和重试真实网络环境远没有代码看起来这么顺。一个接口偶尔超时另一个接口返回 500都需要处理。这里我把请求部分封装成独立函数加上指数退避重试HEADERS {User-Agent: Mozilla/5.0} def fetch_json(url, max_retries3): for attempt in range(max_retries): try: resp requests.get(url, timeout5, headersHEADERS) resp.raise_for_status() return resp.json() except Exception: if attempt max_retries - 1: raise time.sleep(0.5 * (2 ** attempt))timeout5是必须的。很多线程卡死的根源不是并发数量而是某个请求一直不返回线程就永远等在那里。加了超时最多 5 秒就抛异常重试机制才有机会介入。指数退避的核心理念是出错后先等短时间再等更长时间避免在服务端还没恢复时疯狂重试把对方打垮。0.5、1、2 秒这样的阶梯足够温和。4.4 把结果安全写盘多线程写同一个文件是最容易出问题的操作之一。如果每个线程拿到结果后直接open(...).write(...)你很可能会看到文件内容互相穿插、JSON 被截断、或者写入顺序完全随机。更稳妥的做法是像 4.2 节那样线程只负责把结果放进输出队列主线程等所有任务结束后统一写盘。单一写入者可以避免锁竞争和文件句柄混乱而且数据落盘顺序可控。如果需要边采边落盘也可以专门开一个写盘线程从输出队列取数据一条条追加到文件。但要注意追加模式下的缓冲问题避免程序崩溃丢数据。总之多线程共享状态越少事故越少。5. 多线程之外协程和多进程到底该怎么选5.1 一张表理清选择思路聊到这里你已经知道多线程不是唯一答案了。做技术选型时可以从任务类型倒推方案。任务类型推荐方案核心原因CPU 密集大量计算、循环、加密多进程避开 GIL利用多核IO 密集请求、文件、数据库多线程 / 协程等待外部资源时让出 CPU高并发网络 IO成千上万个连接协程线程切换开销大协程更轻量混合型先计算后请求再计算多进程 多线程/协程拆解任务各取所长这张表不是绝对规则而是帮助你在拿到一个模糊需求时不至于上来就写 ThreadPoolExecutor。先把任务拆开看瓶颈是计算还是 IO再决定方案。5.2 线程到协程的思维转变协程是比线程更轻量的“用户态调度”。一个线程里可以跑成千上万个协程协程之间通过 await 让出控制权。对于大量网络 IO 的场景协程比多线程更节省资源。简单感受一下 asyncio 的写法import asyncio async def fetch(session, url): print(开始请求, url) await asyncio.sleep(1) # 模拟IO等待 print(完成请求, url) return url async def main(): tasks [fetch(None, fhttp://example.com/{i}) for i in range(5)] results await asyncio.gather(*tasks) print(results) asyncio.run(main())async def定义协程函数await表示让出 CPU。当程序执行到await asyncio.sleep(1)时事件循环会去调度其他协程所以 5 个任务几乎同时进入等待同时完成。这段代码在执行效率上能达到多线程的效果而线程数只有 1。但协程有一个学习门槛你不能在异步函数里随意调用阻塞库。比如requests.get是同步阻塞的直接放进 async 函数里会卡住整个事件循环。如果项目里已经用了大量同步库多线程反而是最小改动方案。5.3 多线程和协程结合的小技巧实际工程里不会那么纯粹经常碰到“异步框架里调一个同步 SDK”的情况。比如你写了一个 asyncio 服务但某个功能必须调用一个老旧的同步客户端库直接调用会卡住事件循环。这时可以用线程池把它“外包”出去import asyncio import requests async def main(): # Python 3.9 resp await asyncio.to_thread(requests.get, http://example.com, timeout5) print(resp.status_code)或者用更低层的方式import asyncio from concurrent.futures import ThreadPoolExecutor async def main(): loop asyncio.get_running_loop() with ThreadPoolExecutor(max_workers4) as pool: result await loop.run_in_executor(pool, requests.get, http://example.com, timeout5)这种做法让你既能享受异步框架的高并发调度又能兼容同步阻塞库。线程池的默认大小可能不够最好根据自己的 IO 并发量显式指定。它是连接“异步世界”和“同步世界”的一座桥。5.4 量化回测、批量任务中的真实取舍前面说了很多技术细节最后落到热搜里的两个关键词“量化交易策略”和“结构化数据”。在量化回测中为了测试不同参数组合常常要跑几十组策略。每组策略内部有大量因子计算属于 CPU 密集用多线程会被 GIL 按住所以常见做法是multiprocessing.Pool并行跑不同参数的回测每个子进程独立计算再把结果汇集到一起。实时行情订阅则是另一个方向。行情推送是持续不断的网络 IO适合用异步或多线程接收再把收到的数据写入数据库或消息队列。数据清洗和结构化转换pandas 底层很多操作是 C 库实现并不完全受 GIL 控制所以某些情况下多线程也有价值但最稳妥的方案还是把数据按时间段分块交给多进程并行处理最后合并结果。关键不是背结论而是先压测再选择。一个任务拿到手先想想它是“等得久”还是“算得久”然后在小规模数据上分别用线程、进程、协程跑一遍谁快用谁。多线程是工具箱里的一把好锤子但别把所有问题都看成钉子。6. 最后分享一点我的体会写并发代码这几年我最大的体会是先把单线程版本跑通再考虑加速。很多人一上来就堆线程结果逻辑错误和并发错误混在一起调试难度翻倍。正确的步骤是先写一个简单、正确、能出结果的版本然后找到真正的性能瓶颈再决定要不要上多线程、多进程还是协程。另一个小技巧是多线程排错先从日志入手。给每个线程取一个可读的名字所有日志统一走 logging带上线程名打印出来。一旦程序卡住或者结果不对先看日志别瞎猜。用队列传递数据比共享变量加锁更容易写对线程池管理生命周期比手动创建线程更省心。避开嵌套线程并发代码会好写很多。最后再补一句实在话多线程提升的是系统吞吐不是单个任务的延迟。你的场景如果是“数量多、等待多”多线程是真香如果是“单个任务计算量大”老老实实去研究多进程和协程。选对工具比努力调参数重要得多。