wordflow
返回博客

41. 多进程与多线程

41. 多进程与多线程

第四十一篇：多进程与多线程——并行编程入门

前言

Python 提供了多种实现并发和并行执行的方式。多进程绕过 GIL 限制，适合 CPU 密集型任务；多线程共享内存空间，适合 I/O 密集型任务。本文将从基本概念出发，系统讲解 multiprocessing 和 threading 模块的使用。

为什么需要并发？

import time

# 串行执行
def download_image(url):
    time.sleep(1)  # 模拟网络延迟
    return f"下载 {url}"

start = time.perf_counter()
for url in [f"http://example.com/image{i}.jpg" for i in range(5)]:
    print(download_image(url))
print(f"串行耗时: {time.perf_counter() - start:.2f}s")
# 串行耗时: ~5.0s（5 × 1s）

# 如果用多线程/多进程，这 5 个下载可以并行进行！
# 理论上只需要 ~1s 即可完成所有下载

multiprocessing —— 多进程编程

基本用法

import multiprocessing
import time

def worker(name):
    """工作进程"""
    print(f"[PID {multiprocessing.current_process().pid}] {name} 开始工作...")
    time.sleep(1)  # 模拟耗时操作
    print(f"[PID {multiprocessing.current_process().pid}] {name} 完成!")
    return f"{name} 的结果"

if __name__ == "__main__":
    # 方法一：Process — 手动创建和管理进程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=worker, args=(f"进程-{i+1}",))
        p.start()
        processes.append(p)
    
    # 等待所有进程完成
    for p in processes:
        p.join()
    print("所有进程完成!")

    # 方法二：Pool — 进程池（推荐用于批量任务）
    pool = multiprocessing.Pool(processes=4)
    results = pool.map(worker, [f"任务-{i}" for i in range(10)])
    pool.close()
    pool.join()
    print(results)

Pool 常用方法对比

import multiprocessing
import time

def square(x):
    time.sleep(0.1)
    return x * x

if __name__ == "__main__":
    pool = multiprocessing.Pool(processes=4)
    
    # map — 阻塞式，等所有结果返回
    result1 = pool.map(square, range(10))
    print(f"map: {result1}")
    
    # map_async — 异步非阻塞
    async_result = pool.map_async(square, range(10))
    if async_result.ready():  # 检查是否完成
        print(async_result.get())
    else:
        print("正在计算中...")
        result = async_result.get(timeout=5)  # 等待最多 5 秒
        print(result)
    
    # apply_async — 异步执行单个任务
    single_result = pool.apply_async(square, (7,))
    print(single_result.get())  # 49
    
    # imap — 惰性迭代（流式处理大量数据）
    for r in pool.imap(square, range(20)):
        pass  # 逐个获取结果
    
    # imap_unordered — 不保证顺序，哪个先完成先返回
    for r in pool.imap_unordered(square, range(20)):
        pass  # 适合不关心顺序的场景
    
    pool.close()
    pool.join()

进程间通信

import multiprocessing
import time

# 方式一：Queue — 线程安全的队列
def producer(q):
    for i in range(5):
        q.put(f"消息 {i}")
        time.sleep(0.2)
    q.put(None)  # 结束信号

def consumer(q):
    while True:
        msg = q.get()
        if msg is None:
            break
        print(f"消费: {msg}")

if __name__ == "__main__":
    queue = multiprocessing.Queue()
    p1 = multiprocessing.Process(target=producer, args=(queue,))
    p2 = multiprocessing.Process(target=consumer, args=(queue,))
    
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print("队列通信完成!")

# 方式二：Pipe — 双向管道（两个端点）
def send(pipe):
    pipe.send({"data": "hello from parent"})
    pipe.send([1, 2, 3])
    pipe.close()

def receive(pipe):
    while True:
        try:
            msg = pipe.recv()
            print(f"收到: {msg}")
        except EOFError:
            break

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=send, args=(parent_conn,))
    p2 = multiprocessing.Process(target=receive, args=(child_conn,))
    
    p1.start()
    p2.start()
    p1.join()
    p2.join()

# 方式三：Manager — 共享数据结构
def fill_dict(shared_dict, key_range):
    for i in key_range:
        shared_dict[i] = i ** 2

def fill_list(shared_list, values):
    shared_list.extend(values)

if __name__ == "__main__":
    manager = multiprocessing.Manager()
    shared_dict = manager.dict()
    shared_list = manager.list()
    
    p1 = multiprocessing.Process(target=fill_dict, args=(shared_dict, range(5)))
    p2 = multiprocessing.Process(target=fill_dict, args=(shared_dict, range(5, 10)))
    p3 = multiprocessing.Process(target=fill_list, args=(shared_list, [10, 20, 30]))
    p4 = multiprocessing.Process(target=fill_list, args=(shared_list, [40, 50]))
    
    for p in [p1, p2, p3, p4]:
        p.start()
    for p in [p1, p2, p3, p4]:
        p.join()
    
    print(f"Dict: {dict(shared_dict)}")
    print(f"List: {list(shared_list)}")

