Python 并发编程
1. 线程 vs 进程 vs 协程对比
| 特性 | 线程 (threading) | 进程 (multiprocessing) | 协程 (asyncio) |
|---|---|---|---|
| 创建开销 | 低 | 高 | 极低 |
| 内存共享 | 共享进程内存 | 独立内存空间 | 共享线程内存 |
| GIL 限制 | 受限（CPU 密集型无法并行） | 不受限 | 受限（单线程） |
| 适用场景 | I/O 密集型 | CPU 密集型 | 高并发 I/O |
| 上下文切换 | 内核态切换 | 内核态切换 | 用户态切换 |
| 通信方式 | 共享变量 + 锁 | Queue / Pipe | asyncio.Queue |
python
# 线程：I/O 密集型
import threading
def io_task():
import time
time.sleep(1)
print(f'{threading.current_thread().name} done')
threads = [threading.Thread(target=io_task) for _ in range(5)]
for t in threads:
t.start()
for t in threads:
t.join()
# 总耗时约 1 秒（并发执行）
# 进程：CPU 密集型
from multiprocessing import Process
def cpu_task():
total = sum(range(10_000_000))
print(f'Process done: {total}')
processes = [Process(target=cpu_task) for _ in range(4)]
for p in processes:
p.start()
for p in processes:
p.join()
# 真正的多核并行2. threading 同步原语
python
import threading
import time
# Lock — 互斥锁
counter = 0
lock = threading.Lock()
def increment():
global counter
for _ in range(100000):
with lock: # 推荐使用 with 语句
counter += 1
# RLock — 可重入锁
rlock = threading.RLock()
def nested():
with rlock:
with rlock: # 同一线程可以多次获取
print('nested')
# Condition — 条件变量
buffer = []
condition = threading.Condition()
def producer():
with condition:
buffer.append(1)
condition.notify() # 通知消费者
def consumer():
with condition:
while not buffer:
condition.wait() # 等待通知
item = buffer.pop()
print(f'Consumed: {item}')
# Event — 事件
event = threading.Event()
def waiter():
event.wait()
print('Event is set!')
event.set() # 唤醒所有 wait()
# Semaphore — 信号量（限制并发数）
sem = threading.Semaphore(3) # 最多 3 个线程同时访问
def limited_task():
with sem:
print(f'{threading.current_thread().name} acquired')
time.sleep(1)
# Barrier — 屏障
barrier = threading.Barrier(3)
def sync_task():
barrier.wait() # 3 个线程到达后一起继续
print('All synchronized!')3. multiprocessing IPC
python
from multiprocessing import Process, Queue, Pipe, Manager
import time
# Queue — 进程间消息队列
def sender(q):
q.put({'type': 'data', 'value': 42})
def receiver(q):
msg = q.get()
print(f'Received: {msg}')
q = Queue()
Process(target=sender, args=(q,)).start()
Process(target=receiver, args=(q,)).start()
# Pipe — 双向管道（点对点）
def child_process(conn):
conn.send('Hello from child')
response = conn.recv()
print(f'Child received: {response}')
conn.close()
parent_conn, child_conn = Pipe()
Process(target=child_process, args=(child_conn,)).start()
print(f'Parent received: {parent_conn.recv()}')
parent_conn.send('Hello from parent')
# Manager — 共享对象
def worker(shared_dict, key, value):
shared_dict[key] = value
with Manager() as manager:
shared_dict = manager.dict()
processes = [
Process(target=worker, args=(shared_dict, f'key{i}', i))
for i in range(5)
]
for p in processes:
p.start()
for p in processes:
p.join()
print(dict(shared_dict))面试要点：multiprocessing 的 fork() 在 Linux 上会复制整个地址空间（CoW），Windows 上使用 spawn()。Python 3.12+ 默认使用 forkserver。
4. asyncio 事件循环与 epoll
python
import asyncio
import time
# asyncio 基于事件循环（底层使用 epoll/kqueue）
async def fetch_data(name, delay):
print(f'{name} 开始请求')
await asyncio.sleep(delay) # 非阻塞等待
print(f'{name} 完成')
return f'{name}_result'
async def main():
# 并发执行三个协程
start = time.time()
results = await asyncio.gather(
fetch_data('A', 2),
fetch_data('B', 1),
fetch_data('C', 3),
)
print(f'耗时: {time.time() - start:.1f}s') # 约 3s（取最长）
print(f'结果: {results}')
asyncio.run(main())事件循环底层（简化）：
python
# 伪代码：事件循环核心
class EventLoop:
def __init__(self):
self._ready = collections.deque() # 就绪队列
self._selector = selectors.DefaultSelector() # epoll/kqueue
def call_soon(self, callback, *args):
self._ready.append((callback, args))
def run_forever(self):
while self._ready or self._selector.get_map():
# 1. 执行所有就绪回调
while self._ready:
callback, args = self._ready.popleft()
callback(*args)
# 2. 用 epoll 等待 I/O 事件
events = self._selector.select(timeout=None)
for key, mask in events:
callback = key.data
callback()
def create_task(self, coroutine):
task = Task(coroutine, loop=self)
self._ready.append((task._step, ()))
return task5. Task 调度
python
import asyncio
async def task(name, duration):
print(f'{name} started')
await asyncio.sleep(duration)
print(f'{name} finished after {duration}s')
return duration
async def main():
# create_task: 立即调度，不阻塞
t1 = asyncio.create_task(task('T1', 3))
t2 = asyncio.create_task(task('T2', 1))
t3 = asyncio.create_task(task('T3', 2))
# 按完成顺序收集结果
results = []
for coro in asyncio.as_completed([t1, t2, t3]):
result = await coro
results.append(result)
print(f'Got result: {result}')
print(f'All results: {results}')
asyncio.run(main())
# 输出顺序：T2 先完成（1s），T3（2s），T1（3s）6. 异步上下文管理器与异步迭代器
python
import asyncio
# 异步上下文管理器
class AsyncDBConnection:
def __init__(self, name):
self.name = name
async def __aenter__(self):
print(f'Connecting to {self.name}...')
await asyncio.sleep(0.1) # 模拟连接
self.connected = True
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print(f'Disconnecting from {self.name}...')
await asyncio.sleep(0.1)
self.connected = False
return False # 不抑制异常
async def use_db():
async with AsyncDBConnection('mydb') as db:
print(f'Connected: {db.connected}')
# 异步迭代器
class AsyncCounter:
def __init__(self, stop):
self.current = 0
self.stop = stop
def __aiter__(self):
return self
async def __anext__(self):
if self.current >= self.stop:
raise StopAsyncIteration
await asyncio.sleep(0.1)
self.current += 1
return self.current
async def count():
async for num in AsyncCounter(5):
print(num) # 1, 2, 3, 4, 5
asyncio.run(count())7. 死锁的四个必要条件
python
import threading
# 死锁的四个条件（必须同时满足）：
# 1. 互斥条件：资源一次只能被一个线程持有
# 2. 持有并等待：线程持有资源的同时在等待其他资源
# 3. 不可剥夺：已获取的资源不能被强制释放
# 4. 循环等待：线程之间形成资源等待环
# 死锁示例
lock_a = threading.Lock()
lock_b = threading.Lock()
def thread1():
with lock_a:
print('Thread1: acquired A')
# time.sleep(0.1) # 增加死锁概率
with lock_b:
print('Thread1: acquired B')
def thread2():
with lock_b:
print('Thread2: acquired B')
# time.sleep(0.1)
with lock_a:
print('Thread2: acquired A')
# 可能死锁！
# threading.Thread(target=thread1).start()
# threading.Thread(target=thread2).start()死锁预防策略：
python
# 策略1：固定加锁顺序
def safe_thread1():
with lock_a: # 总是先获取 A
with lock_b:
print('safe1')
def safe_thread2():
with lock_a: # 总是先获取 A
with lock_b:
print('safe2')
# 策略2：使用超时
def timeout_thread():
acquired_a = lock_a.acquire(timeout=1)
if acquired_a:
try:
acquired_b = lock_b.acquire(timeout=1)
if acquired_b:
try:
print('both acquired')
finally:
lock_b.release()
finally:
lock_a.release()
# 策略3：使用 RLock 避免自死锁8. TaskGroup / ExceptionGroup (Python 3.11+)
python
import asyncio
async def fetch(url):
if url == 'bad':
raise ValueError(f'Failed to fetch {url}')
await asyncio.sleep(0.1)
return f'data from {url}'
async def main():
# ExceptionGroup — 捕获多个异常
try:
raise ExceptionGroup('errors', [
ValueError('bad value'),
TypeError('bad type'),
])
except* ValueError as eg:
print(f'Caught ValueError: {eg.exceptions}')
except* TypeError as eg:
print(f'Caught TypeError: {eg.exceptions}')
# TaskGroup — 结构化并发
async with asyncio.TaskGroup() as tg:
task1 = tg.create_task(fetch('url1'))
task2 = tg.create_task(fetch('url2'))
# task3 = tg.create_task(fetch('bad')) # 会引发 ExceptionGroup
# 所有任务完成后才能退出 with 块
print(task1.result())
print(task2.result())
asyncio.run(main())TaskGroup vs asyncio.gather：
| 特性 | TaskGroup | gather |
|---|---|---|
| 异常处理 | 结构化，抛出 ExceptionGroup | 第一个异常即中断 |
| 取消行为 | 取消所有任务 | 取消剩余任务 |
| 完成保证 | 所有任务完成/取消后才退出 | 任务可能仍在运行 |
| 引入版本 | 3.11 | 3.4 |
python
# TaskGroup 的取消行为
async def long_task():
await asyncio.sleep(10)
return 'done'
async def main():
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(long_task())
raise RuntimeError('Something went wrong')
# tg.create_task(long_task())
except* RuntimeError as eg:
print(f'Error: {eg}')
# long_task 会被自动取消面试要点：Python 3.11+ 推荐使用 TaskGroup 替代 gather，因为它提供了更好的异常处理和资源管理保证。