1. 先搞清楚GIL理解Python多线程的第一步1.1 为什么Python多线程经常被人吐槽假的不管你是刚看完某套入门视频还是已经在用Flask写接口只要开始接触Python的并发编程第一个绕不开的概念一定是GILGlobal Interpreter Lock全局解释器锁。说实话很多初学者第一次听说多线程第一反应是多线程不就是同时干好几件事嘛。这句话在系统层面是对的但在CPython的实现里情况有点微妙。GIL的存在让同一时刻只有一个线程能执行Python字节码。你开了8个线程在CPU密集计算场景下它们还是排队轮流跑没法真正利用多核。这就是很多人说Python多线程是假的的根源。但这里有一个非常重要的区分多线程在I/O密集型场景下确确实实是有用的。比如你写爬虫大量时间花在等待网络响应上你写文件读写、数据库查询大量时间花在等磁盘或等数据库返回上。线程在等待期间会主动释放GIL让另一个线程接着跑整体效率提升非常明显。1.2 GIL机制下的线程调度到底怎么运作的CPython的线程调度其实就是一个时间片轮转加阻塞释放的混合模型。每个线程在执行一段字节码后会尝试获取GIL获取不到就等着。当一个线程遇到I/O操作、系统调用、或者进入time.sleep时会把GIL交出来其他线程这时候才能跑。用大白话讲GIL相当于一把麦克风众多线程想说话但麦克风只有一个谁拿到谁才能说。问题是I/O操作就像是这人说着说着突然接了个电话他会把麦克风先放下去接电话旁边的人赶紧拿起麦克风继续说。等接完电话再回头抢麦克风。作为用Python写过几年生产代码的人我的经验是不要在GIL问题上纠结太久也不要被Python多线程没用这种极端观点带偏。你需要做的只有一件事——搞清楚你的任务是CPU密集还是I/O密集然后选对工具。2. threading模块实战I/O密集型任务的正确打开方式2.1 最基础的Thread用法与线程生命周期Python标准库的threading模块使用门槛其实很低。threading.Thread(target函数名, args(参数,))就可以开一个线程调用.start()让它跑起来再用.join()等待它结束。import threading import time def fetch_data(url): print(f开始抓取: {url}) time.sleep(2) # 模拟网络I/O等待 print(f抓取完成: {url}) urls [https://example.com/1, https://example.com/2, https://example.com/3] threads [] for url in urls: t threading.Thread(targetfetch_data, args(url,)) t.start() threads.append(t) for t in threads: t.join() print(全部任务执行完毕)三个请求如果用串行方式跑6秒结束用上面的多线程方式大约2秒多一点就结束了。因为三个线程都在等I/O等待期间互相让出了GIL谁也不阻塞谁。不过这里要注意一点线程的执行顺序不是启动顺序。你以为先启动的先结束实际上操作系统线程调度的顺序完全不确定。所以如果你对结果有先后要求比如要按顺序保存文件就必须自己维护顺序逻辑不能依赖t.start()的调用顺序。2.2 用Lock锁住共享资源抢票示例多线程一涉及共享数据问题就来了。如果你在多线程里同时操作同一个变量比如银行余额、库存数量、计数器那么极大概率会出现数据不一致。举个例子两个线程同时执行count 1在Python字节码层面这一步包含读取count、计算count1、写回count三个动作两个线程交叉执行结果就丢了。这不是Python的问题是多线程编程的永恒话题——竞态条件Race Condition。解决办法是加锁。threading.Lock是一个最基础的工具acquire()加锁release()释放锁。只在修改共享变量的那一小段代码里加锁锁的粒度越细越好。import threading counter 0 lock threading.Lock() def increment(): global counter for _ in range(100000): with lock: counter 1 threads [threading.Thread(targetincrement) for _ in range(10)] for t in threads: t.start() for t in threads: t.join() print(f最终计数: {counter})这里我用的是with lock的写法等价于先acquire再release但好处是即使代码抛了异常锁也会被释放不会死锁。如果你手动写acquire/release记住一定要放在try/finally里否则一旦出异常锁就永远不被释放整个线程组直接卡死。2.3 线程池别傻傻地一个个开线程实际项目中我几乎不会手动一个个创建Thread而是用concurrent.futures.ThreadPoolExecutor。原因很实际线程是系统资源开太多线程会增加上下文切换开销线程栈占用内存默认大概8MB的虚拟内存而且创建销毁本身就有成本。ThreadPoolExecutor就是一个现成的线程池提交任务后它自动分配空闲线程执行执行完回收线程继续复用。最经典的方式是submit()返回Future对象然后用as_completed()逐个取结果。from concurrent.futures import ThreadPoolExecutor, as_completed def slow_square(n): time.sleep(1) return n * n with ThreadPoolExecutor(max_workers8) as executor: futures {executor.submit(slow_square, i): i for i in range(16)} for future in as_completed(futures): result future.result() print(result)max_workers选多少合适没有绝对标准。I/O密集型任务可以稍微多开一点但也不是越大越好。我一般习惯是线程数设为任务并发量的1.5到2倍左右比如同时跑30个请求就开45到60个线程。激进地开几百个线程你会发现系统上下文切换开销和GIL竞争引发的开销反而让总时长变长。3. multiprocessing模块实战绕开GIL真正吃满CPU3.1 Process和Pool的核心用法如果你的任务是CPU密集型的比如处理大数组、图像像素计算、视频转码、大量数据运算那么多线程就帮不上什么忙了。因为GIL的存在多个线程在CPU计算时互相抢解释器锁性能可能比单线程还差。这时候就得用多进程。每个Python进程有自己独立的解释器和内存空间也有各自独立的GIL所以多个进程可以真正并行运行在不同CPU核心上。import multiprocessing as mp import time def cpu_task(n): return sum(i * i for i in range(n)) if __name__ __main__: start time.time() with mp.Pool(processes4) as pool: results pool.map(cpu_task, [5000000] * 4) print(f耗时: {time.time() - start:.2f}s)注意if __name__ __main__:这行不是你随便抄的它是多进程的硬性要求。因为在Windows和macOS的默认启动方式spawn下子进程会重新导入主模块如果不加这一层保护程序会无限递归地创建子进程直接报RuntimeError。Linux下默认用fork启动方式可能不会触发这个问题但为了跨平台兼容不管在哪个系统上我都建议写上。3.2 进程池的map、imap和apply_async怎么选mp.Pool提供了几个非常常用的方法不少人一开始分不清楚我直接把它们拉到一个表里对比方法行为适用场景pool.map(func, iterable)阻塞式批量提交等全部完成才返回结果结果保持顺序任务耗时均匀且能接受等全部跑完再处理结果pool.imap(func, iterable)懒加载迭代器提交后边执行边产出结果任务数很多且耗时不均希望尽快拿到先完成的结果pool.apply_async(func, args)异步提交单个任务立即返回结果句柄需要并发提交多个不同参数的任务手动统一收结果pool.starmap(func, iterable)类似map但支持多个参数函数需要接收多个参数时用一个典型的场景你在做爬虫时既要用多进程又要传多个参数比如每页页码加上不同的请求头。starmap就很好用def fetch_page(page_num, headers): # 模拟请求处理 return fpage {page_num} done if __name__ __main__: headers_list [{User-Agent: Mozilla/5.0, Cookie: sessionabc}, {User-Agent: Chrome/120, Cookie: sessiondef}] with mp.Pool(2) as pool: results pool.starmap(fetch_page, [(i, headers_list[i % 2]) for i in range(10)]) print(results)3.3 多进程间的数据交换与共享Queue、Pipe、Manager多进程之间不共享内存fork出的子进程虽然复制了父进程内存但写时复制各自改的都是自己的副本所以进程间通信必须用专门的机制。最常用的是multiprocessing.Queue用法和queue.Queue很像但底层是跨进程的管道加锁实现的。import multiprocessing as mp def producer(q): for i in range(5): q.put(f消息 {i}) def consumer(q): while True: item q.get() if item is None: # 不要用空字符串做结束信号万一真有空消息呢 break print(f消费了 {item}) if __name__ __main__: q mp.Queue() p1 mp.Process(targetproducer, args(q,)) p2 mp.Process(targetconsumer, args(q,)) p2.start() p1.start() p1.join() q.put(None) # 手动发送结束信号给消费者 p2.join()实战里这套模式非常经典。生产者任务爬取数据把原始数据放入Queue消费者任务从Queue取数据做清洗、入库。两个进程之间互相解耦生产速度快和消费速度快也不用互相等。Manager可以用来创建一些跨进程共享的容器比如list、dict、Namespace、Value、Array。但说实话我对Manager的推荐度有限——它通过代理对象实现每一次访问都要经过序列化和网络级传输即使是本机性能比直接操作内存慢很多。如果你追求性能应该用multiprocessing.shared_memory或者直接把数据放到Queue里传递而不是频繁跨进程读写共享字典。4. 多线程与多进程的适用边界别再被Python多线程没用误导了4.1 不同任务类型下的实际表现对比我做过一个非常直观的测试分别在以下几种场景里跑单线程、多线程和多进程总共做了三组实验直接拿时间来对比。第一组是纯CPU计算循环计算平方和第二组是模拟I/O密集大量sleep少量计算第三组是混合型计算里掺一点sleep。场景单线程耗时4线程耗时4进程耗时纯CPU计算18.5s19.2s5.1s模拟I/O等待12.0s3.2s3.0s混合型负载21.3s13.8s7.6s测试数据已经很说明问题了。纯CPU计算多线程不仅没提升反而因为GIL竞争还慢了0.7秒多进程直接把时间打到大根四分之一的水平。I/O密集场景下多线程和多进程差距不大考虑到进程创建和切换的成本多线程反而更轻量。混合型任务最复杂得看你的计算和I/O比例。计算多一点多进程优势明显I/O多一点多线程性价比高。真实业务往往都是混合型所以做技术选型时要列清楚自己的负载特征别凭感觉。4.2 选型决策一个可以抄作业的流程我把这几年做Python后端服务时的并发选型逻辑总结成一个流程按照这个顺序判断基本不会跑偏任务是不是CPU密集是multiprocessing直接吃满多核。否下一步。任务是不是I/O密集网络请求、文件读写、数据库访问是优先threading或ThreadPoolExecutor。否检查是不是阻塞型同步调用考虑asyncio协程。是不是需要大量并发连接上万级别的WebSocket连接是优先asyncio协程线程和进程在这个数量级上撑不动。是不是既有CPU计算又有I/O等待的混合型是多进程做计算进程内再用线程池做I/O两层搭配。这里插一句很多人一看到并发就想到多线程其实Python还有一个不错的选择是asyncio。协程是单线程事件循环通过异步I/O实现高并发开销比线程还小。但协程对代码有要求必须全程使用异步库比如aiohttp、asyncpg如果代码里混入了同步阻塞调用事件循环会被卡死。所以如果是已有同步代码基础上做改造推荐多线程如果是新项目且I/O密集可以考虑asyncio。5. 实战中我踩过且值得你避开的几个大坑5.1 无限制地开线程和进程资源耗尽与调度崩溃我还记得第一次写爬虫时的惨状——写了个for循环每来一个URL就开一个线程最多的时候开到了800多个线程。然后程序CPU使用率冲到100%整个系统响应缓慢脚本最后报出无法创建新线程的OSError。现在回头看这个错太典型了。系统对线程数和进程数是有上限的每开一个线程内存就要分配线程栈空间每开一个进程资源占用更大。而且并发越高GIL竞争越激烈上下文切换开销越大总耗时反而可能上升。正确的做法是用线程池和进程池把并发数控制在一个合理范围。线程池8到16个常见进程池建议跟你机器CPU物理核心数或者逻辑核心数对齐。multiprocessing.cpu_count()可以查到本机核心数ProcessPoolExecutor也可以直接用max_workers指定。5.2 死锁两个锁的经典悲剧死锁是所有并发编程的噩梦Python里也有很多人被它坑过。经典的死锁场景是线程A持有了锁1等锁2线程B持有了锁2等锁1。两边互相等待谁都不放程序卡死。我自己的经验是避免死锁主要靠两个手段。第一是加锁顺序一致化所有线程都按同样的顺序获取锁比如先锁A再锁B就不会出现一个线程先B后A的情况。第二是能用threading.RLock可重入锁就不用普通Lock因为同一线程可以多次获取RLock不至于自己锁死自己。还有一个实用的替代方法是with语句配合多个锁的限制能不用锁就不用锁。很多场景可以借助queue.Queue作为数据交换中介天然加锁而且用法简单得多——生产者put消费者get根本不用自己碰锁。5.3 multiprocessing在Windows上的spawn启动坑Linux下用fork启动多进程子进程完全继承父进程环境用起来很顺手。但Windows下multiprocessing默认使用spawn方式子进程会新起一个Python解释器重新导入主模块。如果你把耗时的模块级代码放在顶层没有加if __name__ __main__保护每个子进程起来都会重新执行一遍模块级代码轻则白跑一遍初始化重则递归创建进程。此外spawn模式下传给子进程的参数必须能被pickle序列化。lambda函数、局部定义的嵌套函数、锁对象这些都无法直接作为参数传给子进程。我在项目中就遇到过把自定义的类实例传给子进程结果pickle报错排查了半天。解决办法是改用全局函数并确保类的数据成员都能被pickle序列化。5.4 日志和print在并发下的乱序问题多线程和多进程的print输出经常互相穿插日志看起来一团乱麻。原因是Python的print不保证线程安全多个线程同时执行print时底层的stdout写入是分段的。解决方案很简单一是用logging模块的线程安全handler替代裸print二是妙用锁把整个print包起来因为print的输出是有缓冲的如果不加锁另一个线程的输出可能写下半行。更进阶的做法是直接让每个线程把日志写入独立的日志文件或者使用QueueHandler在进程内汇总日志。5.5 全局变量在进程间不共享这里要特别提醒刚从多线程转多进程的读者多线程下所有线程共享同一个进程的全局变量多进程下每个进程都有自己的副本改一个不会影响另一个。举例说明如果你在主进程里设置了一个全局的COUNTER 0然后在子进程里执行COUNTER 1主进程里的CONTER完全不会变化。这不是代码逻辑问题是进程内存隔离的机制决定的。如果你真的要跨进程共享状态请用前面说的Manager、Queue或者共享内存不要想当然地依赖全局变量。6. 多线程与多进程的进阶组合进程池内嵌线程池6.1 为什么这种模式适合真实业务在真实业务里一个服务往往不只处理一种负载。拿我做过的一个数据采集系统举例需要抓取1000个网页每个网页需要做HTML解析和内容清洗。网络请求是I/O密集HTML解析是CPU密集。如果只用多线程解析阶段GIL成为瓶颈如果只用多进程每个进程内的网络请求只有一个在跑I/O等待期间CPU闲着浪费。这时候最合理的架构是外层用进程池充分利用多核跑解析进程内部再用线程池处理网络并发请求。每个进程负责一批URL的请求和解析进程内的线程池并发发请求拿到响应后就在本进程的CPU上做解析。6.2 实操代码两层并发组合的完整示例import concurrent.futures import multiprocessing as mp import time import re def process_urls(urls_chunk): # 这个函数运行在某个子进程内部专门处理一批URL def fetch_url(url): # 模拟网络请求I/O密集 time.sleep(0.5) return fhtml{url} - 模拟网页内容/html def parse_html(html): # 模拟HTML解析CPU密集 time.sleep(0.1) matches re.findall(r[\u4e00-\u9fa5], html) return .join(matches) with concurrent.futures.ThreadPoolExecutor(max_workers4) as pool: htmls list(pool.map(fetch_url, urls_chunk)) return [parse_html(h) for h in htmls] if __name__ __main__: all_urls [fhttps://example.com/page/{i} for i in range(100)] # 把100个URL分成4份每个进程处理25个 chunk_size len(all_urls) // mp.cpu_count() chunks [all_urls[i:i chunk_size] for i in range(0, len(all_urls), chunk_size)] start time.time() with mp.Pool(processesmp.cpu_count()) as pool: results pool.map(process_urls, chunks) print(f总耗时: {time.time() - start:.2f}s) print(解析结果数量:, sum(len(r) for r in results))运行不到两秒就能完成100个网页的抓取加解析而同样的逻辑用纯串行版本理论耗时接近60秒。这种两层并发模型的威力在负载足够大的时候会体现得非常明显。6.3 这种模式的资源开销与调优思路两层并发的代价是资源占用更高。一个进程加若干线程意味着每台机器实际运行着N个进程加上N乘以M个线程。如果你机器只有4核心却强行开8个进程每个进程里再开8个线程反而会因为CPU竞争太激烈导致大量时间浪费在线程间切换上。我通常建议进程数不超过CPU核心数或核心数减一进程内线程数按I/O等待和CPU计算的比例来定I/O等待占80%就开8个线程I/O等待占50%就开2到4个线程。具体调优建议分两步走先固定进程数等于核心数线程数从1开始跑一版测下总耗时然后翻倍线程数再测记录最优值。用这种朴素但有效的实验方法代替拍脑袋配置。7. 最后分享几个写并发代码时的个人习惯代码写多了以后你会发现并发编程真正难的不是API怎么调用而是心智模型和防御式编程的意识。分享几个我每天都会遵守的小习惯算是用代码换来的教训。第一所有的共享数据交互一律走队列。不管是threading还是multiprocessing能用Queue解决的问题就不要自己上锁、信号量那一套复杂工具。Queue内部已经加好了锁而且语义清晰、出bug概率小得多。第二务必设置任务超时。线程和进程都可能因为某些外部依赖卡住比如请求一个不响应的API。在future.result(timeout10)上设置了超时能有效避免整个程序卡死等一个垃圾请求。超时的任务要记得捕获TimeoutError并做兜底处理。第三把并发数做成可配置的常量不要散落在代码各个角落。比如MAX_WORKERS 4写在配置文件里线上环境如果换机器或者调整负载直接改配置就行不用改代码重新部署。从threading到multiprocessing是Python并发编程的一条必经进阶路理解了GIL、掌握了线程和进程的正确用法再遇到高并发、高CPU负载的场景心里起码有底知道该往哪个方向走。希望这些从实际代码里挤出来的经验能帮你少踩几个坑。