在 Python 里写并发，很多人的第一反应是开线程。但真正动手之前，必须先搞清楚一件事：GIL。CPython 解释器在同一时刻只允许一个线程执行 Python 字节码，这条限制直接决定了线程该怎么用。
一、先判断任务类型，再决定用不用线程
原文给出的判断标准很干脆：
- I/O 密集型（网络请求、文件读写、数据库查询）：线程在等待 I/O 时 GIL 会释放，多线程有效。
- CPU 密集型（大量计算、图像处理）：线程之间抢 GIL，多线程不但没用，还可能更慢。
一句话概括：Python 多线程适合"等"，不适合"算"。这个前提定下来，后面的代码才有意义。
二、创建线程的两种方式
最常用的是直接把函数交给 Thread，用 target 指定入口、args 传参：
- import threading
- import time
- def worker(name, delay):
- for i in range(3):
- print(f"{name} 第 {i+1} 次")
- time.sleep(delay)
- t1 = threading.Thread(target=worker, args=("A", 1))
- t2 = threading.Thread(target=worker, args=("B", 2))
- t1.start()
- t2.start()
- t1.join() # 等 t1 结束
- t2.join() # 等 t2 结束
- print("全部完成")
这里有两个关键方法：start() 负责启动线程，join() 负责阻塞等待线程结束。主线程如果不 join()，子线程可能还没跑完，程序就已经退出了。
第二种是继承 Thread 并重写 run()，适合需要把线程状态封装进对象的场景：
- class DownloadTask(threading.Thread):
- def __init__(self, url):
- super().__init__()
- self.url = url
- self.result = None
- def run(self):
- # 线程启动后自动执行
- print(f"下载：{self.url}")
- time.sleep(2)
- self.result = f"{self.url} 完成"
- task = DownloadTask("https://example.com/file.zip")
- task.start()
- task.join()
- print(task.result)
日常开发用第一种就够了，需要在线程对象上挂载结果或状态时，再考虑第二种。
三、守护线程：主程序退出就一起退
普通线程会阻止程序退出，主线程结束后还要等所有子线程收尾。把 daemon 设为 True 则相反，主线程一退出，守护线程被直接终止：
- def background():
- while True:
- print("后台运行...")
- time.sleep(1)
- t = threading.Thread(target=background, daemon=True)
- t.start()
- time.sleep(3)
- print("主程序结束") # 3 秒后退出，守护线程随之终止
典型用途是后台监控、心跳检测这类"主程序不跑了就没必要继续跑"的任务。
四、线程同步：共享变量为什么会被写乱
下面这段代码的执行结果会低于预期：
- counter = 0
- def increment():
- global counter
- for _ in range(100000):
- counter += 1
- t1 = threading.Thread(target=increment)
- t2 = threading.Thread(target=increment)
- t1.start()
- t2.start()
- t1.join()
- t2.join()
- print(counter) # 期望 200000，实际可能是 189xxx
原因是 counter += 1 并非原子操作，它由"读、算、写"三步组成，两个线程交替执行时会互相覆盖对方的结果。
解决办法是加锁。推荐用 with lock 的写法，离开代码块自动释放，不用手动 acquire() 和 release()，也就不会忘记释放：
- counter = 0
- lock = threading.Lock()
- def increment():
- global counter
- for _ in range(100000):
- with lock:
- counter += 1
- # 离开 with 自动释放锁
五、线程间通信优先用 Queue
多个线程之间传数据，不要自己拿锁去维护共享列表，直接用 queue.Queue，它内部已经实现了线程安全。生产者-消费者是最经典的场景：
- import queue
- import threading
- q = queue.Queue()
- def producer():
- for i in range(5):
- q.put(i)
- print(f"生产：{i}")
- q.put(None) # 结束信号
- def consumer():
- while True:
- item = q.get()
- if item is None:
- break
- print(f"消费：{item}")
- t1 = threading.Thread(target=producer)
- t2 = threading.Thread(target=consumer)
- t1.start()
- t2.start()
- t1.join()
- t2.join()
Queue 有三种形态：Queue 先进先出（默认）、LifoQueue 后进先出（栈）、PriorityQueue 按优先级取出。绝大多数情况用默认的 Queue 即可。
六、其他同步工具的定位
先知道有这些东西，用到时再查文档：
- Lock：互斥锁，同一时刻只允许一个线程访问，用于保护共享变量。
- RLock：可重入锁，同一线程可多次获取，用于递归函数里加锁。
- Semaphore：信号量，限制同时访问的线程数量，典型用法是限流，例如最多 3 个线程同时请求 API。
- Event：事件通知，一个线程等待另一个线程的信号，适合等待初始化完成或等待任务结束。
- Condition：条件变量，更灵活的通知机制，可用于生产者-消费者，但通常用 Queue 就够了。
实际开发中最高频的是 Lock 和 Queue 两个。
七、线程池：不要手动创建一堆线程
创建几个线程没问题，但要开几十上百个线程时，频繁创建销毁的开销就显现出来了。concurrent.futures 里的 ThreadPoolExecutor 负责复用线程：
- from concurrent.futures import ThreadPoolExecutor
- def task(n):
- return n * 2
- with ThreadPoolExecutor(max_workers=3) as executor:
- # 方式1：提交单个任务
- future = executor.submit(task, 5)
- print(future.result()) # 10
- # 方式2：批量提交
- results = executor.map(task, range(5))
- print(list(results)) # [0, 2, 4, 6, 8]
with 语句退出时会自动等待所有任务完成并清理线程池。submit 和 map 的区别在于：submit 提交单个任务并返回 Future 对象，可以单独取结果；map 批量提交，返回结果迭代器，输出顺序与输入顺序一致。
如果需要"谁先完成先处理谁"，用 as_completed：
- from concurrent.futures import ThreadPoolExecutor, as_completed
- with ThreadPoolExecutor(max_workers=3) as executor:
- futures = [executor.submit(task, i) for i in range(5)]
- for future in as_completed(futures):
- print(future.result())
八、方案选型对照
- 网络请求、爬虫：线程或线程池，I/O 等待期间释放 GIL。
- 文件读写：线程，属于 I/O 操作。
- 大量计算（CPU 密集）：multiprocessing 多进程，线程受 GIL 限制无法提速。
- 高并发网络服务：asyncio 异步 I/O，单线程更高效。
- 简单并发需求：concurrent.futures，接口统一，切换方案方便。
简化成一句判断逻辑：任务在等外部（网络、磁盘、数据库）就用线程；任务在算内部（循环、数学运算）就用多进程或异步。
九、完整示例：批量下载的串行与并行对比
- from concurrent.futures import ThreadPoolExecutor
- import requests
- import time
- urls = [
- "https://httpbin.org/delay/1",
- "https://httpbin.org/delay/2",
- "https://httpbin.org/delay/1",
- "https://httpbin.org/delay/3",
- ]
- def download(url):
- start = time.time()
- resp = requests.get(url, timeout=10)
- elapsed = time.time() - start
- return {"url": url, "status": resp.status_code, "time": elapsed}
- # 串行
- start = time.time()
- for url in urls:
- print(download(url))
- print(f"串行耗时：{time.time() - start:.2f}s")
- # 并行（线程池）
- start = time.time()
- with ThreadPoolExecutor(max_workers=4) as executor:
- results = executor.map(download, urls)
- for r in results:
- print(r)
- print(f"并行耗时：{time.time() - start:.2f}s")
4 个请求分别等待 1、2、1、3 秒，串行累计需要约 7 秒，线程池并行只需等最长的那一个，约 3 秒。这个例子也再次印证了前面的结论：线程的收益来自"重叠等待"，而不是"同时计算"。
把上面几点串起来看，Python 多线程的实践路径其实很清晰：先用 GIL 判断任务是 I/O 还是 CPU 密集，I/O 类再决定是裸线程还是线程池；共享变量一律用 Lock 保护，线程间传数据一律走 Queue；需要限制并发数量就上 Semaphore 或 max_workers。踩坑最多的地方往往不是线程 API 本身，而是忘了 join、忘了加锁、以及把计算任务硬塞进线程里。 |
发表于 
发表于