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(target=worker, args=(f"worker-{i}", 2)) threads.append(t) t.start() # 注意是start，不是run for t in threads: t.join() # 等待所有线程结束，主线程才能继续往下走 print("所有线程结束，主程序退出")这里有两个新手必踩的坑。第一，Thread(target=...)只是创建了线程对象，一定要调用start()才会真正启动；如果直接调用run()，那就变成了普通函数调用，完全串行执行。第二，主线程默认不会等待子线程结束，如果你不调join()，可能主线程已经跑完了，子线程还在后台干活，程序就直接退出了。
join()还支持超时参数，比如t.join(timeout=5)表示最多等5秒，5秒后还没结束就继续往下走。这个在关闭资源、优雅退出的时候非常有用——你不能无限期等一个可能卡死的线程。
2.2 守护线程daemon：退出程序时要不要强制回收
守护线程的概念很简单：daemon=True的线程，在主线程退出时会跟着一起被强制终止，不管它活没干完。非守护线程则会在主线程退出后仍然继续运行，直到它自己结束。
t = threading.Thread(target=worker, args=("daemon-task", 30), daemon=True) 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(target=increase) 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(target=db_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()。注意一点：如果初始值设成1，Semaphore就是一把Lock。还有BoundedSemaphore，它在release()次数超过acquire()次数时抛异常，能帮你发现代码逻辑错误，普通Semaphore则不会管这个。
3.3 Event：一个线程等另一个线程的信号
Event适合一个线程等待另一个线程完成某个动作的场景，比如主线程等待所有工作线程初始化完成再下发任务：
import threading import time ready_event = threading.Event() def worker(worker_id): print(f"worker-{worker_id} 正在初始化") time.sleep(1) print(f"worker-{worker_id} 初始化完成") ready_event.set() # 发信号 def main(): t = threading.Thread(target=worker, 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(target=producer) t2 = threading.Thread(target=consumer, 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(maxsize=20) 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(block=True)自然阻塞。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_workers=3) as pool: results = pool.map(get_data, cities) for result in results: print(result) # 方式二：submit + as_completed，谁先完成谁先输出 with ThreadPoolExecutor(max_workers=3) 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_workers=min(32, os.cpu_count() + 4))，这是官方文档给出的IO密集建议值。
线程池不是越大越好。线程太多会让操作系统频繁做上下文切换，反而拖慢整体性能。我见过有人一个爬虫开了300个线程，结果目标网站服务器先撑不住了，自己的机器CPU也飙升，这属于设计问题。
4.4 注意with语句退出时的行为
ThreadPoolExecutor支持上下文管理器，with块退出时会自动调用shutdown(wait=True)，也就是等所有已提交的任务执行完才退出。如果你不想等待，可以显式调用shutdown(wait=False)。
这里有个细节：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_workers=4) as pool: results = list(pool.map(heavy_compute, tasks)) print(results)注意这里必须有if __name__ == "__main__"保护。Windows上的多进程默认用spawn方式创建子进程，它需要重新导入主模块来初始化，没有这个保护的话，子进程会把创建子进程的代码再执行一遍，造成无限递归。
5.2 进程池大小该设多少
进程池的大小，核心依据是CPU核心数。max_workers=os.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(target=producer, args=(q,)) p2 = Process(target=consumer, 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(target=modify, 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(target=add, 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(timeout=5)，超时会抛出异常或者返回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 Ctrl+C无法退出程序
多线程程序经常遇到一个现象：按Ctrl+C没反应，或者程序卡在join()那里不退出。原因是默认情况下KeyboardInterrupt只发送给主线程，如果子线程还活着，主线程的join()不会被打断。
我的处理办法有两种：
import signal import threading # 方法一：给子线程设置daemon=True # 方法二：用event控制线程退出 stop_event = threading.Event() def worker(): while not stop_event.is_set(): # 干活的逻辑 pass t = threading.Thread(target=worker, daemon=True) t.start()线程里循环判断stop_event，收到退出信号就stop_event.set()，这样既能让线程优雅退出，又能让主线程及时退出。进程池那边，一般用pool.shutdown(wait=False)配合主进程的退出逻辑。
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(maxsize=200) 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 = [f"file_{i}.csv" for i in range(20)] with ProcessPoolExecutor(max_workers=4) as cpu_pool: parse_futures = [cpu_pool.submit(parse_data, fn) for fn in filenames] with ThreadPoolExecutor(max_workers=4) 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开始，把最基础的场景跑通，再逐步尝试进程池和复杂的同步原语。先把能用的工具用熟练，再深入研究背后的原理，这条路最稳。
