Python多线程与多进程：异同与实战场景解析-CFANZ编程社区

一、从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(f"Thread 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(f"Process 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(f"Downloaded {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(f"Thread {threading.get_ident()} | Process {os.getpid()}")

# 用tracemalloc调试内存问题
import tracemalloc
tracemalloc.start()
# ...执行代码...
snapshot = tracemalloc.take_snapshot()

最后提醒：在Jupyter Notebook中使用多进程时，记得把代码封装成模块，否则会出现奇怪的pickle错误——这个坑我花了整个周末才爬出来。