threading —— 多线程编程

基本用法

import threading
import time

# 方式一：Thread + 函数
def task(name, duration):
    print(f"[线程-{name}] 开始")
    time.sleep(duration)
    print(f"[线程-{name}] 完成，耗时 {duration}s")

threads = []
for i in range(3):
    t = threading.Thread(target=task, args=(f"T{i+1}", i+1))
    t.start()
    threads.append(t)

for t in threads:
    t.join()
print("所有线程完成!")

# 方式二：继承 Thread 类
class MyThread(threading.Thread):
    def __init__(self, name, work_time):
        super().__init__()
        self.name = name
        self.work_time = work_time
    
    def run(self):
        print(f"[{self.name}] 开始工作 ({self.work_time}s)")
        time.sleep(self.work_time)
        print(f"[{self.name}] 完成!")

t1 = MyThread("Worker-A", 2)
t2 = MyThread("Worker-B", 1)
t1.start()
t2.start()
t1.join()
t2.join()

线程安全与锁

import threading
import time

class BankAccount:
    """银行账户 — 演示线程安全问题"""
    
    def __init__(self, balance=1000):
        self.balance = balance
        self.lock = threading.Lock()  # 创建锁
    
    def deposit(self, amount):
        with self.lock:  # 自动获取/释放锁
            current = self.balance
            time.sleep(0.001)  # 模拟银行处理延迟
            self.balance = current + amount
            print(f"存款 {amount}, 余额: {self.balance}")
    
    def withdraw(self, amount):
        with self.lock:
            if self.balance >= amount:
                current = self.balance
                time.sleep(0.001)
                self.balance = current - amount
                print(f"取款 {amount}, 余额: {self.balance}")
                return True
            else:
                print("余额不足!")
                return False
    
    def get_balance(self):
        with self.lock:
            return self.balance

# 测试线程安全
account = BankAccount(1000)
threads = []

