1. 为什么Python需要多进程与多线程当你用Python处理一个需要重复执行的任务时比如批量处理图片或者爬取网页数据单线程程序就像一个人在厨房里既切菜又煮汤——效率低下且浪费时间。这时候多线程和多进程就像请来了帮手可以同时处理多个任务。但Python有个特殊限制全局解释器锁GIL。这就像厨房里只有一个灶台即使有多个厨师线程同一时间也只能有一个人在炒菜。不过别担心多进程可以绕过这个限制因为每个进程都有自己的灶台CPU核心。我做过一个实验用单线程处理100张图片需要120秒而用4个进程只需要35秒。这就是为什么理解多进程和多线程如此重要。2. 如何判断任务类型2.1 CPU密集型任务这类任务主要消耗CPU资源比如数学计算如机器学习模型训练图像处理数据加密/解密判断技巧如果你的代码大部分时间都在做数学运算几乎没有文件读写或网络请求那就是CPU密集型。我曾经优化过一个图像处理脚本发现90%的时间都花在了像素计算上。这时候增加线程数反而让程序更慢因为线程切换带来了额外开销。2.2 IO密集型任务这类任务大部分时间在等待外部响应比如网络请求爬虫、API调用文件读写数据库查询有个简单的方法可以判断在Linux下运行top命令如果CPU使用率很低但程序运行很慢很可能就是IO密集型。实际案例我开发过一个爬虫单线程每小时只能爬取200个页面。改成多线程后50个线程性能提升了15倍而CPU使用率只增加了20%。3. 多进程实战充分利用CPU核心3.1 获取CPU核心数import multiprocessing cpu_count multiprocessing.cpu_count() print(f你的CPU有{cpu_count}个逻辑核心)注意超线程技术会让CPU核心数显示为物理核心的2倍。对于计算密集型任务建议使用物理核心数。3.2 进程池的最佳实践from multiprocessing import Pool import time def cpu_bound_task(n): return sum(i*i for i in range(n)) if __name__ __main__: start time.time() with Pool(processescpu_count) as pool: results pool.map(cpu_bound_task, [1000000]*8) print(f耗时{time.time()-start:.2f}秒)在我的6核CPU上测试进程数6耗时4.2秒进程数12耗时4.5秒反而更慢进程数3耗时6.8秒结论对于纯计算任务进程数等于物理核心数时效率最高。4. 多线程调优突破IO瓶颈4.1 线程数计算公式对于IO密集型任务一个经验公式最佳线程数 (IO等待时间 CPU处理时间) / CPU处理时间 * CPU核心数举例一个网络请求任务每次请求CPU处理时间5ms网络等待时间95msCPU核心数4计算(955)/5 * 4 80个线程4.2 线程池示例from concurrent.futures import ThreadPoolExecutor import requests def fetch_url(url): response requests.get(url) return len(response.text) urls [https://example.com]*100 with ThreadPoolExecutor(max_workers80) as executor: results list(executor.map(fetch_url, urls))在我的测试中线程数从10增加到80时完成时间从18秒降到2.3秒。超过100后改善不明显反而可能因内存不足导致性能下降。5. 混合使用多进程和多线程对于既有CPU计算又有IO等待的任务可以这样组合from multiprocessing import Pool from concurrent.futures import ThreadPoolExecutor import numpy as np def process_data(data): # CPU密集型计算 result np.mean(data)**2 return result def io_task(url): # IO密集型任务 data download_data(url) with ThreadPoolExecutor(4) as executor: processed list(executor.map(process_data, data)) return processed if __name__ __main__: urls [...] # 100个数据源URL with Pool(processes6) as pool: all_results pool.map(io_task, urls)这种架构下外层用多进程绕过GIL限制内层用多线程处理IO等待每个进程内的线程数建议为CPU核心数的1-2倍6. 常见误区与性能陷阱盲目增加线程数我见过有人设置1000个线程结果导致系统崩溃。线程切换开销会抵消并发优势。忽略资源共享问题多进程间共享数据需要使用Queue或共享内存直接共享变量会导致数据不一致。错误使用全局变量多线程修改全局变量需要使用锁否则会出现竞态条件。忘记关闭资源线程池/进程池如果不正确关闭会导致资源泄漏。一定要使用with语句或手动调用shutdown()。一个真实案例某次我用多进程处理数据时忘记关闭数据库连接导致达到最大连接数限制整个服务不可用。后来改用连接池才解决问题。7. 高级技巧动态调整并发数对于不确定任务类型的情况可以实现动态调节import time from threading import active_count from concurrent.futures import ThreadPoolExecutor class DynamicThreadPool: def __init__(self, max_workers): self.executor ThreadPoolExecutor(max_workers) self.max_workers max_workers self.adjust_interval 5 def monitor(self): while True: active active_count() if active self.max_workers * 0.8: # 考虑增加线程数 pass time.sleep(self.adjust_interval)这种模式适合任务类型变化大的场景比如白天处理用户请求IO密集夜间跑批处理任务CPU密集。8. 实际项目经验分享去年优化一个电商价格计算服务时我经历了这样的调优过程最初版本单进程单线程处理10万商品需要45分钟第一版优化改用多线程20线程时间降到15分钟发现问题CPU使用率只有30%因为涉及大量数据库查询最终方案4个进程每个进程内10个线程最终耗时4分钟关键发现价格计算本身是CPU密集型但需要先查询数据库IO密集。混合方案取得了最佳效果。监控工具推荐psutil监控CPU/内存使用率time.perf_counter()测量代码段执行时间threading.enumerate()查看线程状态