一、从GIL锁说起
第一次用Python写爬虫时，我发现明明开了5个线程，速度却比单线程还慢。盯着top命令里100%的CPU使用率发呆时，才意识到踩中了GIL（全局解释器锁）的坑。这个让无数Python开发者又爱又恨的机制，正是理解多线程/多进程的关键起点。
二、线程与进程的本质差异
1. 内存模型对比
import threading
import multiprocessing
value = 0
def thread_task():
global value
value += 1
def process_task(val):
val.value += 1
# 多线程示例
threads = []
for _ in range(10):
t = threading.Thread(target=thread_task)
threads.append(t)
t.start()
[t.join() for t in threads]
print(fThread final value: {value}) # 结果不确定
# 多进程示例
val = multiprocessing.Value('i', 0)
processes = []
for _ in range(10):
p = multiprocessing.Process(target=process_task, args=(val,))
processes.append(p)
p.start()
[p.join() for p in processes]
print(fProcess final value: {val.value}) # 恒为0
关键差异：
- 线程共享相同内存空间（导致
value竞争） - 进程拥有独立内存（需要
Value共享内存）
2. 创建开销实测
在我的ThinkPad X1上测试（单位：毫秒）：
import time
def test_overhead(creator, count):
start = time.time()
tasks = [creator(target=lambda: None) for _ in range(count)]
[t.start() for t in tasks]
[t.join() for t in tasks]
return (time.time() - start)*1000
print(f线程创建开销：{test_overhead(threading.Thread, 1000):.2f}ms)
print(f进程创建开销：{test_overhead(multiprocessing.Process, 1000):.2f}ms)
典型输出：
线程创建开销：35.21ms
进程创建开销：1204.57ms
三、真实场景选择指南
适合多线程的场景
- I/O密集型任务：爬虫是最典型案例
def download(url):
# 模拟网络请求
time.sleep(0.5)
print(fDownloaded {url})
urls = [...] # 100个URL
# 线程池优化
with ThreadPoolExecutor(max_workers=20) as executor:
executor.map(download, urls)
- GUI应用：保持界面响应
# PyQt示例
def long_running_task():
time.sleep(5) # 模拟耗时操作
print(Done)
app = QApplication([])
button = QPushButton(Run Task)
button.clicked.connect(lambda: threading.Thread(target=long_running_task).start())
button.show()
app.exec_()
适合多进程的场景
- CPU密集型计算：我在量化交易中处理Tick数据时深有体会
def process_chunk(data):
return sum(x**2 for x in data)
data = [...] # 1亿条数据
chunks = np.array_split(data, 8) # 8核CPU
with Pool(8) as p:
results = p.map(process_chunk, chunks)
total = sum(results)
- 避免GIL限制：用C扩展反而更麻烦时
# 加密解密场景
def encrypt_file(path):
# 使用CPU密集型加密算法
...
files = [...] # 1000个文件
with Pool() as p:
p.map(encrypt_file, files)
四、进阶技巧与坑点记录
- 线程池的隐藏陷阱：
# 错误示范（可能导致死锁）
def process(data):
with ThreadPoolExecutor(4) as executor:
return list(executor.map(sub_task, data))
# 正确做法（控制嵌套层级）
executor = ThreadPoolExecutor(4)
results = list(executor.map(process, big_data))
- 进程间通信的实践：
# 使用Queue的实际案例
def producer(q):
while True:
data = get_data()
q.put(data)
def consumer(q):
while True:
data = q.get()
process(data)
queue = multiprocessing.Queue(maxsize=10)
p1 = Process(target=producer, args=(queue,))
p2 = Process(target=consumer, args=(queue,))
五、我的工具选择心得
-
现代替代方案：
- 协程（asyncio）：适合高并发I/O（比线程更轻量）
- 分布式框架（Celery）：跨机器扩展
-
调试建议：
# 打印线程/进程ID
print(fThread {threading.get_ident()} | Process {os.getpid()})
# 用tracemalloc调试内存问题
import tracemalloc
tracemalloc.start()
# ...执行代码...
snapshot = tracemalloc.take_snapshot()
最后提醒：在Jupyter Notebook中使用多进程时，记得把代码封装成模块，否则会出现奇怪的pickle错误——这个坑我花了整个周末才爬出来。