Python并发编程实战多线程池、多进程池与线程同步的那些坑和技巧这几年写Python几乎每个项目都会碰到同一个问题并发量上来了程序变慢了第一反应就是开线程。但真正把多线程、多进程、线程池进程池、锁和队列这些东西全部摸清楚我觉得至少得踩三四个实际的坑才算入门。这篇文章就是把我这些年在项目里折腾线程同步、线程池和进程池的经验整理出来包含可以直接抄的代码、参数选择思路以及常规文档里不会告诉你的排查方法。适合刚接触并发编程的初学者也适合写了一段时间多线程但总觉得哪里不对劲的开发者。说到底Python并发要解决的就两个问题让IO等待时间不再白费让多核CPU真正跑起来。前者用多线程后者用多进程。但这中间涉及GIL、共享状态、死锁、序列化开销每一个都能让程序在线上突然卡死。所以我从最基础的概念讲起一步步带你走完线程、同步、池化、进程通信整个链路。1. 先搞清楚核心问题Python并发到底在解决什么1.1 GIL是绕不开的第一道坎很多新手看教程上来就学Thread然后用线程跑一个纯计算的函数发现两个线程加起来比单线程还慢就一脸懵。这个现象背后就是GIL全局解释器锁。Python解释器同一时刻只允许一个线程执行Python字节码也就是说如果是纯CPU计算任务多线程等于把并发执行变成串行还额外付出线程切换的开销。真正让多线程有用的场景是IO密集型任务——比如网络请求、文件读写、数据库查询。这些操作在等待外部响应的时候会主动释放GIL让其他线程趁这个机会跑起来这时候多线程就能实打实地提升吞吐量。我自己的判断习惯是任务里有明显的阻塞等待网络、磁盘、用户输入用多线程任务全是CPU运算图像处理、加解密、大规模数学计算直接上多进程。别指望一条路走到黑。1.2 线程和进程各管哪一片线程共享进程的内存空间创建成本低切换快但是共享变量的时候容易出现数据竞争得加锁保护。进程拥有独立的内存空间天然隔离不需要担心变量被别的进程乱改但进程创建开销大进程间通信还要靠队列、管道这些专门的机制。用一个生活化的类比线程像一家人共用一个厨房大家共用锅碗瓢盆做饭快但容易抢东西打架进程像每家每户有独立厨房互不干扰但你要借个酱油得走门串户。所以在设计并发方案的时候我通常先问三个问题任务是IO密集还是CPU密集任务之间需不需要共享数据共享的数据量有多大单个任务的耗时是秒级还是毫秒级这三个问题回答完用线程池还是进程池基本就定了。2. threading基础线程的创建、生命周期与正确关闭2.1 最基础的线程写法和为什么必须join先看一个最简单的线程模型import threading import time def worker(name, seconds): print(f线程 {name} 开始工作预计耗时 {seconds}s) time.sleep(seconds) print(f线程 {name} 工作完成) threads [] for i in range(3): t threading.Thread(targetworker, args(fworker-{i}, 2)) threads.append(t) t.start() # 注意是start不是run for t in threads: t.join() # 等待所有线程结束主线程才能继续往下走 print(所有线程结束主程序退出)这里有两个新手必踩的坑。第一Thread(target...)只是创建了线程对象一定要调用start()才会真正启动如果直接调用run()那就变成了普通函数调用完全串行执行。第二主线程默认不会等待子线程结束如果你不调join()可能主线程已经跑完了子线程还在后台干活程序就直接退出了。join()还支持超时参数比如t.join(timeout5)表示最多等5秒5秒后还没结束就继续往下走。这个在关闭资源、优雅退出的时候非常有用——你不能无限期等一个可能卡死的线程。2.2 守护线程daemon退出程序时要不要强制回收守护线程的概念很简单daemonTrue的线程在主线程退出时会跟着一起被强制终止不管它活没干完。非守护线程则会在主线程退出后仍然继续运行直到它自己结束。t threading.Thread(targetworker, args(daemon-task, 30), daemonTrue) t.start() # 主线程做完事情直接退出daemon线程被强制中断实际项目中日志监听、心跳上报、看门狗这类后台任务设成daemon比较合理——你不想因为一个后台线程卡住导致整个程序退不了。但如果是正在写文件、提交数据库事务这种任务千万别设成daemon否则程序退出的时候数据可能丢一半。这是我在一个数据采集项目里踩过的坑当时为了图省事把采集线程设成了daemon结果是进程退得倒是干净数据库里留下了一堆不完整记录。2.3 共享变量要用锁一个简单的计数器引发的血案先看一个经典问题。两个线程同时对同一个全局变量做count 1每个线程循环100万次你觉得结果是多少import threading count 0 def increase(): global count for _ in range(1_000_000): count 1 threads [threading.Thread(targetincrease) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(count) # 结果往往不是2000000我第一次跑这个例子的时候输出的是124万左右少了几十万。原因很简单count 1在底层是三条指令——读取count、加1、写回count。两个线程可能同时读到同一个值各自加1再写回结果有一方的修改被覆盖了。这个就是数据竞争。解决办法就是加锁把读取-修改-写回做成不可分割的原子操作import threading count 0 lock threading.Lock() def increase(): global count for _ in range(1_000_000): with lock: # 等价于lock.acquire() finally lock.release() count 1用with lock而不是手动acquire()/release()好处是就算代码中途抛异常锁也会被自动释放不会出现拿到锁之后忘了解锁导致死锁的情况。这个习惯我从踩过一次死锁之后就一直保持了。3. 线程同步机制Lock、RLock、Semaphore、Event、Condition怎么选3.1 Lock与RLock可重入锁解决递归加锁问题Lock也叫互斥锁同一时刻只能被一个线程持有。但如果同一个线程在持锁期间又去acquire()同一把锁就会自己把自己卡死。看这个场景import threading lock threading.Lock() def outer(): lock.acquire() inner() # 内部又去acquire同一把锁 lock.release() def inner(): lock.acquire() # 死锁 lock.release()这种情况用RLock可重入锁就解决了。RLock允许同一个线程多次acquire()每次acquire()对应一次release()只有计数归零才算真正释放。在实际项目里一个函数内部嵌套调用另一个也加了锁的函数是非常常见的所以我的项目里凡是涉及同一线程可能递归进入的锁一律用RLock省得排查半天是这个原因导致的卡死。3.2 Semaphore限制并发数的信号量Semaphore用来控制同时访问某个资源的线程数量。比如你的程序最多允许5个线程同时连数据库不然数据库会被打爆import threading import time semaphore threading.Semaphore(5) def db_query(query_id): with semaphore: print(f查询 {query_id} 开始占用一个连接) time.sleep(1) print(f查询 {query_id} 结束释放连接) threads [threading.Thread(targetdb_query, args(i,)) for i in range(20)] for t in threads: t.start() for t in threads: t.join()Semaphore(5)表示初始允许5个线程通行第6个线程调用acquire()时会阻塞直到前面某个线程release()。注意一点如果初始值设成1Semaphore就是一把Lock。还有BoundedSemaphore它在release()次数超过acquire()次数时抛异常能帮你发现代码逻辑错误普通Semaphore则不会管这个。3.3 Event一个线程等另一个线程的信号Event适合一个线程等待另一个线程完成某个动作的场景比如主线程等待所有工作线程初始化完成再下发任务import threading import time ready_event threading.Event() def worker(worker_id): print(fworker-{worker_id} 正在初始化) time.sleep(1) print(fworker-{worker_id} 初始化完成) ready_event.set() # 发信号 def main(): t threading.Thread(targetworker, args(1,)) t.start() ready_event.wait() # 阻塞等待信号 print(所有初始化完成主线程开始正式干活) main()Event的核心就两个方法wait()阻塞等待set()发出信号让所有等待的线程通过clear()把事件重置。有一个细节是clear()很容易被忽略——如果事件是一次性的不用管如果是周期性用的用完之后必须调clear()不然下一次wait()直接通过了逻辑就全乱了。3.4 Condition生产者消费者模型的标准解法Condition是更高级的同步原语本质上是锁加等待通知机制。最经典的应用就是生产者消费者模型生产者往缓冲区放数据消费者从缓冲区取数据缓冲区满的时候生产者要等缓冲区空的时候消费者要等。import threading import time from collections import deque condition threading.Condition() queue deque() MAX_SIZE 5 def producer(): for i in range(20): with condition: while len(queue) MAX_SIZE: print(缓冲区满了生产者等待) condition.wait() queue.append(i) print(f生产者放入 {i}当前队列 {len(queue)}) condition.notify() # 唤醒一个消费者 time.sleep(0.2) def consumer(cid): for _ in range(10): with condition: while not queue: print(f消费者{cid}队列为空等待) condition.wait() item queue.popleft() print(f消费者{cid} 取出 {item}) condition.notify() time.sleep(0.3) t1 threading.Thread(targetproducer) t2 threading.Thread(targetconsumer, args(1,)) t2.start() t1.start()这里有个重要细节wait()一定要放在while循环里而不是if里。原因是即使线程被notify()唤醒了也可能因为调度原因再次竞争失败或者被虚假唤醒所以唤醒之后必须重新检查条件。notify()只唤醒一个等待线程notify_all()唤醒所有。如果你的场景里有一个生产者多个消费者而生产者放入一条数据只希望一个消费者去处理用notify()如果是某种配置变更需要通知所有消费者用notify_all()。3.5 线程安全的Queue能不用锁就不用锁严格来说queue.Queue不是同步原语但它内部已经实现了锁和条件变量是线程安全的直接拿来用就行。我实际开发中大量生产者消费者场景根本不手动加锁直接用Queue就够了from queue import Queue import threading import time task_queue Queue(maxsize20) def producer(): for i in range(100): task_queue.put(f任务{i}) task_queue.put(None) # 哨兵表示结束 def consumer(cid): while True: task task_queue.get() if task is None: task_queue.task_done() break print(f消费者{cid} 处理{task}) time.sleep(0.1) task_queue.task_done()put()在队列满时报错raise Full但推荐用put_nowait()配合逻辑处理或者设置put(blockTrue)自然阻塞。get()同理。加哨兵比如None结束消费者是常见的规范做法比给消费者下任务偷偷数任务数量要可靠得多。Queue还有个很有用的方法join()配合task_done()可以让主线程等待队列被消费完毕。task_queue.join()会阻塞直到所有已放入的任务都被task_done()这个在生产-消费的收尾阶段特别好用能确保所有任务真正处理完再退出程序。4. ThreadPoolExecutor别再手动管理线程了4.1 为什么推荐用线程池而不是自己开线程手动创建Thread对象每个任务都开一个线程当任务数量多、时间不规律的时候线程的创建销毁成本会很高而且你很难控制并发上限稍不注意就开几百个线程把内存吃满。线程池的核心思路是预先创建一批线程任务来了就往池子里丢线程复用既控制了并发数量又避免了频繁创建销毁的开销。Python从3.2开始提供了一个非常好用的高层接口concurrent.futures.ThreadPoolExecutor。比起手动管理Thread列表它最大的优势是返回Future对象可以统一管理和获取结果。4.2 submit与map的区别以及as_completed的用法先看最基础的两个提交方式from concurrent.futures import ThreadPoolExecutor, as_completed import time def get_data(city): time.sleep(0.5) return f{city} 的天气数据 cities [北京, 上海, 广州, 深圳, 杭州] # 方式一map保持输入顺序输出 with ThreadPoolExecutor(max_workers3) as pool: results pool.map(get_data, cities) for result in results: print(result) # 方式二submit as_completed谁先完成谁先输出 with ThreadPoolExecutor(max_workers3) as pool: futures {pool.submit(get_data, city): city for city in cities} for future in as_completed(futures): city futures[future] print(f{city} 的请求完成结果{future.result()})map()会按输入的顺序返回结果但它是同步等待的如果第一个任务很慢后面就算先完成也得排队等着。submit()as_completed()则是按完成顺序输出如果一个任务特别慢不影响其他任务的返回。真实业务中我一般推荐后者因为你可以每完成一个就立刻处理它的结果不用等所有任务做完。future.result()还有一个特性如果任务抛出异常你调用result()的时候会重新抛出这个异常所以必须在result()周围做异常处理。用as_completed遍历的时候每个future.result()都可能抛异常最好放在try/except里。4.3 线程池的大小怎么定线程池max_workers没有标准答案我根据场景给几个参考值IO密集型任务HTTP请求、读文件、查数据库线程等待时间占比高max_workers可以设大一些通常是CPU核心数的2到5倍甚至可以到几十。比如一个爬虫项目主要开销在网络等待我经常设20到50个线程。如果你的IO操作有外部并发限制比如数据库连接池只有10个连接线程数就设为连接数上限别多设多设只会排队。如果需要精确控制可以用ThreadPoolExecutor(max_workersmin(32, os.cpu_count() 4))这是官方文档给出的IO密集建议值。线程池不是越大越好。线程太多会让操作系统频繁做上下文切换反而拖慢整体性能。我见过有人一个爬虫开了300个线程结果目标网站服务器先撑不住了自己的机器CPU也飙升这属于设计问题。4.4 注意with语句退出时的行为ThreadPoolExecutor支持上下文管理器with块退出时会自动调用shutdown(waitTrue)也就是等所有已提交的任务执行完才退出。如果你不想等待可以显式调用shutdown(waitFalse)。这里有个细节with退出时已经提交但还没开始执行的任务也会被执行完而不是丢弃。这是shutdown()的默认行为。所以如果你提交了大量任务但中途想取消光靠with是不行的得用future.cancel()——而且cancel()只能取消还没开始运行的任务已经开始跑的没办法中途停。5. 彻底理解多进程池ProcessPoolExecutor和进程通信5.1 多进程解决的是CPU密集型问题之前说过多线程受GIL限制CPU密集任务跑不快。多进程绕过了GIL——每个进程有独立的Python解释器实例可以并行使用多个CPU核心。一个典型的例子是用多进程计算大量数据的平方和from concurrent.futures import ProcessPoolExecutor import os def heavy_compute(n): total 0 for i in range(n): total i * i return total if __name__ __main__: tasks [10_000_000, 10_000_000, 10_000_000, 10_000_000] with ProcessPoolExecutor(max_workers4) as pool: results list(pool.map(heavy_compute, tasks)) print(results)注意这里必须有if __name__ __main__保护。Windows上的多进程默认用spawn方式创建子进程它需要重新导入主模块来初始化没有这个保护的话子进程会把创建子进程的代码再执行一遍造成无限递归。5.2 进程池大小该设多少进程池的大小核心依据是CPU核心数。max_workersos.cpu_count()是常见做法但要注意如果你的机器同时还在跑其他负载最好留一个核心给操作系统和其他程序。我一般用os.cpu_count() - 1作为CPU密集任务的默认值有时候任务特别重还会减得更多。还有一个大家容易忽略的进程池越大内存开销越大。每个Python进程都有自己的内存空间如果任务本身内存占用大进程开多了很容易OOM。比如处理大图片的程序每个进程可能占几百MB内存8个进程就是好几个GB。5.3 进程间通信Queue、Pipe、Manager进程之间不共享内存通信需要通过专门的机制。multiprocessing.Queue是最常用的用法和queue.Queue基本一致from multiprocessing import Process, Queue def producer(q): for i in range(10): q.put(i) q.put(None) # 结束信号 def consumer(q): while True: item q.get() if item is None: break print(f消费到{item}) if __name__ __main__: q Queue() p1 Process(targetproducer, args(q,)) p2 Process(targetconsumer, args(q,)) p1.start() p2.start() p1.join() p2.join()multiprocessing.Pipe更适合两个进程之间的双向通信但底层传输的实际上是经过序列化的数据注意大对象传输的效率问题。Manager则提供了跨进程共享的列表、字典等容器用起来像本地对象但每次访问都要经过序列化和进程间通信性能比本地的list和dict慢很多适合少量数据共享。from multiprocessing import Process, Manager def modify(dct, lst): dct[key] value lst.append(42) if __name__ __main__: with Manager() as manager: dct manager.dict() lst manager.list() p Process(targetmodify, args(dct, lst)) p.start() p.join() print(dct, lst)5.4 进程同步共享内存加锁用multiprocessing.Value或Array可以共享内存中的数据但这个共享同样需要锁保护否则多个进程同时写共享内存数据会乱。这个锁和线程锁是一个套路from multiprocessing import Process, Value, Lock def add(shared_value, lock): for _ in range(100_000): with lock: shared_value.value 1 if __name__ __main__: v Value(i, 0) lock Lock() procs [Process(targetadd, args(v, lock)) for _ in range(4)] for p in procs: p.start() for p in procs: p.join() print(v.value)5.5 ProcessPoolExecutor的序列化这个坑用ProcessPoolExecutor提交任务时函数和参数都要被pickle序列化后传到子进程。这意味着lambda表达式不能被pickle所以不能往进程池提交lambda。类的方法通常也不能直接pickle除非类的定义在模块顶层。传大对象到进程池开销很大可能要花大量时间在序列化上最后发现比直接串行还慢。我有个朋友用ProcessPoolExecutor处理一批几MB级别的字符串列表结果每个任务传字符串的时间比计算时间还长。后来改成让子进程从共享文件或者数据库读数据只传索引性能才正常。所以用进程池的时候尽量传小参数大对象放共享存储里。6. 线程同步与池化中的高频问题和排查实录6.1 死锁的排查思路死锁最常见的原因有两个锁的获取顺序不一致或者持锁期间调用了一个也试图获取同一把锁的逻辑。排查方法我总结了一个规律先看看有没有嵌套加锁的代码如果有改造成只在一个方向上加锁或者改用RLock。给锁加超时lock.acquire(timeout5)超时会抛出异常或者返回False至少能让程序告诉你锁在哪个位置卡住了。这个调试技巧帮我定位过不少问题。用threading.stack_size()加faulthandler很难直接看锁状态所以更推荐的是在项目早期就把所有锁的获取都加上超时机制线上出问题能快速预警。实际遇到死锁的时候pdb很难定位因为卡住的时候不一定在出错的地方。我的经验是先减少锁的粒度把锁保护的代码块缩短能不加锁就不加锁能用Queue就不用锁从根源上减少死锁的可能性。很多并发问题凭空分析很难但你在代码中加足够的日志包括加锁释放锁的日志跑一遍就能看出在哪里互相等待了。6.2 多线程程序为何性能反而变差这个问题非常常见。如果你用多线程跑CPU密集任务因为有GIL性能不仅不提升还会因为线程切换带来额外开销。判断方法很简单把max_workers从1改成8如果你的程序运行时间反而变长了那它大概率就是CPU密集的应该换成ProcessPoolExecutor。如果任务是IO密集但性能还是提不上去通常是以下原因线程数不够IO等待没被充分利用。你用了全局锁保护了整个业务流程锁竞争严重等于全串行。你的IO操作其实是同步的外部库底层的调用没有释放GIL。排查的话先用cProfile来看每个线程的时间分布如果大量时间花在lock.acquire上说明锁竞争太激烈需要降低锁粒度。6.3 为什么进程池没在Windows上跑起来Windows默认用spawn方式创建进程这就意味着子进程会重新导入主模块。如果你没有在最外层加上if __name__ __main__保护子进程导入主模块时会再次执行创建子进程的代码递归创建进程最后直接报错或者崩溃。Linux和macOS默认是fork直接从父进程内存拷贝不需要重新导入主模块但也建议养成写if __name__ __main__的习惯保证代码跨平台能跑。6.4 CtrlC无法退出程序多线程程序经常遇到一个现象按CtrlC没反应或者程序卡在join()那里不退出。原因是默认情况下KeyboardInterrupt只发送给主线程如果子线程还活着主线程的join()不会被打断。我的处理办法有两种import signal import threading # 方法一给子线程设置daemonTrue # 方法二用event控制线程退出 stop_event threading.Event() def worker(): while not stop_event.is_set(): # 干活的逻辑 pass t threading.Thread(targetworker, daemonTrue) t.start()线程里循环判断stop_event收到退出信号就stop_event.set()这样既能让线程优雅退出又能让主线程及时退出。进程池那边一般用pool.shutdown(waitFalse)配合主进程的退出逻辑。6.5 一次数据竞争问题的完整排查过程说一个我实际排查过的模拟项目。某数据分析系统用多线程抓取多个数据源抓回来的数据汇总写入同一个统计表。上线后经常出现统计数字对不上的情况有时多统计有时少统计。排查过程是这样的先在写统计表的位置加了锁问题还在。仔细看发现读取当前统计值-加增量-写回是整个统计的核心流程虽然我加了锁但锁加在了外层而统计值本身是通过一个外部方法返回的读和写在代码里分开了中间插入了很多日志和校验操作。把读-改-写的整个过程压缩到一个独立函数里整个函数都加锁问题解决。这个案例说明一个关键原则锁保护的范围必须是完整的临界区不是只保护最后一行写操作。只要有读到旧值别人也读到旧值两人各自计算后写回的窗口数据就会错乱。7. 实际工作中我推荐的并发组合方案7.1 一套覆盖多数业务场景的通用模板根据我这几年的经验绝大多数Python项目其实只需要三个组件ThreadPoolExecutor处理IO密集、ProcessPoolExecutor处理CPU密集、queue.Queue或者multiprocessing.Queue串接任务流。我在这里给一个整合思路。比如一个数据管道批量读文件 - 解析处理CPU密集 - 写入数据库IO密集。最优方案是多进程池负责解析解析结果放进队列然后线程池负责消费队列写数据库。这样每个环节都可以独立调优。from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from queue import Queue import time # 用队列连接两个阶段 stage_queue Queue(maxsize200) def parse_data(filename): # CPU密集解析大文件 time.sleep(1) return f解析结果 {filename} def write_db(data): # IO密集写库 time.sleep(0.2) return f已写入{data} if __name__ __main__: filenames [ffile_{i}.csv for i in range(20)] with ProcessPoolExecutor(max_workers4) as cpu_pool: parse_futures [cpu_pool.submit(parse_data, fn) for fn in filenames] with ThreadPoolExecutor(max_workers4) as io_pool: write_futures [] for future in parse_futures: data future.result() write_futures.append(io_pool.submit(write_db, data)) for f in write_futures: f.result() print(全部完成)这个模板虽然简单但已经把两个池的定位用清楚了CPU密集的任务绝对不跟线程池混在一起IO部分用线程池避免进程池里的IO等待浪费进程资源。7.2 池子的数量到底应该怎么评估除了之前提到的os.cpu_count()还有一个值得参考的经验看任务本身的耗时比例。用IO密集的网络请求做例子假设一次请求的纯CPU消耗只有10%等待占90%那么理论上线程数可以设到CPU核心数的10到20倍都不会太离谱。但这只是理论上限实际还要考虑内存和系统限制。我推荐用一个简单公式线程数 目标吞吐量 / 单任务每秒可处理数。如果目标是每秒处理100个请求单线程每秒能处理10个那就需要10个线程。这个比拍脑袋设一个数字科学得多。进程池就不一样了我很少看到进程池数量需要超过CPU核心数的场景。如果任务真的太多就分批处理不要让进程池排着几百个任务等着——那样不如把任务顺序喂进去保持流水线式处理。8. 收尾我这些年最实在的几个提醒做并发编程这些年最大的体会就是并发不是写出来就完了而是要跑不崩、退得掉、数据对。很多项目第一版能跑上线之后数据一多就发现各种问题归根到底是在设计阶段没有想清楚用线程还是进程、锁的粒度在哪里、如何优雅退出。最后分享两个小技巧。第一个是开发的时候在关键临界区和锁操作附近加日志打印线程名和时间戳。出问题的时候这些日志就是第一手排查资料。第二个是给自己写一个通用的并发包装函数把submit、as_completed、超时和异常处理封装起来后续项目直接调能省下不少反复排查的时间。如果你刚开始学并发我建议从线程池Queue开始把最基础的场景跑通再逐步尝试进程池和复杂的同步原语。先把能用的工具用熟练再深入研究背后的原理这条路最稳。
Python并发编程实战多线程池、多进程池与线程同步的那些坑和技巧这几年写Python几乎每个项目都会碰到同一个问题并发量上来了程序变慢了第一反应就是开线程。但真正把多线程、多进程、线程池进程池、锁和队列这些东西全部摸清楚我觉得至少得踩三四个实际的坑才算入门。这篇文章就是把我这些年在项目里折腾线程同步、线程池和进程池的经验整理出来包含可以直接抄的代码、参数选择思路以及常规文档里不会告诉你的排查方法。适合刚接触并发编程的初学者也适合写了一段时间多线程但总觉得哪里不对劲的开发者。说到底Python并发要解决的就两个问题让IO等待时间不再白费让多核CPU真正跑起来。前者用多线程后者用多进程。但这中间涉及GIL、共享状态、死锁、序列化开销每一个都能让程序在线上突然卡死。所以我从最基础的概念讲起一步步带你走完线程、同步、池化、进程通信整个链路。1. 先搞清楚核心问题Python并发到底在解决什么1.1 GIL是绕不开的第一道坎很多新手看教程上来就学Thread然后用线程跑一个纯计算的函数发现两个线程加起来比单线程还慢就一脸懵。这个现象背后就是GIL全局解释器锁。Python解释器同一时刻只允许一个线程执行Python字节码也就是说如果是纯CPU计算任务多线程等于把并发执行变成串行还额外付出线程切换的开销。真正让多线程有用的场景是IO密集型任务——比如网络请求、文件读写、数据库查询。这些操作在等待外部响应的时候会主动释放GIL让其他线程趁这个机会跑起来这时候多线程就能实打实地提升吞吐量。我自己的判断习惯是任务里有明显的阻塞等待网络、磁盘、用户输入用多线程任务全是CPU运算图像处理、加解密、大规模数学计算直接上多进程。别指望一条路走到黑。1.2 线程和进程各管哪一片线程共享进程的内存空间创建成本低切换快但是共享变量的时候容易出现数据竞争得加锁保护。进程拥有独立的内存空间天然隔离不需要担心变量被别的进程乱改但进程创建开销大进程间通信还要靠队列、管道这些专门的机制。用一个生活化的类比线程像一家人共用一个厨房大家共用锅碗瓢盆做饭快但容易抢东西打架进程像每家每户有独立厨房互不干扰但你要借个酱油得走门串户。所以在设计并发方案的时候我通常先问三个问题任务是IO密集还是CPU密集任务之间需不需要共享数据共享的数据量有多大单个任务的耗时是秒级还是毫秒级这三个问题回答完用线程池还是进程池基本就定了。2. threading基础线程的创建、生命周期与正确关闭2.1 最基础的线程写法和为什么必须join先看一个最简单的线程模型import threading import time def worker(name, seconds): print(f线程 {name} 开始工作预计耗时 {seconds}s) time.sleep(seconds) print(f线程 {name} 工作完成) threads [] for i in range(3): t threading.Thread(targetworker, args(fworker-{i}, 2)) threads.append(t) t.start() # 注意是start不是run for t in threads: t.join() # 等待所有线程结束主线程才能继续往下走 print(所有线程结束主程序退出)这里有两个新手必踩的坑。第一Thread(target...)只是创建了线程对象一定要调用start()才会真正启动如果直接调用run()那就变成了普通函数调用完全串行执行。第二主线程默认不会等待子线程结束如果你不调join()可能主线程已经跑完了子线程还在后台干活程序就直接退出了。join()还支持超时参数比如t.join(timeout5)表示最多等5秒5秒后还没结束就继续往下走。这个在关闭资源、优雅退出的时候非常有用——你不能无限期等一个可能卡死的线程。2.2 守护线程daemon退出程序时要不要强制回收守护线程的概念很简单daemonTrue的线程在主线程退出时会跟着一起被强制终止不管它活没干完。非守护线程则会在主线程退出后仍然继续运行直到它自己结束。t threading.Thread(targetworker, args(daemon-task, 30), daemonTrue) t.start() # 主线程做完事情直接退出daemon线程被强制中断实际项目中日志监听、心跳上报、看门狗这类后台任务设成daemon比较合理——你不想因为一个后台线程卡住导致整个程序退不了。但如果是正在写文件、提交数据库事务这种任务千万别设成daemon否则程序退出的时候数据可能丢一半。这是我在一个数据采集项目里踩过的坑当时为了图省事把采集线程设成了daemon结果是进程退得倒是干净数据库里留下了一堆不完整记录。2.3 共享变量要用锁一个简单的计数器引发的血案先看一个经典问题。两个线程同时对同一个全局变量做count 1每个线程循环100万次你觉得结果是多少import threading count 0 def increase(): global count for _ in range(1_000_000): count 1 threads [threading.Thread(targetincrease) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(count) # 结果往往不是2000000我第一次跑这个例子的时候输出的是124万左右少了几十万。原因很简单count 1在底层是三条指令——读取count、加1、写回count。两个线程可能同时读到同一个值各自加1再写回结果有一方的修改被覆盖了。这个就是数据竞争。解决办法就是加锁把读取-修改-写回做成不可分割的原子操作import threading count 0 lock threading.Lock() def increase(): global count for _ in range(1_000_000): with lock: # 等价于lock.acquire() finally lock.release() count 1用with lock而不是手动acquire()/release()好处是就算代码中途抛异常锁也会被自动释放不会出现拿到锁之后忘了解锁导致死锁的情况。这个习惯我从踩过一次死锁之后就一直保持了。3. 线程同步机制Lock、RLock、Semaphore、Event、Condition怎么选3.1 Lock与RLock可重入锁解决递归加锁问题Lock也叫互斥锁同一时刻只能被一个线程持有。但如果同一个线程在持锁期间又去acquire()同一把锁就会自己把自己卡死。看这个场景import threading lock threading.Lock() def outer(): lock.acquire() inner() # 内部又去acquire同一把锁 lock.release() def inner(): lock.acquire() # 死锁 lock.release()这种情况用RLock可重入锁就解决了。RLock允许同一个线程多次acquire()每次acquire()对应一次release()只有计数归零才算真正释放。在实际项目里一个函数内部嵌套调用另一个也加了锁的函数是非常常见的所以我的项目里凡是涉及同一线程可能递归进入的锁一律用RLock省得排查半天是这个原因导致的卡死。3.2 Semaphore限制并发数的信号量Semaphore用来控制同时访问某个资源的线程数量。比如你的程序最多允许5个线程同时连数据库不然数据库会被打爆import threading import time semaphore threading.Semaphore(5) def db_query(query_id): with semaphore: print(f查询 {query_id} 开始占用一个连接) time.sleep(1) print(f查询 {query_id} 结束释放连接) threads [threading.Thread(targetdb_query, args(i,)) for i in range(20)] for t in threads: t.start() for t in threads: t.join()Semaphore(5)表示初始允许5个线程通行第6个线程调用acquire()时会阻塞直到前面某个线程release()。注意一点如果初始值设成1Semaphore就是一把Lock。还有BoundedSemaphore它在release()次数超过acquire()次数时抛异常能帮你发现代码逻辑错误普通Semaphore则不会管这个。3.3 Event一个线程等另一个线程的信号Event适合一个线程等待另一个线程完成某个动作的场景比如主线程等待所有工作线程初始化完成再下发任务import threading import time ready_event threading.Event() def worker(worker_id): print(fworker-{worker_id} 正在初始化) time.sleep(1) print(fworker-{worker_id} 初始化完成) ready_event.set() # 发信号 def main(): t threading.Thread(targetworker, args(1,)) t.start() ready_event.wait() # 阻塞等待信号 print(所有初始化完成主线程开始正式干活) main()Event的核心就两个方法wait()阻塞等待set()发出信号让所有等待的线程通过clear()把事件重置。有一个细节是clear()很容易被忽略——如果事件是一次性的不用管如果是周期性用的用完之后必须调clear()不然下一次wait()直接通过了逻辑就全乱了。3.4 Condition生产者消费者模型的标准解法Condition是更高级的同步原语本质上是锁加等待通知机制。最经典的应用就是生产者消费者模型生产者往缓冲区放数据消费者从缓冲区取数据缓冲区满的时候生产者要等缓冲区空的时候消费者要等。import threading import time from collections import deque condition threading.Condition() queue deque() MAX_SIZE 5 def producer(): for i in range(20): with condition: while len(queue) MAX_SIZE: print(缓冲区满了生产者等待) condition.wait() queue.append(i) print(f生产者放入 {i}当前队列 {len(queue)}) condition.notify() # 唤醒一个消费者 time.sleep(0.2) def consumer(cid): for _ in range(10): with condition: while not queue: print(f消费者{cid}队列为空等待) condition.wait() item queue.popleft() print(f消费者{cid} 取出 {item}) condition.notify() time.sleep(0.3) t1 threading.Thread(targetproducer) t2 threading.Thread(targetconsumer, args(1,)) t2.start() t1.start()这里有个重要细节wait()一定要放在while循环里而不是if里。原因是即使线程被notify()唤醒了也可能因为调度原因再次竞争失败或者被虚假唤醒所以唤醒之后必须重新检查条件。notify()只唤醒一个等待线程notify_all()唤醒所有。如果你的场景里有一个生产者多个消费者而生产者放入一条数据只希望一个消费者去处理用notify()如果是某种配置变更需要通知所有消费者用notify_all()。3.5 线程安全的Queue能不用锁就不用锁严格来说queue.Queue不是同步原语但它内部已经实现了锁和条件变量是线程安全的直接拿来用就行。我实际开发中大量生产者消费者场景根本不手动加锁直接用Queue就够了from queue import Queue import threading import time task_queue Queue(maxsize20) def producer(): for i in range(100): task_queue.put(f任务{i}) task_queue.put(None) # 哨兵表示结束 def consumer(cid): while True: task task_queue.get() if task is None: task_queue.task_done() break print(f消费者{cid} 处理{task}) time.sleep(0.1) task_queue.task_done()put()在队列满时报错raise Full但推荐用put_nowait()配合逻辑处理或者设置put(blockTrue)自然阻塞。get()同理。加哨兵比如None结束消费者是常见的规范做法比给消费者下任务偷偷数任务数量要可靠得多。Queue还有个很有用的方法join()配合task_done()可以让主线程等待队列被消费完毕。task_queue.join()会阻塞直到所有已放入的任务都被task_done()这个在生产-消费的收尾阶段特别好用能确保所有任务真正处理完再退出程序。4. ThreadPoolExecutor别再手动管理线程了4.1 为什么推荐用线程池而不是自己开线程手动创建Thread对象每个任务都开一个线程当任务数量多、时间不规律的时候线程的创建销毁成本会很高而且你很难控制并发上限稍不注意就开几百个线程把内存吃满。线程池的核心思路是预先创建一批线程任务来了就往池子里丢线程复用既控制了并发数量又避免了频繁创建销毁的开销。Python从3.2开始提供了一个非常好用的高层接口concurrent.futures.ThreadPoolExecutor。比起手动管理Thread列表它最大的优势是返回Future对象可以统一管理和获取结果。4.2 submit与map的区别以及as_completed的用法先看最基础的两个提交方式from concurrent.futures import ThreadPoolExecutor, as_completed import time def get_data(city): time.sleep(0.5) return f{city} 的天气数据 cities [北京, 上海, 广州, 深圳, 杭州] # 方式一map保持输入顺序输出 with ThreadPoolExecutor(max_workers3) as pool: results pool.map(get_data, cities) for result in results: print(result) # 方式二submit as_completed谁先完成谁先输出 with ThreadPoolExecutor(max_workers3) as pool: futures {pool.submit(get_data, city): city for city in cities} for future in as_completed(futures): city futures[future] print(f{city} 的请求完成结果{future.result()})map()会按输入的顺序返回结果但它是同步等待的如果第一个任务很慢后面就算先完成也得排队等着。submit()as_completed()则是按完成顺序输出如果一个任务特别慢不影响其他任务的返回。真实业务中我一般推荐后者因为你可以每完成一个就立刻处理它的结果不用等所有任务做完。future.result()还有一个特性如果任务抛出异常你调用result()的时候会重新抛出这个异常所以必须在result()周围做异常处理。用as_completed遍历的时候每个future.result()都可能抛异常最好放在try/except里。4.3 线程池的大小怎么定线程池max_workers没有标准答案我根据场景给几个参考值IO密集型任务HTTP请求、读文件、查数据库线程等待时间占比高max_workers可以设大一些通常是CPU核心数的2到5倍甚至可以到几十。比如一个爬虫项目主要开销在网络等待我经常设20到50个线程。如果你的IO操作有外部并发限制比如数据库连接池只有10个连接线程数就设为连接数上限别多设多设只会排队。如果需要精确控制可以用ThreadPoolExecutor(max_workersmin(32, os.cpu_count() 4))这是官方文档给出的IO密集建议值。线程池不是越大越好。线程太多会让操作系统频繁做上下文切换反而拖慢整体性能。我见过有人一个爬虫开了300个线程结果目标网站服务器先撑不住了自己的机器CPU也飙升这属于设计问题。4.4 注意with语句退出时的行为ThreadPoolExecutor支持上下文管理器with块退出时会自动调用shutdown(waitTrue)也就是等所有已提交的任务执行完才退出。如果你不想等待可以显式调用shutdown(waitFalse)。这里有个细节with退出时已经提交但还没开始执行的任务也会被执行完而不是丢弃。这是shutdown()的默认行为。所以如果你提交了大量任务但中途想取消光靠with是不行的得用future.cancel()——而且cancel()只能取消还没开始运行的任务已经开始跑的没办法中途停。5. 彻底理解多进程池ProcessPoolExecutor和进程通信5.1 多进程解决的是CPU密集型问题之前说过多线程受GIL限制CPU密集任务跑不快。多进程绕过了GIL——每个进程有独立的Python解释器实例可以并行使用多个CPU核心。一个典型的例子是用多进程计算大量数据的平方和from concurrent.futures import ProcessPoolExecutor import os def heavy_compute(n): total 0 for i in range(n): total i * i return total if __name__ __main__: tasks [10_000_000, 10_000_000, 10_000_000, 10_000_000] with ProcessPoolExecutor(max_workers4) as pool: results list(pool.map(heavy_compute, tasks)) print(results)注意这里必须有if __name__ __main__保护。Windows上的多进程默认用spawn方式创建子进程它需要重新导入主模块来初始化没有这个保护的话子进程会把创建子进程的代码再执行一遍造成无限递归。5.2 进程池大小该设多少进程池的大小核心依据是CPU核心数。max_workersos.cpu_count()是常见做法但要注意如果你的机器同时还在跑其他负载最好留一个核心给操作系统和其他程序。我一般用os.cpu_count() - 1作为CPU密集任务的默认值有时候任务特别重还会减得更多。还有一个大家容易忽略的进程池越大内存开销越大。每个Python进程都有自己的内存空间如果任务本身内存占用大进程开多了很容易OOM。比如处理大图片的程序每个进程可能占几百MB内存8个进程就是好几个GB。5.3 进程间通信Queue、Pipe、Manager进程之间不共享内存通信需要通过专门的机制。multiprocessing.Queue是最常用的用法和queue.Queue基本一致from multiprocessing import Process, Queue def producer(q): for i in range(10): q.put(i) q.put(None) # 结束信号 def consumer(q): while True: item q.get() if item is None: break print(f消费到{item}) if __name__ __main__: q Queue() p1 Process(targetproducer, args(q,)) p2 Process(targetconsumer, args(q,)) p1.start() p2.start() p1.join() p2.join()multiprocessing.Pipe更适合两个进程之间的双向通信但底层传输的实际上是经过序列化的数据注意大对象传输的效率问题。Manager则提供了跨进程共享的列表、字典等容器用起来像本地对象但每次访问都要经过序列化和进程间通信性能比本地的list和dict慢很多适合少量数据共享。from multiprocessing import Process, Manager def modify(dct, lst): dct[key] value lst.append(42) if __name__ __main__: with Manager() as manager: dct manager.dict() lst manager.list() p Process(targetmodify, args(dct, lst)) p.start() p.join() print(dct, lst)5.4 进程同步共享内存加锁用multiprocessing.Value或Array可以共享内存中的数据但这个共享同样需要锁保护否则多个进程同时写共享内存数据会乱。这个锁和线程锁是一个套路from multiprocessing import Process, Value, Lock def add(shared_value, lock): for _ in range(100_000): with lock: shared_value.value 1 if __name__ __main__: v Value(i, 0) lock Lock() procs [Process(targetadd, args(v, lock)) for _ in range(4)] for p in procs: p.start() for p in procs: p.join() print(v.value)5.5 ProcessPoolExecutor的序列化这个坑用ProcessPoolExecutor提交任务时函数和参数都要被pickle序列化后传到子进程。这意味着lambda表达式不能被pickle所以不能往进程池提交lambda。类的方法通常也不能直接pickle除非类的定义在模块顶层。传大对象到进程池开销很大可能要花大量时间在序列化上最后发现比直接串行还慢。我有个朋友用ProcessPoolExecutor处理一批几MB级别的字符串列表结果每个任务传字符串的时间比计算时间还长。后来改成让子进程从共享文件或者数据库读数据只传索引性能才正常。所以用进程池的时候尽量传小参数大对象放共享存储里。6. 线程同步与池化中的高频问题和排查实录6.1 死锁的排查思路死锁最常见的原因有两个锁的获取顺序不一致或者持锁期间调用了一个也试图获取同一把锁的逻辑。排查方法我总结了一个规律先看看有没有嵌套加锁的代码如果有改造成只在一个方向上加锁或者改用RLock。给锁加超时lock.acquire(timeout5)超时会抛出异常或者返回False至少能让程序告诉你锁在哪个位置卡住了。这个调试技巧帮我定位过不少问题。用threading.stack_size()加faulthandler很难直接看锁状态所以更推荐的是在项目早期就把所有锁的获取都加上超时机制线上出问题能快速预警。实际遇到死锁的时候pdb很难定位因为卡住的时候不一定在出错的地方。我的经验是先减少锁的粒度把锁保护的代码块缩短能不加锁就不加锁能用Queue就不用锁从根源上减少死锁的可能性。很多并发问题凭空分析很难但你在代码中加足够的日志包括加锁释放锁的日志跑一遍就能看出在哪里互相等待了。6.2 多线程程序为何性能反而变差这个问题非常常见。如果你用多线程跑CPU密集任务因为有GIL性能不仅不提升还会因为线程切换带来额外开销。判断方法很简单把max_workers从1改成8如果你的程序运行时间反而变长了那它大概率就是CPU密集的应该换成ProcessPoolExecutor。如果任务是IO密集但性能还是提不上去通常是以下原因线程数不够IO等待没被充分利用。你用了全局锁保护了整个业务流程锁竞争严重等于全串行。你的IO操作其实是同步的外部库底层的调用没有释放GIL。排查的话先用cProfile来看每个线程的时间分布如果大量时间花在lock.acquire上说明锁竞争太激烈需要降低锁粒度。6.3 为什么进程池没在Windows上跑起来Windows默认用spawn方式创建进程这就意味着子进程会重新导入主模块。如果你没有在最外层加上if __name__ __main__保护子进程导入主模块时会再次执行创建子进程的代码递归创建进程最后直接报错或者崩溃。Linux和macOS默认是fork直接从父进程内存拷贝不需要重新导入主模块但也建议养成写if __name__ __main__的习惯保证代码跨平台能跑。6.4 CtrlC无法退出程序多线程程序经常遇到一个现象按CtrlC没反应或者程序卡在join()那里不退出。原因是默认情况下KeyboardInterrupt只发送给主线程如果子线程还活着主线程的join()不会被打断。我的处理办法有两种import signal import threading # 方法一给子线程设置daemonTrue # 方法二用event控制线程退出 stop_event threading.Event() def worker(): while not stop_event.is_set(): # 干活的逻辑 pass t threading.Thread(targetworker, daemonTrue) t.start()线程里循环判断stop_event收到退出信号就stop_event.set()这样既能让线程优雅退出又能让主线程及时退出。进程池那边一般用pool.shutdown(waitFalse)配合主进程的退出逻辑。6.5 一次数据竞争问题的完整排查过程说一个我实际排查过的模拟项目。某数据分析系统用多线程抓取多个数据源抓回来的数据汇总写入同一个统计表。上线后经常出现统计数字对不上的情况有时多统计有时少统计。排查过程是这样的先在写统计表的位置加了锁问题还在。仔细看发现读取当前统计值-加增量-写回是整个统计的核心流程虽然我加了锁但锁加在了外层而统计值本身是通过一个外部方法返回的读和写在代码里分开了中间插入了很多日志和校验操作。把读-改-写的整个过程压缩到一个独立函数里整个函数都加锁问题解决。这个案例说明一个关键原则锁保护的范围必须是完整的临界区不是只保护最后一行写操作。只要有读到旧值别人也读到旧值两人各自计算后写回的窗口数据就会错乱。7. 实际工作中我推荐的并发组合方案7.1 一套覆盖多数业务场景的通用模板根据我这几年的经验绝大多数Python项目其实只需要三个组件ThreadPoolExecutor处理IO密集、ProcessPoolExecutor处理CPU密集、queue.Queue或者multiprocessing.Queue串接任务流。我在这里给一个整合思路。比如一个数据管道批量读文件 - 解析处理CPU密集 - 写入数据库IO密集。最优方案是多进程池负责解析解析结果放进队列然后线程池负责消费队列写数据库。这样每个环节都可以独立调优。from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor from queue import Queue import time # 用队列连接两个阶段 stage_queue Queue(maxsize200) def parse_data(filename): # CPU密集解析大文件 time.sleep(1) return f解析结果 {filename} def write_db(data): # IO密集写库 time.sleep(0.2) return f已写入{data} if __name__ __main__: filenames [ffile_{i}.csv for i in range(20)] with ProcessPoolExecutor(max_workers4) as cpu_pool: parse_futures [cpu_pool.submit(parse_data, fn) for fn in filenames] with ThreadPoolExecutor(max_workers4) as io_pool: write_futures [] for future in parse_futures: data future.result() write_futures.append(io_pool.submit(write_db, data)) for f in write_futures: f.result() print(全部完成)这个模板虽然简单但已经把两个池的定位用清楚了CPU密集的任务绝对不跟线程池混在一起IO部分用线程池避免进程池里的IO等待浪费进程资源。7.2 池子的数量到底应该怎么评估除了之前提到的os.cpu_count()还有一个值得参考的经验看任务本身的耗时比例。用IO密集的网络请求做例子假设一次请求的纯CPU消耗只有10%等待占90%那么理论上线程数可以设到CPU核心数的10到20倍都不会太离谱。但这只是理论上限实际还要考虑内存和系统限制。我推荐用一个简单公式线程数 目标吞吐量 / 单任务每秒可处理数。如果目标是每秒处理100个请求单线程每秒能处理10个那就需要10个线程。这个比拍脑袋设一个数字科学得多。进程池就不一样了我很少看到进程池数量需要超过CPU核心数的场景。如果任务真的太多就分批处理不要让进程池排着几百个任务等着——那样不如把任务顺序喂进去保持流水线式处理。8. 收尾我这些年最实在的几个提醒做并发编程这些年最大的体会就是并发不是写出来就完了而是要跑不崩、退得掉、数据对。很多项目第一版能跑上线之后数据一多就发现各种问题归根到底是在设计阶段没有想清楚用线程还是进程、锁的粒度在哪里、如何优雅退出。最后分享两个小技巧。第一个是开发的时候在关键临界区和锁操作附近加日志打印线程名和时间戳。出问题的时候这些日志就是第一手排查资料。第二个是给自己写一个通用的并发包装函数把submit、as_completed、超时和异常处理封装起来后续项目直接调能省下不少反复排查的时间。如果你刚开始学并发我建议从线程池Queue开始把最基础的场景跑通再逐步尝试进程池和复杂的同步原语。先把能用的工具用熟练再深入研究背后的原理这条路最稳。
