wwt｜逻辑与碎片

首页 / 知识库 / 0基础入门-阅读资料 / 0基础-python入门到精通

第 19 章 并发编程

19.1 GIL 的本质与影响

GIL（Global Interpreter Lock，全局解释器锁） 是 CPython 的一个互斥锁，确保同一时刻只有一个线程执行 Python 字节码。

# 即使创建多个线程，CPU 密集型任务也无法真正并行
import threading
import time

def cpu_bound(n):
    total = 0
    for i in range(n):
        total += i * i
    return total

start = time.perf_counter()

# 单线程
cpu_bound(10_000_000)
cpu_bound(10_000_000)
print(f"单线程: {time.perf_counter() - start:.2f}s")

start = time.perf_counter()

# 两个线程（不会更快，因为 GIL）
t1 = threading.Thread(target=cpu_bound, args=(10_000_000,))
t2 = threading.Thread(target=cpu_bound, args=(10_000_000,))
t1.start(); t2.start()
t1.join(); t2.join()
print(f"双线程: {time.perf_counter() - start:.2f}s")
# 双线程甚至可能比单线程更慢（线程切换开销）

GIL 的影响：

任务类型多线程效果推荐方案
CPU 密集型无加速（GIL 限制）multiprocessing
I/O 密集型有效加速threading 或 asyncio
混合型部分加速视具体情况选择

GIL 在 I/O 等待时会释放，因此多线程对 I/O 密集型任务仍然有效。

19.2 多线程 threading

19.2.1 线程创建与管理

import threading
import time

# 方式 1：传入函数
def worker(name, seconds):
    print(f"[{name}] 开始工作")
    time.sleep(seconds)
    print(f"[{name}] 完成")

t1 = threading.Thread(target=worker, args=("线程A", 2))
t2 = threading.Thread(target=worker, args=("线程B", 1))

t1.start()
t2.start()

t1.join()  # 等待 t1 完成
t2.join()  # 等待 t2 完成
print("所有线程完成")

# 方式 2：继承 Thread
class DownloadThread(threading.Thread):
    def __init__(self, url):
        super().__init__()
        self.url = url
        self.result = None
    
    def run(self):
        print(f"下载 {self.url}")
        time.sleep(1)  # 模拟下载
        self.result = f"来自 {self.url} 的数据"

t = DownloadThread("https://example.com")
t.start()
t.join()
print(t.result)

守护线程

# 守护线程在主线程结束时自动终止
def background_task():
    while True:
        print("后台运行中...")
        time.sleep(1)

t = threading.Thread(target=background_task, daemon=True)
t.start()

time.sleep(3)
print("主线程退出")
# 守护线程自动结束，不会阻止程序退出

19.2.2 锁和同步原语

import threading

# 无锁 — 竞态条件
counter = 0

def increment(n):
    global counter
    for _ in range(n):
        counter += 1  # 不是原子操作！读 → 加 → 写