# 100 个线程同时存钱
for _ in range(100):
    t = threading.Thread(target=account.deposit, args=(10,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

print(f"最终余额: {account.get_balance()}")  # 应该是 2000
# 如果没有锁，这个值可能不正确!

Lock、RLock 与 Condition

import threading
import time

# RLock — 可重入锁（同一线程可多次获取）
rlock = threading.RLock()

def safe_increment(counter):
    with rlock:
        counter[0] += 1
        with rlock:  # 嵌套持有锁
            counter[0] += 1

counter = [0]
threads = [threading.Thread(target=safe_increment, args=(counter,)) for _ in range(100)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"计数器: {counter[0]}")  # 200

# Event — 线程间事件通知
event = threading.Event()

def waiter(name):
    print(f"{name} 等待事件...")
    event.wait()  # 阻塞直到事件被设置
    print(f"{name} 继续!")

def notifier():
    time.sleep(2)
    print("设置事件!")
    event.set()

for i in range(3):
    threading.Thread(target=waiter, args=(f"等待者-{i+1}",)).start()

threading.Thread(target=notifier).start()

# Condition — 条件变量（等待特定条件满足）
condition = threading.Condition()

items = []

def producer():
    for i in range(5):
        with condition:
            items.append(f"物品{i}")
            condition.notify_all()  # 通知所有消费者
            print(f"生产: 物品{i}, 库存: {len(items)}")
            time.sleep(0.5)

def consumer():
    for _ in range(5):
        with condition:
            while not items:
                condition.wait(timeout=3)  # 等待物品可用
            item = items.pop(0)
            print(f"消费: {item}")

threading.Thread(target=producer).start()
for _ in range(3):
    threading.Thread(target=consumer).start()

Semaphore — 控制并发数量

import threading
import time

class PhotoGallery:
    """照片墙 — 最多同时显示 3 张"""
    
    def __init__(self, max_display=3):
        self.semaphore = threading.Semaphore(max_display)
        self.displayed = []
        self.lock = threading.Lock()
    
    def display_photo(self, photo_id):
        self.semaphore.acquire()
        try:
            time.sleep(0.5)  # 模拟加载时间
            with self.lock:
                self.displayed.append(photo_id)
            print(f"显示照片 {photo_id}, 当前显示: {len(self.displayed)} 张")
        finally:
            self.semaphore.release()
    
    def remove_photo(self, photo_id):
        with self.lock:
            if photo_id in self.displayed:
                self.displayed.remove(photo_id)
                print(f"移除照片 {photo_id}, 剩余: {len(self.displayed)} 张")

gallery = PhotoGallery(max_display=3)
photos = [f"photo_{i}" for i in range(8)]

threads = [threading.Thread(target=gallery.display_photo, args=(p,)) for p in photos]
for t in threads:
    t.start()

for t in threads:
    t.join()

print(f"\n最终显示: {gallery.displayed}")

本地线程 ThreadLocal

import threading
import time

# 每个线程有自己独立的数据副本
local_data = threading.local()

def worker(task_id):
    local_data.user_id = f"user_{task_id}"
    local_data.request_id = f"req_{time.time()}"
    time.sleep(0.1)
    print(f"线程 {task_id}: user_id={local_data.user_id}, request_id={local_data.request_id}")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

实践场景：并发爬虫

import threading
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from collections import defaultdict

def simulate_fetch(url):
    """模拟网页抓取（含随机延迟）"""
    delay = 0.1 + hash(url) % 20 / 100  # 0.1~0.3s
    time.sleep(delay)
    return {"url": url, "status": 200, "size": 1024 * (hash(url) % 100 + 1)}

def scrape_urls(urls, max_workers=5, use_threads=True):
    """并发抓取多个 URL"""
    results = {}
    errors = []
    lock = threading.Lock()
    
    executor_class = ThreadPoolExecutor if use_threads else ProcessPoolExecutor
    
    with executor_class(max_workers=max_workers) as executor:
        future_to_url = {executor.submit(simulate_fetch, url): url for url in urls}
        
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                result = future.result()
                with lock:
                    results[url] = result
            except Exception as e:
                with lock:
                    errors.append((url, str(e)))
            
            progress = len(results) + len(errors)
            if progress % 5 == 0:
                print(f"进度: {progress}/{len(urls)}")
    
    total_size = sum(r["size"] for r in results.values())
    return results, errors, total_size

# 测试
urls = [f"http://example.com/page{i}" for i in range(20)]
results, errors, total_size = scrape_urls(urls[:10], max_workers=5, use_threads=True)
print(f"成功: {len(results)}, 失败: {len(errors)}, 总大小: {total_size/1024:.1f}KB")

实战：任务队列处理器

import threading
import queue
import time
from datetime import datetime

class TaskProcessor:
    """线程安全的任务队列处理器"""
    
    def __init__(self, num_workers=4):
        self.task_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.num_workers = num_workers
        self.workers = []
        self.running = False
    
    def _worker(self, worker_id):
        """工作线程循环"""
        while self.running:
            try:
                task = self.task_queue.get(timeout=1)
            except queue.Empty:
                continue
            
            try:
                result = task["func"](*task.get("args", []), **task.get("kwargs", {}))
                self.result_queue.put({
                    "status": "success",
                    "task_id": task.get("id"),
                    "result": result,
                    "worker": worker_id,
                    "timestamp": datetime.now().isoformat(),
                })
            except Exception as e:
                self.result_queue.put({
                    "status": "error",
                    "task_id": task.get("id"),
                    "error": str(e),
                    "worker": worker_id,
                    "timestamp": datetime.now().isoformat(),
                })
            finally:
                self.task_queue.task_done()
    
    def submit_task(self, func, *args, task_id=None, **kwargs):
        """提交任务到队列"""
        task = {
            "id": task_id or str(hash(str(args) + str(kwargs))),
            "func": func,
            "args": args,
            "kwargs": kwargs,
        }
        self.task_queue.put(task)
        return task["id"]
    
    def start(self):
        """启动工作线程"""
        self.running = True
        for i in range(self.num_workers):
            t = threading.Thread(target=self._worker, args=(i,), daemon=True)
            t.start()
            self.workers.append(t)
        print(f"已启动 {self.num_workers} 个工作线程")
    
    def stop(self):
        """停止工作线程"""
        self.running = False
    
    def wait_completion(self):
        """等待队列中所有任务完成"""
        self.task_queue.join()
    
    def get_results(self):
        """获取已完成的结果"""
        results = []
        while not self.result_queue.empty():
            try:
                results.append(self.result_queue.get_nowait())
            except queue.Empty:
                break
        return results


# 使用示例
def add(a, b):
    time.sleep(0.05)
    return a + b

def multiply(a, b):
    time.sleep(0.05)
    return a * b

processor = TaskProcessor(num_workers=4)
processor.start()

# 提交混合任务
task_ids = []
for i in range(1, 11):
    if i <= 5:
        tid = processor.submit_task(add, i, i*2, task_id=f"add_{i}")
    else:
        tid = processor.submit_task(multiply, i, i, task_id=f"mul_{i}")
    task_ids.append(tid)

processor.wait_completion()
results = processor.get_results()
processor.stop()

print(f"\n共 {len(results)} 个结果:")
for r in results:
    status_icon = "✓" if r["status"] == "success" else "✗"
    if r["status"] == "success":
        print(f"  [{status_icon}] {r['task_id']}: {r['result']}")
    else:
        print(f"  [{status_icon}] {r['task_id']}: 错误 - {r['error']}")

GIL 到底是什么？

import threading
import multiprocessing
import time

# Python 的 GIL（Global Interpreter Lock）限制：
# 同一时刻只有一个线程在执行 Python 字节码
# 
# → 多线程 NOT 能加速 CPU 密集型计算！
# → 多线程可以加速 I/O 密集型操作（网络、文件、数据库）
# → 多进程才能真正利用多核 CPU

def cpu_bound_work(n):
    """CPU 密集型任务"""
    total = 0
    for i in range(n):
        total += i * i
    return total

def io_bound_work(n):
    """I/O 密集型任务"""
    time.sleep(0.1)  # 模拟等待
    return "done"

# 对比：GIL 对两种任务的影響
print("=== CPU 密集型 (计算平方和) ===")
n = 10_000_000

# 单线程
start = time.perf_counter()
cpu_bound_work(n)
single_thread = time.perf_counter() - start
print(f"单线程: {single_thread:.2f}s")

# 多线程（受 GIL 限制，不会更快）
start = time.perf_counter()
threads = [threading.Thread(target=cpu_bound_work, args=(n//4,)) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
multi_thread = time.perf_counter() - start
print(f"多线程: {multi_thread:.2f}s")

# 多进程（突破 GIL，充分利用多核）
start = time.perf_counter()
processes = [multiprocessing.Process(target=cpu_bound_work, args=(n//4,)) for _ in range(4)]
for p in processes: p.start()
for p in processes: p.join()
multi_process = time.perf_counter() - start
print(f"多进程: {multi_process:.2f}s")

print("\n=== I/O 密集型 (sleep) ===")
# 多线程 vs 单线程处理 20 次 0.1s 的 sleep
start = time.perf_counter()
for _ in range(20): io_bound_work(1)
single_io = time.perf_counter() - start
print(f"单线程 I/O: {single_io:.2f}s (~2s)")

start = time.perf_counter()
threads = [threading.Thread(target=io_bound_work, args=(1,)) for _ in range(20)]
for t in threads: t.start()
for t in threads: t.join()
multi_io = time.perf_counter() - start
print(f"多线程 I/O: {multi_io:.2f}s (~0.1s!)")

最佳实践总结

✅ 选择策略

场景 推荐方案 原因
CPU 密集型（计算、数据处理） 多进程 突破 GIL，利用多核
I/O 密集型（网络请求、文件读写） 多线程 阻塞时释放 GIL，开销小
GUI 应用 多线程 主线程处理 UI，子线程做后台工作
数据并行处理 多进程 + Pool 天然隔离，无需锁
实时响应服务 线程 + Queue 生产者-消费者模型

❌ 避免陷阱

import threading
import multiprocessing

# ❌ 在线程间直接共享可变对象（需要加锁）
shared_list = []  # 不安全！
shared_list.append(1)  # 可能被另一个线程打断

# ✅ 使用锁保护共享状态
lock = threading.Lock()
with lock:
    shared_list.append(1)

# ❌ 忘记 join() — 主线程退出后子线程可能被杀死
t = threading.Thread(target=some_function)
t.start()
# 没有 t.join() ← 主线程退出可能导致程序提前结束

# ✅ 始终等待子线程/子进程
t.join()

# ❌ 在 pickle 支持的范围内传递闭包（多进程中不可序列化）
closure_data = {"x": 1}
p = multiprocessing.Process(target=lambda: closure_data["x"])  # ✗ 可能失败

# ✅ 多进程中通过参数传递或 Queue/Pipe 通信
p = multiprocessing.Process(target=function_with_args, args=(closure_data["x"],))

# ❌ 忽略 KeyboardInterrupt 在多进程中的传播
pool = multiprocessing.Pool()
try:
    results = pool.map(func, items)
except KeyboardInterrupt:
    pool.terminate()
    pool.join()

总结

本文系统讲解了 Python 并发编程的核心知识：

  1. 多进程 multiprocessing — Process、Pool（map/imap）、Queue、Pipe、Manager
  2. 多线程 threading — Thread、锁(Lock/RLock)、Event、Condition、Semaphore
  3. GIL 详解 — 为什么多线程不能加速 CPU 密集任务
  4. 通信机制 — Queue、Pipe、Manager 的适用场景
  5. 同步原语 — Lock/RLock/Event/Condition/Semaphore 的设计模式
  6. 实战案例 — 并发爬虫、任务队列处理器
  7. 最佳实践 — 按任务类型选择方案、避免常见陷阱

下一篇文章将深入讲解 ThreadLocal 与分布式进程。

本文是 Python 教程系列的第四十一篇。