threads = [threading.Thread(target=increment, args=(100_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"期望: 1000000, 实际: {counter}")  # 可能小于 1000000

# 用 Lock 解决
counter = 0
lock = threading.Lock()

def safe_increment(n):
    global counter
    for _ in range(n):
        with lock:  # 获取锁
            counter += 1

threads = [threading.Thread(target=safe_increment, args=(100_000,)) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(f"结果: {counter}")  # 精确 1000000

其他同步原语

import threading

# RLock — 可重入锁（同一线程可以多次获取）
rlock = threading.RLock()
with rlock:
    with rlock:  # 不会死锁
        pass

# Semaphore — 信号量（限制并发数）
sem = threading.Semaphore(3)  # 最多 3 个线程同时进入

def limited_task(name):
    with sem:
        print(f"{name} 获得许可")
        time.sleep(1)

# Event — 线程间信号
event = threading.Event()

def waiter():
    print("等待信号...")
    event.wait()       # 阻塞直到 event 被 set
    print("收到信号！")

def setter():
    time.sleep(2)
    print("发送信号")
    event.set()

# Condition — 条件变量
condition = threading.Condition()
items = []

def producer():
    with condition:
        items.append("item")
        condition.notify()  # 通知等待的消费者

def consumer():
    with condition:
        while not items:
            condition.wait()   # 释放锁并等待通知
        item = items.pop()
        print(f"消费: {item}")

19.2.3 线程安全

import threading
from queue import Queue

# Queue 是线程安全的 — 推荐用于线程间通信
def producer(q: Queue):
    for i in range(5):
        q.put(f"任务-{i}")
        time.sleep(0.1)
    q.put(None)  # 结束信号

def consumer(q: Queue):
    while True:
        item = q.get()  # 阻塞等待
        if item is None:
            break
        print(f"处理: {item}")
        q.task_done()

q = Queue(maxsize=10)
threading.Thread(target=producer, args=(q,)).start()
threading.Thread(target=consumer, args=(q,)).start()
# threading.local() — 线程本地存储
local_data = threading.local()

def worker(name):
    local_data.name = name  # 每个线程有独立的副本
    time.sleep(0.1)
    print(f"线程 {threading.current_thread().name}: {local_data.name}")

t1 = threading.Thread(target=worker, args=("Alice",))
t2 = threading.Thread(target=worker, args=("Bob",))
t1.start(); t2.start()
t1.join(); t2.join()
# 不会互相干扰

19.3 多进程 multiprocessing

多进程绕过 GIL，实现真正的并行计算：

19.3.1 进程创建与进程池

import multiprocessing
import time
import os

def cpu_bound(n):
    print(f"进程 {os.getpid()} 开始计算")
    total = sum(i * i for i in range(n))
    return total

if __name__ == "__main__":
    start = time.perf_counter()
    
    # 方式 1：手动创建进程
    p1 = multiprocessing.Process(target=cpu_bound, args=(10_000_000,))
    p2 = multiprocessing.Process(target=cpu_bound, args=(10_000_000,))
    p1.start(); p2.start()
    p1.join(); p2.join()
    
    print(f"双进程: {time.perf_counter() - start:.2f}s")
    
    # 方式 2：进程池
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(cpu_bound, [2_500_000] * 4)
    print(f"总计: {sum(results)}")
    
    # 方式 3：Pool 的其他方法
    with multiprocessing.Pool(4) as pool:
        # apply_async — 异步提交单个任务
        future = pool.apply_async(cpu_bound, (10_000_000,))
        result = future.get(timeout=30)
        
        # imap — 惰性版 map
        for result in pool.imap(cpu_bound, [1_000_000] * 10):
            print(result)

19.3.2 进程间通信

import multiprocessing

# Queue — 进程安全队列
def producer(q):
    for i in range(5):
        q.put(f"数据-{i}")
    q.put(None)

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"收到: {item}")

if __name__ == "__main__":
    q = multiprocessing.Queue()
    p = multiprocessing.Process(target=producer, args=(q,))
    c = multiprocessing.Process(target=consumer, args=(q,))
    p.start(); c.start()
    p.join(); c.join()

# Pipe — 双向管道（两个进程间通信）
def sender(conn):
    conn.send("Hello from sender")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = multiprocessing.Pipe()
    p = multiprocessing.Process(target=sender, args=(child_conn,))
    p.start()
    print(parent_conn.recv())  # Hello from sender
    p.join()

# 共享内存
from multiprocessing import Value, Array

def increment(counter, lock):
    for _ in range(100_000):
        with lock:
            counter.value += 1

if __name__ == "__main__":
    counter = Value("i", 0)         # 共享整数
    lock = multiprocessing.Lock()
    
    processes = [
        multiprocessing.Process(target=increment, args=(counter, lock))
        for _ in range(4)
    ]
    for p in processes: p.start()
    for p in processes: p.join()
    print(f"结果: {counter.value}")  # 400000

19.4 concurrent.futures 统一接口

concurrent.futures 为线程和进程提供统一的高层接口：

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def download(url):
    time.sleep(1)  # 模拟网络请求
    return f"{url} 的内容"

def compute(n):
    return sum(i * i for i in range(n))

# 线程池（I/O 密集型）
with ThreadPoolExecutor(max_workers=5) as executor:
    urls = [f"https://example.com/{i}" for i in range(10)]
    
    # map — 按顺序返回结果
    results = executor.map(download, urls)
    for result in results:
        print(result)

# 进程池（CPU 密集型）
with ProcessPoolExecutor(max_workers=4) as executor:
    # submit — 返回 Future 对象
    futures = [executor.submit(compute, 1_000_000) for _ in range(8)]
    
    # as_completed — 谁先完成谁先返回
    from concurrent.futures import as_completed
    for future in as_completed(futures):
        print(f"结果: {future.result()}")

Future 对象

from concurrent.futures import ThreadPoolExecutor, Future

with ThreadPoolExecutor(3) as executor:
    future: Future = executor.submit(download, "https://example.com")
    
    print(future.done())       # False（可能还没完成）
    print(future.running())    # True
    
    result = future.result(timeout=5)  # 阻塞等待结果
    print(future.done())       # True
    
    # 回调
    def on_complete(f: Future):
        print(f"完成: {f.result()}")
    
    future2 = executor.submit(download, "https://example.com/2")
    future2.add_done_callback(on_complete)

19.5 子解释器与 free-threaded Python（Python 3.13+）

子解释器（Subinterpreters）

每个子解释器有独立的 GIL，可以实现真正的线程并行：

# Python 3.12+ 的子解释器 API
import _interpreters as interpreters

interp_id = interpreters.create()
interpreters.run_string(interp_id, """
import math
print(f"子解释器中计算 pi = {math.pi}")
""")
interpreters.destroy(interp_id)

free-threaded Python（实验性，Python 3.13+）

Python 3.13 引入了实验性的无 GIL 构建（--disable-gil）：

# 安装无 GIL 版本
# 需要从源码编译或使用特殊安装方式
python3.13t  # t 后缀表示 free-threaded 版本
# 在 free-threaded Python 中，多线程 CPU 密集型任务可以真正并行
import threading
import sys

print(sys._is_gil_enabled())  # False（如果使用 free-threaded 构建）

# 现在多线程 CPU 密集型任务能获得真正的加速

注意：free-threaded Python 仍处于实验阶段，许多第三方 C 扩展尚不支持。

本章小结：Python 并发编程的选择取决于任务类型——I/O 密集用多线程或 asyncio，CPU 密集用多进程。concurrent.futures 提供了统一的高层接口，是大多数场景的首选。理解 GIL 是正确选择并发方案的关键。Python 3.13 的 free-threaded 模式预示着 Python 并发的未来。