IT面试题|麻瓜面试

Python 并发编程的核心是理解 GIL 的限制，根据场景选择多线程、多进程或协程。并发不是银弹，选对工具才能提升性能。

Q1: Python中实现并发的方式有哪些？各有什么优缺点？「🟡 中级」

考察点：考察候选人对 Python 并发编程全貌的理解，是否清楚多线程、多进程、协程各自的适用场景和局限性，筛选只会用某一种方式而不会根据场景选型的人。

参考答案：

Python 中实现并发主要有三种方式：多线程、多进程和协程，三者各有侧重。

1. 多线程（threading）

原理：在同一进程内创建多个线程，由操作系统调度，线程间共享进程的内存空间。
优点：
- 轻量级，创建和切换开销小
- 共享内存，线程间通信方便（直接读写共享变量）
- 编程模型相对简单
缺点：
- 受 GIL（全局解释器锁） 限制，同一时刻只有一个线程执行 Python 字节码，无法利用多核 CPU 进行计算密集型任务
- 线程安全问题，需要锁等同步机制
- 线程数量有限（几千级别），太多会导致频繁上下文切换
适用场景：IO 密集型任务（网络请求、文件读写、数据库查询等）

2. 多进程（multiprocessing）

原理：创建多个独立的进程，每个进程有自己独立的内存空间和 GIL，由操作系统调度。
优点：
- 绕过 GIL，可以真正利用多核 CPU 并行计算
- 进程间相互独立，一个进程崩溃不会影响其他进程
- 适合 CPU 密集型任务
缺点：
- 进程创建和销毁开销大
- 进程间通信复杂（需要 IPC 机制，如队列、管道、共享内存等）
- 内存占用大（每个进程有独立的内存空间）
适用场景：CPU 密集型任务（数值计算、图像处理、数据处理等）

3. 协程（asyncio）

原理：在单线程内，由程序（事件循环）协作式调度，遇到 IO 操作时主动挂起，切换到其他协程执行。
优点：
- 极低的切换开销（用户态切换，几纳秒级别）
- 可以创建大量协程（几万甚至几十万级别）
- 没有线程安全问题（单线程内顺序执行）
缺点：
- 需要异步库支持（同步库会阻塞整个事件循环）
- 编程模型复杂，需要理解 async/await、事件循环等概念
- 无法利用多核 CPU（单线程）
- 调试困难（异常堆栈不直观）
适用场景：高并发 IO 密集型任务（Web 服务器、爬虫、实时聊天等）

对比表

对比维度	多线程	多进程	协程
适用场景	IO 密集型	CPU 密集型	高并发 IO 密集型
资源开销	中等	大	极小
切换开销	中等（内核态）	大（内核态）	极小（用户态）
通信方式	共享内存（需锁）	IPC（队列/管道/共享内存）	共享变量（单线程）
GIL 限制	受限制	不受限	受限制（单线程）
编程复杂度	中等	高	较高
并发数量	几千	几十到几百	几万到几十万
稳定性	一个线程崩溃影响进程	进程独立，稳定性高	一个协程崩溃影响整个事件循环

选型建议

python

# CPU 密集型 → 多进程
from multiprocessing import Pool

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

if __name__ == "__main__":
    with Pool(4) as p:
        results = p.map(cpu_bound_task, [10**6] * 10)

# IO 密集型 → 多线程 或 协程
import threading
import requests

def fetch(url):
    return requests.get(url)

threads = [threading.Thread(target=fetch, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()

# 高并发 IO 密集型 → 协程（asyncio + aiohttp）
import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.text()

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        await asyncio.gather(*tasks)

asyncio.run(main())

追问延伸：

GIL 具体是怎么工作的？在什么情况下会释放？
如果一个任务既有 CPU 计算又有 IO 操作，你会怎么选型？
协程和线程的本质区别是什么？为什么协程切换更快？
你在项目中是如何做并发选型的？举一个实际案例。

Q2: 多线程和多进程的区别？怎么选择？「🟡 中级」

考察点：考察对线程和进程本质区别的理解，以及根据实际场景进行技术选型的能力，筛选那些只会死记硬背概念而不会应用的候选人。

参考答案：

核心区别

维度	多线程	多进程
内存空间	线程共享进程的堆内存，有自己独立的栈空间	每个进程有独立的内存空间（地址空间）
创建开销	小（只需分配栈和寄存器等上下文）	大（需要分配独立的地址空间、复制父进程资源等）
切换开销	较小（只需保存和恢复寄存器、程序计数器等）	较大（需要切换页表、刷新 TLB、保存整个进程上下文）
GIL 影响	受 GIL 限制，同一时刻只有一个线程执行 Python 字节码	每个进程有独立的 GIL，不受限制，可真正并行
通信方式	共享变量（需加锁保证线程安全）	IPC 机制（Queue、Pipe、共享内存、Manager 等）
通信速度	快（直接读写内存）	慢（需要内核参与或数据拷贝）
稳定性	一个线程崩溃可能导致整个进程崩溃	进程间相互独立，一个崩溃不影响其他进程
资源占用	少	多
并发数量	几千级别（受栈空间和调度开销限制）	几十到几百级别（受内存和 PID 限制）

深入理解

1. 内存共享的利与弊

线程共享进程的内存空间，这既是优点也是缺点：

优点：线程间通信简单高效，可以直接读写共享变量
缺点：容易出现竞态条件（Race Condition），需要使用锁等同步机制，增加了编程复杂度

python

# 线程安全问题示例
import threading

count = 0

def increment():
    global count
    for _ in range(100000):
        count += 1  # 非原子操作，存在竞态条件

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(count)  # 结果通常小于 1,000,000，而不是等于

进程有独立的内存空间，所以不存在线程安全问题，但通信成本高：

python

# 多进程通信示例
from multiprocessing import Process, Queue

def worker(q):
    q.put("hello from child")

if __name__ == "__main__":
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    print(q.get())  # 从队列中获取数据
    p.join()

2. GIL 的影响

线程：GIL 使得 Python 多线程在 CPU 密集型任务上无法真正并行，性能甚至可能不如单线程（因为有线程切换开销）
进程：每个进程有独立的 Python 解释器和 GIL，可以在多核 CPU 上真正并行执行

3. 稳定性差异

线程共享进程地址空间，一个线程如果出现内存越界等严重错误，可能导致整个进程崩溃
进程有独立的地址空间，一个进程崩溃不会影响其他进程，这也是为什么浏览器等软件使用多进程架构

选型原则

场景	推荐方案	原因
CPU 密集型（数值计算、图像处理等）	多进程	绕过 GIL，真正利用多核
IO 密集型（网络请求、文件读写等）	多线程或协程	IO 时 GIL 会释放，线程切换开销可以接受
高并发 IO 密集型（Web 服务、爬虫等）	协程	极高的并发能力和极低的切换开销
任务需要高隔离度（稳定性优先）	多进程	进程间独立，不会互相影响
任务间需要频繁共享大量数据	多线程	共享内存，通信效率高

混合使用

实际项目中也可以混合使用多进程和多线程/协程：

架构示例：多进程 + 协程
├── 进程1（CPU 核1）：事件循环 + 多个协程处理 IO
├── 进程2（CPU 核2）：事件循环 + 多个协程处理 IO
├── 进程3（CPU 核3）：事件循环 + 多个协程处理 IO
└── 进程4（CPU 核4）：事件循环 + 多个协程处理 IO

这种架构结合了多进程的多核利用和协程的高并发 IO 能力，是高性能 Web 服务器（如 Uvicorn、Gunicorn + Gevent）的常用模式。

追问延伸：

进程间通信有哪些方式？各有什么优缺点？
线程安全问题有哪些解决方法？
Python 中为什么有了 GIL 还需要线程锁？
你在项目中遇到过死锁吗？怎么排查和解决的？

Q3: threading模块的常用方法？线程同步方式有哪些？「🟢 校招/初级」

考察点：考察对 Python 标准库 threading 模块的熟悉程度，以及对各种线程同步原语的理解和使用场景，筛选基础不扎实的初级候选人。

参考答案：

一、Thread 类的常用方法

threading.Thread 是创建线程的核心类：

方法/属性	说明
`Thread(target, args, kwargs, daemon)`	创建线程对象
`start()`	启动线程（调用 run 方法）
`run()`	线程执行的逻辑（可以被子类重写）
`join(timeout)`	等待线程结束
`is_alive()`	检查线程是否存活
`name`	线程名称
`daemon`	是否为守护线程
`ident`	线程标识符

python

import threading
import time

# 方式一：传入 target 函数
def worker(name, delay):
    print(f"Worker {name} started")
    time.sleep(delay)
    print(f"Worker {name} finished")

t = threading.Thread(target=worker, args=("A", 2), daemon=True)
t.start()
t.join()  # 等待线程结束

# 方式二：继承 Thread 类
class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self):
        print(f"Thread {self.name} running")

t = MyThread("B")
t.start()

守护线程（daemon）

守护线程是后台线程，当所有非守护线程结束后，守护线程会自动终止
常用于后台任务，如日志收集、心跳检测等
设置方式：t.daemon = True 或 Thread(daemon=True)

二、线程同步方式

当多个线程访问共享资源时，需要同步机制来保证数据一致性。

1. Lock（互斥锁）

最基本的同步原语，同一时刻只有一个线程能持有锁。

python

import threading

lock = threading.Lock()
count = 0

def increment():
    global count
    for _ in range(100000):
        with lock:  # 自动 acquire/release
            count += 1

# 也可以手动调用：
# lock.acquire()
# count += 1
# lock.release()

特点：

不可重入（同一线程重复 acquire 会导致死锁）
是最常用的同步原语

2. RLock（可重入锁）

可重入锁，同一线程可以多次 acquire，不会死锁，但需要对应次数的 release。

python

rlock = threading.RLock()

def func1():
    with rlock:
        func2()  # 同一线程再次 acquire，不会死锁

def func2():
    with rlock:
        print("func2")

适用场景：递归函数、嵌套调用中需要加锁的情况。

3. Condition（条件变量）

在 Lock 的基础上增加了等待/通知机制，用于线程间协作。

python

import threading
import time

cond = threading.Condition()
items = []
MAX_SIZE = 5

def producer():
    for i in range(10):
        with cond:
            while len(items) >= MAX_SIZE:
                cond.wait()  # 队列满，等待消费者消费
            items.append(i)
            print(f"Produced: {i}")
            cond.notify()  # 通知消费者

def consumer():
    for i in range(10):
        with cond:
            while len(items) == 0:
                cond.wait()  # 队列空，等待生产者生产
            item = items.pop(0)
            print(f"Consumed: {item}")
            cond.notify()  # 通知生产者

常用方法：

wait(timeout)：释放锁并等待，被 notify 后重新获取锁
notify()：唤醒一个等待的线程
notify_all()：唤醒所有等待的线程

适用场景：生产者-消费者模式。

4. Semaphore（信号量）

控制并发数量，允许多个线程同时访问资源。

python

sem = threading.Semaphore(3)  # 最多 3 个线程同时执行

def worker(id):
    with sem:
        print(f"Worker {id} is working")
        time.sleep(1)

特殊情况：Semaphore(1) 等价于 Lock。

适用场景：连接池、限流等需要控制并发数的场景。

5. Event（事件）

用于线程间简单的信号通知，一个线程发出事件，其他线程等待。

python

event = threading.Event()

def waiter():
    print("Waiting for event...")
    event.wait()  # 阻塞等待事件被设置
    print("Event received!")

def setter():
    time.sleep(2)
    event.set()  # 设置事件，唤醒所有等待的线程

常用方法：

set()：设置事件（内部标志为 True）
clear()：清除事件（内部标志为 False）
wait(timeout)：等待事件被设置
is_set()：检查事件是否被设置

适用场景：线程间的简单通知、启动/停止信号。

6. Barrier（屏障）

等待多个线程都到达屏障点后，再一起继续执行。

python

barrier = threading.Barrier(3)  # 需要 3 个线程都到达

def worker(id):
    print(f"Worker {id} before barrier")
    barrier.wait()  # 等待其他线程
    print(f"Worker {id} after barrier")

适用场景：多线程分阶段执行，需要等待所有线程准备好后再开始下一阶段。

同步原语对比

同步原语	用途	关键方法	适用场景
Lock	互斥访问共享资源	`acquire()/release()`	最常用，保护临界区
RLock	可重入的互斥锁	`acquire()/release()`	递归/嵌套加锁
Condition	线程间协作等待	`wait()/notify()/notify_all()`	生产者消费者
Semaphore	控制并发数量	`acquire()/release()`	连接池、限流
Event	简单事件通知	`set()/clear()/wait()`	启动/停止信号
Barrier	多线程同步点	`wait()`	分阶段并行计算

追问延伸：

Lock 和 RLock 的区别？什么场景下需要用 RLock？
生产者消费者模式用什么同步原语实现？为什么不用 Event？
Semaphore 和 Lock 的关系？
为什么 Condition 需要配合 Lock 使用？

Q4: 进程间通信（IPC）的方式有哪些？Python中怎么实现？「🟡 中级」

考察点：考察对进程间通信机制的理解，以及在 Python 中实际使用的经验，筛选只会用多线程、对多进程编程缺乏实践的候选人。

参考答案：

进程间通信（Inter-Process Communication, IPC）是多进程编程中的核心问题。由于进程有独立的内存空间，进程间不能直接共享数据，需要通过操作系统提供的机制进行通信。

Python 中常见的 IPC 方式

1. Queue（队列）

multiprocessing.Queue 是最常用的 IPC 方式，基于管道和锁实现，线程/进程安全。

python

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(i)
        print(f"Produced: {i}")

def consumer(q):
    while True:
        item = q.get()
        if item is None:  # 哨兵值，表示结束
            break
        print(f"Consumed: {item}")

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))
    p1.start()
    p2.start()
    p1.join()
    q.put(None)  # 发送结束信号
    p2.join()

特点：

简单易用，进程安全
先进先出（FIFO）
数据需要序列化（pickle），有性能开销
适合大多数场景

2. Pipe（管道）

Pipe 是双向的，可以用于两个进程之间的通信，比 Queue 更轻量。

python

from multiprocessing import Process, Pipe

def worker(conn):
    conn.send("Hello from child")
    data = conn.recv()
    print(f"Child received: {data}")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=worker, args=(child_conn,))
    p.start()
    data = parent_conn.recv()
    print(f"Parent received: {data}")
    parent_conn.send("Hello from parent")
    p.join()

特点：

双向通信（默认全双工）
比 Queue 更快（实现更简单）
适合两个进程间的点对点通信
多进程使用时需要注意锁的问题

3. 共享内存（Shared Memory）

多个进程可以访问同一块内存区域，速度最快。

方式一：Value / Array

python

from multiprocessing import Process, Value, Array

def worker(n, arr):
    n.value = 3.14
    for i in range(len(arr)):
        arr[i] = arr[i] * 2

if __name__ == "__main__":
    num = Value('d', 0.0)  # 'd' 表示 double
    arr = Array('i', range(10))  # 'i' 表示 int
    
    p = Process(target=worker, args=(num, arr))
    p.start()
    p.join()
    
    print(num.value)  # 3.14
    print(arr[:])     # [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

方式二：SharedMemory（Python 3.8+）

python

from multiprocessing import shared_memory
import numpy as np

# 创建共享内存
shm = shared_memory.SharedMemory(create=True, size=100)

# 在子进程中访问同一块共享内存
# 通过名称连接：shm = shared_memory.SharedMemory(name="shm_name")

# 使用 numpy 操作共享内存
arr = np.ndarray((10,), dtype=np.int32, buffer=shm.buf)
arr[:] = range(10)

# 用完释放
shm.close()
shm.unlink()

特点：

速度最快（直接内存访问，无需数据拷贝）
需要手动同步（配合 Lock 使用）
适合大数据量的共享
SharedMemory 比 Value/Array 更灵活

4. Manager（管理器）

Manager 提供了一种更高层的 IPC 方式，支持多种数据类型，还可以跨网络。

python

from multiprocessing import Process, Manager

def worker(d, l):
    d['key'] = 'value'
    d['count'] = 42
    l.append('new item')

if __name__ == "__main__":
    with Manager() as manager:
        d = manager.dict()
        l = manager.list([1, 2, 3])
        
        p = Process(target=worker, args=(d, l))
        p.start()
        p.join()
        
        print(dict(d))  # {'key': 'value', 'count': 42}
        print(list(l))  # [1, 2, 3, 'new item']

特点：

支持 dict、list、Lock、Condition 等多种类型
使用简单，像普通 Python 对象一样操作
可以跨网络通信（BaseManager）
速度较慢（通过进程间的代理对象通信，有序列化开销）

5. 其他同步原语

multiprocessing 模块也提供了类似 threading 的同步原语：

Lock / RLock：进程间互斥锁
Semaphore：进程间信号量
Event：进程间事件
Condition：进程间条件变量

这些同步原语可以配合共享内存使用，保证数据一致性。

6. Socket / RPC（跨机器）

如果需要跨机器通信，可以使用：

Socket：最底层的网络通信
RPC：如 gRPC、XML-RPC（标准库 xmlrpc）
消息队列：如 RabbitMQ、Kafka、Redis 等

各种 IPC 方式对比

方式	速度	复杂度	适用场景	数据量	是否需要同步
Queue	中等	低	通用消息传递	中小	不需要（内置锁）
Pipe	快	低	两进程双向通信	中小	不需要
共享内存	最快	中	大数据共享	大	需要（手动加锁）
Manager	慢	低	灵活数据结构，跨网络	小	不需要（内置同步）
Socket/RPC	慢	高	跨机器通信	不定	视协议而定

选型建议

简单消息传递 → Queue（最常用，最安全）
两进程点对点 → Pipe（比 Queue 更快）
大数据量共享 → 共享内存（配合 Lock）
需要复杂数据结构 → Manager（使用方便）
跨机器通信 → Socket / RPC / 消息队列

追问延伸：

Queue 和 Pipe 的底层实现有什么区别？
共享内存为什么快？有什么缺点？
Manager 的实现原理是什么？为什么比共享内存慢？
你在项目中用过哪种 IPC 方式？为什么选它？
multiprocessing.Queue 和 queue.Queue 有什么区别？

Q5: concurrent.futures 模块是什么？ThreadPoolExecutor 和 ProcessPoolExecutor？「🟡 中级」

考察点：考察对 Python 高层并发抽象的理解，是否熟悉线程池/进程池的统一接口，筛选只会用底层 threading/multiprocessing 而不会用更现代接口的候选人。

参考答案：

concurrent.futures 是 Python 3.2 引入的高层并发模块，提供了统一的接口来管理线程池和进程池，大大简化了并发编程的代码。

核心概念

1. Executor（执行器）

Executor 是抽象基类，有两个具体实现：

ThreadPoolExecutor：线程池执行器
ProcessPoolExecutor：进程池执行器

两者接口完全一致，可以方便地切换。

2. Future（未来对象）

Future 表示一个异步计算的结果，可以用它来查询任务状态、获取结果或异常。

基本用法

ThreadPoolExecutor 示例

python

from concurrent.futures import ThreadPoolExecutor
import requests

def fetch(url):
    resp = requests.get(url)
    return resp.status_code

urls = ["https://example.com"] * 10

# 方式一：submit + result（逐个提交）
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(fetch, url) for url in urls]
    for future in futures:
        print(future.result())  # 阻塞等待结果

# 方式二：map（批量映射，按输入顺序返回）
with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(fetch, urls)
    for result in results:
        print(result)

ProcessPoolExecutor 示例

python

from concurrent.futures import ProcessPoolExecutor
import math

def is_prime(n):
    if n < 2:
        return False
    for i in range(2, int(math.isqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

numbers = [999999937, 999999929, 99999989, 99999983]

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(is_prime, numbers))
    print(results)

注意：ProcessPoolExecutor 中传递的函数和参数必须可序列化（pickle）。

重要 API 详解

1. submit() vs map()

特性	submit()	map()
返回值	Future 对象列表	结果迭代器
顺序	提交顺序（获取结果顺序不定）	严格按输入顺序
异常处理	调用 result() 时抛出	迭代时抛出
灵活性	高（可以单独取消、添加回调）	低（批量处理）
超时控制	result(timeout=)	不直接支持

2. as_completed() — 按完成顺序获取结果

python

from concurrent.futures import as_completed

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(fetch, url): url for url in urls}
    for future in as_completed(futures):
        url = futures[future]
        try:
            result = future.result()
            print(f"{url}: {result}")
        except Exception as e:
            print(f"{url} failed: {e}")

as_completed() 会在任务完成时立即返回对应的 Future，不按提交顺序，适合需要尽早处理结果的场景。

3. wait() — 等待任务完成

python

from concurrent.futures import wait, FIRST_COMPLETED, ALL_COMPLETED

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(fetch, url) for url in urls]
    
    # 等待所有任务完成
    done, not_done = wait(futures, return_when=ALL_COMPLETED)
    
    # 等待第一个任务完成
    # done, not_done = wait(futures, return_when=FIRST_COMPLETED)
    
    # 等待超时
    # done, not_done = wait(futures, timeout=5)

4. Future 对象的方法

方法	说明
`result(timeout=None)`	获取结果（阻塞），超时抛 TimeoutError
`exception(timeout=None)`	获取异常（阻塞）
`done()`	是否完成
`cancelled()`	是否被取消
`cancel()`	尝试取消任务（未开始才可能成功）
`add_done_callback(fn)`	添加完成回调

python

# 回调示例
def on_complete(future):
    print(f"Task completed: {future.result()}")

future = executor.submit(fetch, url)
future.add_done_callback(on_complete)

线程池/进程池的优势

简化代码：统一的高层 API，不用手动管理线程/进程的创建和生命周期
资源复用：线程/进程预先创建，任务来了直接用，避免频繁创建销毁的开销
并发控制：通过 max_workers 控制最大并发数，防止资源耗尽
Future 模式：方便的异步编程模型，支持回调、超时、取消等操作
接口统一：线程池和进程池接口一致，可以根据场景灵活切换

注意事项

ProcessPoolExecutor 的限制

必须可序列化：传递给进程池的函数和参数必须支持 pickle 序列化
不能用 lambda：lambda 函数不可序列化，不能直接传递给 ProcessPoolExecutor
全局变量不共享：每个进程有独立的内存空间，全局变量不共享
if name == "main"：Windows 下必须在主模块中使用，否则会无限递归创建子进程

线程池大小的选择

IO 密集型：可以设置较大，如 2 * CPU核数 或更多（因为线程大部分时间在等待）
CPU 密集型：建议设置为 CPU 核数（避免频繁上下文切换）
默认值：min(32, os.cpu_count() + 4)（Python 3.8+）

对比：底层 API vs concurrent.futures

维度	threading / multiprocessing	concurrent.futures
抽象层级	底层，手动管理	高层，自动管理
代码复杂度	高	低
灵活性	高（可以精细控制）	中（常用场景足够）
结果获取	需手动实现	Future 对象，方便
推荐程度	复杂场景用	大多数场景优先用

追问延伸：

submit 和 map 的区别？各自适用什么场景？
as_completed 和 wait 的区别？
ProcessPoolExecutor 有什么限制？为什么函数必须可序列化？
线程池的大小怎么设置？有什么依据？
Future 对象和协程有什么关系？
你在项目中用过 concurrent.futures 吗？解决了什么问题？

Q6: asyncio 的事件循环原理？核心概念有哪些？「🔴 高级」

考察点：考察对 asyncio 底层原理的深入理解，是否清楚事件循环的工作机制和核心组件，筛选只会用 async/await 语法而不理解原理的候选人。

参考答案：

asyncio 是 Python 的异步编程框架，核心是事件循环（Event Loop），它是一个单线程的循环，负责监听和分发事件、调度协程执行。

一、事件循环的工作原理

事件循环本质上是一个 while True 循环，不断地轮询事件并处理：

┌─────────────────────────────────────────┐
│           事件循环 (Event Loop)          │
│  ┌───────┐    ┌────────┐    ┌────────┐ │
│  │ 等待IO │ →  │ 就绪队列 │ →  │ 执行回调 │ │
│  └───────┘    └────────┘    └────────┘ │
│                  ↑                      │
│                  └──────────────────────┘
└─────────────────────────────────────────┘

基本流程：

事件循环维护一个待处理的任务队列
循环检查哪些 IO 操作已经就绪（通过 selectors 模块）
将就绪的任务/回调加入执行队列
依次执行回调，回调中可能创建新的任务
重复上述过程，直到所有任务完成

二、核心概念

1. 协程（Coroutine）

使用 async def 定义的函数，调用时不会立即执行，而是返回一个协程对象。

python

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

coro = hello()  # 协程对象，不会执行
asyncio.run(coro)  # 交给事件循环执行

协程本身只是一个函数对象，需要被包装成 Task 加入事件循环才能被调度执行。

2. Task（任务）

Task 是协程的包装器，将协程提交给事件循环调度。Task 是 Future 的子类。

python

async def main():
    # 创建 Task（立即被调度）
    task1 = asyncio.create_task(hello())
    task2 = asyncio.create_task(hello())
    
    # 等待任务完成
    await task1
    await task2

asyncio.run(main())

Task vs 协程：

协程：只是一个函数对象，调用不执行
Task：协程的调度单元，加入事件循环后会自动调度执行
await 协程：只是串行等待，不会并发
await asyncio.gather(task1, task2)：并发执行多个任务

3. Future（期程）

Future 是一个底层概念，表示一个异步操作的结果（占位符）。Task 是 Future 的子类。

Future 有状态：PENDING → RUNNING → DONE（正常完成或异常）
可以设置结果：set_result()
可以添加回调：add_done_callback()

python

async def main():
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    
    # 模拟异步操作完成后设置结果
    loop.call_later(1, lambda: future.set_result("done"))
    
    result = await future  # 等待 future 完成
    print(result)

大多数情况下不需要直接操作 Future，用 Task 即可。

4. 事件循环（Event Loop）

事件循环是 asyncio 的核心，负责：

注册和监听 IO 事件
调度协程/任务执行
处理定时器（call_later、call_at）
处理回调（call_soon）

python

# 获取事件循环
loop = asyncio.get_event_loop()  # 旧方式
loop = asyncio.get_running_loop()  # 新方式（在协程中使用）

# 运行协程
loop.run_until_complete(coro)

# 调度回调
loop.call_soon(callback)  # 尽快执行
loop.call_later(1, callback)  # 1秒后执行
loop.call_at(when, callback)  # 指定时间执行

三、事件循环的执行流程

以 asyncio.gather(task1, task2) 为例：

1. 创建两个 Task，加入事件循环的就绪队列
2. 事件循环取出 task1 执行
3. task1 执行到 await asyncio.sleep(1)，挂起自己
   → 注册一个 1 秒后的定时器回调
   → 让出控制权给事件循环
4. 事件循环取出 task2 执行
5. task2 执行到 await asyncio.sleep(2)，挂起自己
   → 注册一个 2 秒后的定时器回调
   → 让出控制权给事件循环
6. 事件循环进入等待状态（没有就绪任务）
7. 1 秒后，task1 的定时器触发，task1 恢复执行
8. task1 执行完成，标记为 DONE
9. 又过了 1 秒，task2 的定时器触发，task2 恢复执行
10. task2 执行完成，标记为 DONE
11. gather 发现所有任务都完成了，返回结果

四、底层实现：selectors 模块

asyncio 的 IO 多路复用底层依赖 selectors 模块，它封装了操作系统提供的 IO 多路复用机制：

系统	机制	说明
Linux	epoll	高效，支持大量连接
macOS/BSD	kqueue	类似 epoll
Solaris	/dev/poll	类似 epoll
Windows	IOCP	完成端口
通用	select	兼容性好，但效率低

事件循环通过 selectors 监听文件描述符（socket 等）的可读/可写事件，当事件就绪时，唤醒对应的协程继续执行。

五、事件循环的类型

Python 提供了不同的事件循环实现：

SelectorEventLoop：默认实现，基于 selectors，适合大多数场景
ProactorEventLoop：Windows 专用，基于 IOCP（Windows 上性能更好）
uvloop：第三方实现，基于 libuv，性能比默认实现高 2-4 倍

python

# 使用 uvloop（需要安装）
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

六、为什么是单线程？

asyncio 采用单线程事件循环模型，原因是：

IO 密集型任务的瓶颈在等待，而不是 CPU 计算
单线程避免了线程切换的开销和线程安全问题
协程切换的开销远小于线程切换
编程模型相对简单（没有多线程的竞态条件）

如果需要利用多核 CPU，可以结合多进程使用（每个进程运行一个事件循环）。

追问延伸：

async/await 的底层原理是什么？和生成器有什么关系？
事件循环是单线程的，为什么能处理高并发？
Task 和 Future 的区别和联系？
什么是 uvloop？为什么更快？
事件循环在什么时候会切换协程？
如果在协程中执行了阻塞操作会怎么样？

Q7: async/await 的底层原理是什么？「🔴 高级」

考察点：考察对 Python 异步编程底层机制的深度理解，是否清楚 async/await 语法糖背后的实现原理，筛选只会用语法而不懂原理的候选人。

参考答案：

async/await 是 Python 3.5 引入的语法糖，底层基于**生成器（generator）**实现。理解其原理需要从生成器和 yield from 讲起。

一、从生成器到协程

1. 生成器基础

生成器是可以暂停和恢复的函数，使用 yield 关键字：

python

def gen():
    yield 1
    yield 2
    yield 3

g = gen()
print(next(g))  # 1
print(next(g))  # 2
print(next(g))  # 3

生成器可以通过 yield 暂停执行，通过 next() 或 send() 恢复执行。这是协程的基础。

2. yield from：生成器的委托

Python 3.3 引入了 yield from，可以在生成器中调用另一个生成器：

python

def inner():
    yield 1
    yield 2

def outer():
    yield from inner()  # 委托给 inner 生成器
    yield 3

for x in outer():
    print(x)  # 1, 2, 3

yield from 不仅可以委托，还可以在调用者和子生成器之间建立双向通道：

值可以从调用者直接传到子生成器
异常可以从子生成器直接传到调用者
返回值可以通过 return 从子生成器传回

3. 基于生成器的协程

在 async/await 出现之前，Python 就有基于生成器的协程（使用 @asyncio.coroutine 装饰器 + yield from）：

python

# Python 3.4 风格的协程（已废弃）
import asyncio

@asyncio.coroutine
def old_style_coro():
    yield from asyncio.sleep(1)
    return "done"

这本质上是把生成器当作协程来用。async/await 就是这种模式的语法糖。

二、async/await 的本质

1. async def：定义协程函数

async def 定义的函数，调用时返回一个协程对象，而不是立即执行函数体。

python

async def foo():
    return 42

coro = foo()
print(type(coro))  # <class 'coroutine'>

协程对象和生成器对象非常相似，都是可以暂停和恢复的执行单元。实际上，协程在 CPython 内部的实现和生成器共享很多代码。

2. await：挂起和恢复

await 是 yield from 的升级版，用于挂起当前协程，等待另一个 awaitable 对象完成。

python

async def bar():
    result = await foo()  # 挂起 bar，等待 foo 完成
    return result

await 的执行过程：

遇到 await 时，当前协程挂起（暂停执行）
控制权返回给事件循环
事件循环去执行其他就绪的协程
当 await 的对象完成后，事件循环将该协程重新加入就绪队列
协程从 await 处恢复执行，拿到结果

三、协程的状态

协程有四种状态（类似生成器）：

状态	说明
`GEN_CREATED`	协程已创建，尚未开始执行
`GEN_RUNNING`	协程正在执行
`GEN_SUSPENDED`	协程在 await 处挂起
`GEN_CLOSED`	协程执行完成或被关闭

可以用 inspect.getcoroutinestate() 查看协程状态：

python

import inspect

async def foo():
    await asyncio.sleep(1)

coro = foo()
print(inspect.getcoroutinestate(coro))  # CORO_CREATED

四、事件循环如何驱动协程

事件循环通过不断地 send() 来推进协程的执行：

python

# 简化的事件循环驱动协程的伪代码
def run_coroutine(coro):
    try:
        # 启动协程，执行到第一个 await
        future = coro.send(None)
        
        # 等待 future 完成
        while not future.done():
            # ... 事件循环做其他事情 ...
            pass
        
        # future 完成后，把结果发送回协程，继续执行
        result = future.result()
        coro.send(result)
        
    except StopIteration as e:
        # 协程执行完毕，返回值在 e.value 中
        return e.value

实际的事件循环要复杂得多，它需要：

管理大量的协程/任务
监听 IO 事件
处理定时器
处理异常和取消

五、Task 的作用

协程本身不能被事件循环直接调度，需要包装成 Task：

协程 (Coroutine) → 包装 → Task → 加入事件循环调度

Task 的作用：

将协程注册到事件循环
驱动协程执行（不断 send）
保存协程的状态和结果
支持取消、回调等操作

可以把 Task 理解为"协程的管理者"，它负责协程的生命周期。

六、awaitable 对象

await 后面可以跟什么？答案是 awaitable 对象，包括：

协程（coroutine）：async def 定义的函数返回的对象
Task：协程的包装
Future：异步结果的占位符
实现了 __await__ 方法的对象：自定义的 awaitable

python

# 自定义 awaitable 对象
class MyAwaitable:
    def __await__(self):
        # __await__ 必须返回一个生成器
        yield "suspend"
        return 42

async def main():
    result = await MyAwaitable()
    print(result)  # 42

七、与生成器的区别

虽然协程底层基于生成器，但两者有本质区别：

特性	生成器	协程
目的	生成数据序列	异步协作
关键字	yield / yield from	async / await
驱动者	调用者（next/send）	事件循环
返回值	通过 yield 产出	通过 return 返回
类型名	generator	coroutine
能否 await	不能（除非加装饰器）	能

八、总结

async def 定义的函数返回协程对象，本质上是一种特殊的生成器
await 是 yield from 的升级版，用于挂起协程并等待另一个 awaitable
协程由事件循环驱动，通过类似 send() 的机制推进执行
Task 是协程的包装，负责协程的调度和管理
整个异步体系建立在"单线程 + 协作式调度"的基础上

追问延伸：

生成器和协程有什么区别和联系？
为什么说 async/await 是语法糖？没有它能不能实现异步？
事件循环是如何知道什么时候该恢复一个挂起的协程的？
await 后面可以跟什么？什么是 awaitable？
协程的栈帧存在哪里？和函数调用栈有什么不同？
你能自己实现一个简单的事件循环吗？

Q8: 异步编程中有哪些常见的陷阱？「🔴 高级」

考察点：考察候选人的实战经验，是否踩过异步编程的坑，筛选只会写简单异步代码而缺乏生产环境经验的候选人。

参考答案：

异步编程虽然能带来高并发，但也有很多容易踩的坑。以下是常见的陷阱：

1. 阻塞调用阻塞整个事件循环

这是最常见也是最严重的问题。在 asyncio 中，任何阻塞调用（同步 IO、CPU 密集型计算）都会阻塞整个事件循环，导致所有协程都被卡住。

python

import asyncio
import requests  # 同步库

async def fetch(url):
    # 错误：requests 是同步的，会阻塞整个事件循环
    resp = requests.get(url)  # ❌ 阻塞！
    return resp.text

async def other_task():
    while True:
        print("Working...")
        await asyncio.sleep(1)

async def main():
    # fetch 中的阻塞会导致 other_task 也无法执行
    await asyncio.gather(
        fetch("https://example.com"),
        other_task()
    )

解决方法：

使用对应的异步库（如 aiohttp 替代 requests）
使用 run_in_executor 将阻塞调用放到线程池中执行

python

async def fetch(url):
    # 方法一：使用异步库
    # async with aiohttp.ClientSession() as session:
    #     async with session.get(url) as resp:
    #         return await resp.text()
    
    # 方法二：run_in_executor 把阻塞调用放到线程池
    loop = asyncio.get_running_loop()
    resp = await loop.run_in_executor(None, requests.get, url)
    return resp.text

2. 未 await 的协程不会执行

调用 async def 函数只是创建了协程对象，不会执行。如果忘记 await，协程永远不会运行，还会产生 RuntimeWarning。

python

async def do_something():
    print("Doing something")
    await asyncio.sleep(1)
    print("Done")

async def main():
    do_something()  # ❌ 只创建了协程对象，不会执行！
    # 正确：await do_something()
    # 或者：asyncio.create_task(do_something())

运行时会警告：

RuntimeWarning: coroutine 'do_something' was never awaited

3. 异常吞噬

Task 的异常如果不主动获取，会被"吞掉"，导致问题难以排查。

python

async def buggy_task():
    await asyncio.sleep(1)
    raise ValueError("something went wrong")

async def main():
    task = asyncio.create_task(buggy_task())
    await asyncio.sleep(2)
    # 任务已经异常结束了，但这里不会有任何报错
    print("Main finished")

解决方法：

await 任务（异常会被抛出）
使用 add_done_callback 检查异常
配置全局异常处理器

python

# 方法一：await
try:
    await task
except Exception as e:
    print(f"Error: {e}")

# 方法二：回调
def on_done(future):
    if future.exception():
        print(f"Task failed: {future.exception()}")

task.add_done_callback(on_done)

# 方法三：全局异常处理器
def exception_handler(loop, context):
    print(f"Exception: {context['exception']}")

loop = asyncio.get_event_loop()
loop.set_exception_handler(exception_handler)

4. 无限制创建任务导致资源耗尽

协程虽然轻量，但也不是无限的。无限制地创建任务可能导致：

内存耗尽
连接数耗尽（数据库连接、文件描述符等）
事件循环负担过重

python

async def process(url):
    await asyncio.sleep(0.1)

async def main():
    urls = [f"https://example.com/{i}" for i in range(100000)]
    # ❌ 一次性创建 10 万个任务，可能导致资源耗尽
    tasks = [asyncio.create_task(process(url)) for url in urls]
    await asyncio.gather(*tasks)

解决方法：使用 asyncio.Semaphore 控制并发数

python

async def process(url, sem):
    async with sem:
        await asyncio.sleep(0.1)

async def main():
    sem = asyncio.Semaphore(100)  # 最多 100 个并发
    urls = [f"https://example.com/{i}" for i in range(100000)]
    tasks = [asyncio.create_task(process(url, sem)) for url in urls]
    await asyncio.gather(*tasks)

5. 取消任务不一定立即生效

task.cancel() 只是请求取消，任务不一定会立即停止，取决于协程的实现。

python

async def long_task():
    try:
        # 计算密集型操作，不会响应取消
        for i in range(10000000):
            pass
        await asyncio.sleep(10)  # 只有 await 点才能响应取消
    except asyncio.CancelledError:
        print("Cancelled!")
        raise

async def main():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(1)
    task.cancel()  # 请求取消
    try:
        await task
    except asyncio.CancelledError:
        print("Task was cancelled")

要点：

取消是通过在 await 点抛出 CancelledError 实现的
如果协程在执行同步代码（没有 await），取消不会立即生效
应该在合适的地方处理 CancelledError，做好清理工作

6. 线程安全问题

asyncio 不是线程安全的。不要在其他线程中直接操作事件循环或调用 asyncio 的 API。

python

# 错误：在另一个线程中调用 asyncio 的方法
def worker_thread(loop):
    # ❌ 不安全！
    asyncio.run_coroutine_threadsafe(coro(), loop)  # 这个是安全的
    # 但很多操作不是

解决方法：使用 call_soon_threadsafe 或 run_coroutine_threadsafe

python

# 正确：线程安全的方式
def worker_thread(loop):
    # 从其他线程向事件循环提交协程
    asyncio.run_coroutine_threadsafe(coro(), loop)
    
    # 从其他线程调度回调
    loop.call_soon_threadsafe(callback, arg)

7. 混用同步和异步代码导致死锁

在异步代码中使用同步锁可能导致死锁，因为锁会阻塞整个事件循环。

python

import threading

lock = threading.Lock()

async def task1():
    with lock:  # ❌ 线程锁会阻塞整个事件循环！
        await asyncio.sleep(1)

async def task2():
    with lock:  # 如果 task1 持有锁并挂起，task2 会在这里阻塞整个事件循环
        await asyncio.sleep(1)

解决方法：使用 asyncio.Lock 而不是 threading.Lock

python

lock = asyncio.Lock()  # ✅ 协程锁，只挂起协程，不阻塞线程

async def task1():
    async with lock:
        await asyncio.sleep(1)

8. 上下文变量（ContextVar）的使用

异步代码中使用全局变量可能导致问题，因为多个协程在同一线程中交替执行，会互相干扰。

解决方法：使用 contextvars.ContextVar 管理协程上下文

python

from contextvars import ContextVar

user_var = ContextVar("user")

async def process_request(user_id):
    user_var.set(user_id)
    await do_something()
    print(f"Processing user {user_var.get()}")

陷阱总结

陷阱	后果	解决方法
阻塞调用阻塞事件循环	所有协程卡住	用异步库或 run_in_executor
忘记 await 协程	协程不执行，RuntimeWarning	记得 await 或 create_task
异常吞噬	错误难以排查	await 任务 / add_done_callback
无限制创建任务	资源耗尽	用 Semaphore 限流
取消不立即生效	任务可能继续运行	合理设计取消点
跨线程操作	数据竞争、崩溃	用 call_soon_threadsafe
混用同步锁	死锁	用 asyncio.Lock
全局变量串扰	数据混乱	用 ContextVar

追问延伸：

如何在异步代码中处理 CPU 密集型任务？
为什么 asyncio 不是线程安全的？哪些操作是线程安全的？
如何优雅地取消一组任务？
你在项目中遇到过哪些异步编程的坑？怎么解决的？
什么是"异步地狱"？如何避免？
如何对异步代码进行单元测试？

Q9: 什么是协程安全？asyncio.Lock 和线程锁的区别？「🟡 中级」

考察点：考察对协程同步机制的理解，是否清楚协程锁和线程锁的本质区别，筛选以为异步就不需要锁、或把线程锁用到异步代码里的候选人。

参考答案：

一、协程也需要锁吗？

很多人以为协程是单线程的，不会有并发安全问题，这是错误的。

协程安全问题示例：

python

import asyncio

balance = 100

async def withdraw(amount):
    global balance
    if balance >= amount:
        # 模拟检查后的一些操作（如网络请求、数据库查询等）
        await asyncio.sleep(0.1)  # 这里会切换到其他协程
        balance -= amount
        print(f"Withdrew {amount}, balance: {balance}")
    else:
        print(f"Insufficient funds, balance: {balance}")

async def main():
    # 两个协程同时取钱
    await asyncio.gather(
        withdraw(80),
        withdraw(80)
    )
    print(f"Final balance: {balance}")  # 可能是负数！

asyncio.run(main())

执行结果可能是 balance = -60，因为：

协程 1 检查余额 100 >= 80，通过
协程 1 在 await 处挂起
协程 2 检查余额 100 >= 80，也通过
协程 2 在 await 处挂起
两个协程都执行扣款，余额变成负数

结论：只要存在多个协程修改共享资源，且中间有 await 切换点，就可能出现竞态条件，需要锁来保护。

二、asyncio.Lock（协程锁）

asyncio.Lock 是 asyncio 提供的互斥锁，专为协程设计。

python

import asyncio

lock = asyncio.Lock()
balance = 100

async def withdraw(amount):
    global balance
    async with lock:  # 获取锁（挂起等待，不阻塞线程）
        if balance >= amount:
            await asyncio.sleep(0.1)
            balance -= amount
            print(f"Withdrew {amount}, balance: {balance}")
        else:
            print(f"Insufficient funds, balance: {balance}")

async def main():
    await asyncio.gather(withdraw(80), withdraw(80))
    print(f"Final balance: {balance}")  # 20，正确！

asyncio.run(main())

asyncio.Lock 的特点：

当锁被占用时，等待的协程会挂起（不是阻塞线程）
事件循环可以继续执行其他协程
锁释放后，等待的协程被唤醒，继续执行
是协作式的，依赖 await 切换

三、asyncio.Lock vs threading.Lock

特性	asyncio.Lock（协程锁）	threading.Lock（线程锁）
等待时的行为	挂起协程，线程继续运行	阻塞整个线程
切换开销	极小（用户态，几纳秒）	大（内核态，几微秒）
调度者	事件循环（程序调度）	操作系统（内核调度）
适用范围	同一线程内的协程之间	不同线程之间
能否用在异步代码中	推荐使用	绝对不要用（会死锁）
是否可重入	不可重入（还有 RLock）	不可重入（还有 RLock）

四、为什么异步代码中不能用线程锁？

python

import asyncio
import threading

lock = threading.Lock()

async def task1():
    with lock:
        print("Task 1 got lock")
        await asyncio.sleep(1)  # 挂起协程，但锁还没释放
        print("Task 1 releasing lock")

async def task2():
    print("Task 2 waiting for lock")
    with lock:  # ❌ 阻塞整个线程！事件循环也卡住了
        print("Task 2 got lock")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())  # 程序会卡死！

死锁原因：

task1 获取线程锁，然后在 await asyncio.sleep(1) 处挂起
task2 尝试获取线程锁，但锁被 task1 持有
线程锁是阻塞的，task2 会阻塞整个线程
事件循环被阻塞，task1 永远没有机会恢复执行来释放锁
→ 死锁！

五、asyncio 中的其他同步原语

asyncio 也提供了类似 threading 的同步原语，都是协程级别的：

同步原语	说明
`asyncio.Lock`	互斥锁
`asyncio.RLock`	可重入锁
`asyncio.Condition`	条件变量（wait/notify）
`asyncio.Semaphore`	信号量（控制并发数）
`asyncio.Event`	事件（set/wait）
`asyncio.Barrier`	屏障（Python 3.11+）
`asyncio.Queue`	异步队列（生产者消费者）

这些同步原语的用法和 threading 模块类似，但都是 async 的，需要用 await 或 async with。

六、协程锁的适用场景

保护共享资源：多个协程修改同一个变量
限制并发数：用 Semaphore 控制同时进行的 IO 操作数量
协程间协作：用 Event/Condition 实现协程间的同步

七、什么时候不需要锁？

单协程操作共享资源：没有并发修改，不需要锁
原子操作：单个字节码指令是原子的（如 list.append），但尽量不要依赖这个
消息传递替代共享内存：用 Queue 传递消息而不是共享变量（CSP 模型）

python

# 用 Queue 代替共享变量（推荐的模式）
async def producer(queue):
    for i in range(10):
        await queue.put(i)

async def consumer(queue):
    while True:
        item = await queue.get()
        # 处理 item
        queue.task_done()

追问延伸：

协程是单线程的，为什么还需要锁？
在异步代码中使用线程锁会有什么问题？为什么？
asyncio.Lock 和 threading.Lock 底层实现有什么区别？
什么是"原子操作"？Python 中哪些操作是原子的？
协程锁和线程锁哪个更快？为什么？
除了锁，还有什么方式可以保证协程安全？

Q10: GIL 在什么情况下会释放？「🟡 中级」

考察点：考察对 GIL（全局解释器锁）工作机制的深入理解，是否清楚 GIL 的释放时机，筛选对 GIL 只有模糊概念的候选人。

参考答案：

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 中的一把互斥锁，确保同一时刻只有一个线程执行 Python 字节码。但 GIL 不是一直占着不放的，它会在特定时机释放。

一、GIL 的释放时机

1. IO 操作时释放

当线程遇到 IO 操作（网络请求、文件读写、sleep 等）时，GIL 会被释放，让其他线程有机会执行。

python

import threading
import time

def io_task():
    # IO 操作会释放 GIL
    time.sleep(1)  # sleep 是 IO 操作，释放 GIL
    # 文件读写也会释放 GIL
    # open("file.txt").read()
    # 网络请求也会释放 GIL
    # requests.get("https://example.com")

原理：IO 操作通常涉及系统调用，线程会进入等待状态。在进入等待之前，Python 会释放 GIL，这样 CPU 不会被浪费。

2. 时间片到期（计算密集型也会释放）

Python 3.2+ 引入了新的 GIL 机制：基于时间的抢占。

默认情况下，每 15 毫秒（sys.getswitchinterval() 可查），当前线程会检查是否需要释放 GIL
如果有其他线程在等待 GIL，当前线程会释放 GIL，让其他线程有机会运行
这意味着即使是纯计算密集型代码，GIL 也会周期性释放

python

import sys
print(sys.getswitchinterval())  # 默认 0.005 秒（Python 3.11），之前是 0.015 秒

注意：虽然 GIL 会周期性释放，但计算密集型任务用多线程仍然不会比单线程更快，因为：

同一时刻只有一个线程执行 Python 字节码
线程切换有开销
GIL 的争抢也有开销

3. C 扩展中手动释放

在 C 扩展中，可以手动释放 GIL，让 Python 线程在执行 C 代码时不持有 GIL。

// C 扩展中释放 GIL 的示例
#include "Python.h"

static PyObject* my_function(PyObject* self, PyObject* args) {
    // 释放 GIL
    Py_BEGIN_ALLOW_THREADS
    
    // 这里执行耗时的 C 代码，不持有 GIL
    // 其他 Python 线程可以并行执行
    
    // 重新获取 GIL
    Py_END_ALLOW_THREADS
    
    Py_RETURN_NONE;
}

很多常用库的 C 扩展都会在适当的地方释放 GIL，例如：

numpy：在大型数组运算时释放 GIL
hashlib：在计算哈希时释放 GIL
zlib：在压缩/解压时释放 GIL

这也是为什么有时候 numpy 的多线程计算会有一定加速效果的原因。

4. 调用某些 C 函数

某些 C 标准库函数（如 sleep、文件读写等）在调用时，Python 解释器会自动释放 GIL。

二、GIL 的获取机制

GIL 的获取不是简单的"先到先得"，有一套复杂的机制：

线程请求 GIL：线程想要执行 Python 字节码时，需要先获取 GIL
如果 GIL 空闲：直接获取
如果 GIL 被占用：线程进入等待状态
持有 GIL 的线程释放时：会通知等待的线程
线程被唤醒后：尝试获取 GIL

在 Python 3.2 之前，GIL 的调度策略是"票机制"，导致 IO 密集型线程和 CPU 密集型线程混跑时性能很差。Python 3.2 之后改成了基于时间片的抢占机制，改善了这种情况。

三、验证 GIL 释放的实验

python

import threading
import time

# CPU 密集型任务
def cpu_bound():
    count = 0
    for _ in range(10**7):
        count += 1
    return count

# IO 密集型任务
def io_bound():
    time.sleep(1)

# 实验 1：单线程 CPU 密集型
start = time.time()
cpu_bound()
print(f"单线程 CPU: {time.time() - start:.2f}s")

# 实验 2：多线程 CPU 密集型
start = time.time()
threads = [threading.Thread(target=cpu_bound) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"4线程 CPU: {time.time() - start:.2f}s")
# 结果：比单线程慢（因为有 GIL 争抢和线程切换开销）

# 实验 3：多线程 IO 密集型
start = time.time()
threads = [threading.Thread(target=io_bound) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"4线程 IO: {time.time() - start:.2f}s")
# 结果：约 1 秒（IO 操作时 GIL 释放，并行等待）

四、GIL 与线程安全的关系

一个常见的误解：有了 GIL，Python 就不需要线程锁了。这是错误的。

GIL 保证的是：同一时刻只有一个线程执行 Python 字节码（解释器层面的原子性）

GIL 不保证的是：

高级操作的原子性（如 count += 1 是多条字节码）
多个操作之间的一致性（如先检查后修改）

python

count = 0

def increment():
    global count
    count += 1  # 这不是原子操作！
    # 字节码：
    # LOAD_GLOBAL count
    # LOAD_CONST 1
    # INPLACE_ADD
    # STORE_GLOBAL count
    # GIL 可能在任意两条字节码之间释放

所以，即使有 GIL，在多线程中操作共享变量仍然需要锁。

五、GIL 释放时机总结

释放时机	说明	示例
IO 操作	遇到 IO 时主动释放	sleep、文件读写、网络请求
时间片到	周期性释放（默认几毫秒）	计算密集型代码也会释放
C 扩展手动释放	C 代码中手动释放	numpy 运算、hashlib 计算
线程结束	线程执行完毕自然释放	线程正常退出

追问延伸：

GIL 是一把什么锁？它保护的是什么？
既然有 GIL，为什么还需要线程锁？
Python 3.2 前后 GIL 的实现有什么区别？
numpy 的多线程为什么能加速？
有没有办法移除 GIL？Gilectomy 项目了解吗？
除了 CPython，其他 Python 实现（如 PyPy）有 GIL 吗？
GIL 和 Java 的 synchronized 有什么区别？

Q11: 什么是线程池？为什么要用线程池？「🟡 中级」

考察点：考察对线程池原理和设计思想的理解，是否清楚线程池的核心参数和使用场景，筛选只会简单使用而不理解原理的候选人。

参考答案：

一、什么是线程池

线程池是一种池化技术，预先创建一批线程，放在一个"池子"里，当有任务需要执行时，从池子中取一个空闲线程来执行任务，任务执行完后线程不销毁，而是放回池中等待下一个任务。

┌─────────────────────────────────┐
│           任务队列               │
│  [任务1][任务2][任务3][任务4]...  │
└──────────────┬──────────────────┘
               │
    ┌──────────▼──────────┐
    │     线程池            │
    │  ┌──┐ ┌──┐ ┌──┐ ┌──┐ │
    │  │线│ │线│ │线│ │线│ │
    │  │程│ │程│ │程│ │程│ │
    │  │1 │ │2 │ │3 │ │4 │ │
    │  └──┘ └──┘ └──┘ └──┘ │
    └──────────────────────┘

二、为什么要用线程池

如果每次执行任务都创建一个新线程，会有以下问题：

创建销毁开销大：线程的创建和销毁需要操作系统介入（分配栈空间、寄存器上下文等），耗时几毫秒到几十毫秒
资源耗尽：无限制创建线程会消耗大量内存（每个线程栈约 1MB），还会导致频繁的上下文切换
缺乏管理：线程数量失控后，系统整体性能下降

线程池的好处：

好处	说明
降低资源消耗	复用线程，减少线程创建和销毁的开销
提高响应速度	任务来了直接用空闲线程，不用等待线程创建
控制并发数	通过最大线程数限制并发，防止资源耗尽
方便管理	可以统一管理、监控、调优
任务队列缓冲	任务过多时可以排队等待，不会直接拒绝或崩溃

三、线程池的核心参数

以 Python 的 ThreadPoolExecutor 为例，核心参数有：

python

from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor(
    max_workers=10,    # 最大线程数
    thread_name_prefix="worker"  # 线程名前缀（方便调试）
)

Java 的 ThreadPoolExecutor 参数更完整，可以作为参考理解：

参数	说明
`corePoolSize`	核心线程数（即使空闲也保留）
`maximumPoolSize`	最大线程数
`keepAliveTime`	空闲线程存活时间
`workQueue`	任务队列
`threadFactory`	线程工厂
`rejectedExecutionHandler`	拒绝策略

Python 的 ThreadPoolExecutor 相对简化：

没有核心线程和最大线程的区分（都是核心线程，且不会超时销毁）
任务队列是无界的（SimpleQueue，Python 3.8+）
没有显式的拒绝策略（队列无界，任务会一直排队）

四、线程池的工作流程

新任务提交
    │
    ▼
当前线程数 < 核心线程数？
    │        │
    是       否
    │        │
    ▼        ▼
创建新线程  任务加入队列
执行任务       │
               │
           队列满了？
               │    │
               否    是
               │    │
               ▼    ▼
           等待执行  当前线程数 < 最大线程数？
                          │      │
                          是      否
                          │      │
                          ▼      ▼
                    创建非核心线程  执行拒绝策略
                    执行任务

Python 的 ThreadPoolExecutor 简化了这个流程：

线程数达到 max_workers 后，新任务进入队列
队列无界，不会拒绝任务（可能导致内存溢出）
线程一旦创建就不会销毁（直到线程池关闭）

五、线程池大小的设置

线程池大小不是越大越好，需要根据任务类型来设置：

1. CPU 密集型任务

线程数 ≈ CPU 核数（或 CPU 核数 + 1）

线程太多会导致频繁的上下文切换，反而降低性能
Python 中因为 GIL 的存在，CPU 密集型应该用进程池而不是线程池

2. IO 密集型任务

线程数 ≈ 2 * CPU 核数，或者更多

IO 密集型任务大部分时间在等待，线程数可以多一些
但也不是越多越好，线程太多会有内存和切换开销
一般经验：线程数 = CPU核数 * (1 + 等待时间/计算时间)

python

import os

cpu_count = os.cpu_count()  # CPU 核数

# IO 密集型：可以设大一些
io_pool = ThreadPoolExecutor(max_workers=cpu_count * 5)

# CPU 密集型：应该用进程池
from concurrent.futures import ProcessPoolExecutor
cpu_pool = ProcessPoolExecutor(max_workers=cpu_count)

3. 混合任务

如果既有 CPU 计算又有 IO，可以根据实际情况调整，或者分开使用不同的线程池。

六、线程池的使用注意事项

1. 任务队列不能无限增长

Python 的 ThreadPoolExecutor 使用无界队列，如果任务提交速度远大于处理速度，队列会不断增长，导致内存耗尽。

解决方法：使用有界队列 + 自定义拒绝策略（需要自己实现）

2. 异常处理

线程池中的任务如果抛出异常，默认不会打印堆栈，需要通过 result() 获取：

python

future = executor.submit(task)
try:
    future.result()
except Exception as e:
    print(f"Task failed: {e}")

3. 正确关闭线程池

使用 with 语句可以自动关闭线程池：

python

with ThreadPoolExecutor(max_workers=10) as executor:
    executor.submit(task)
# 退出 with 块时，线程池自动关闭（等待所有任务完成）

4. 不要在线程池中提交阻塞主线程的任务

可能导致死锁（线程池的线程都在等待队列中的任务，但队列已满）。

七、扩展：进程池

concurrent.futures.ProcessPoolExecutor 是进程池，原理和线程池类似，但用进程代替线程：

对比	线程池	进程池
适用场景	IO 密集型	CPU 密集型
资源开销	较小	较大
数据共享	共享内存（需锁）	独立内存（需 IPC）
GIL 影响	受 GIL 限制	不受 GIL 限制
任务要求	无特殊要求	函数和参数需可序列化

追问延伸：

线程池的核心参数有哪些？怎么设置？
线程池的工作流程是怎样的？
如何确定线程池的大小？有什么依据？
线程池中的任务抛异常了怎么办？
什么情况下线程池的线程会销毁？
线程池有哪些拒绝策略？
你在项目中用过线程池吗？遇到过什么问题？

Q12: 生产者消费者模式？Python中怎么实现？「🟡 中级」

考察点：考察对经典并发模式的理解和实际编码能力，是否能根据不同场景选择合适的实现方式，筛选只会写简单并发代码而不会设计模式的候选人。

参考答案：

一、什么是生产者消费者模式

生产者消费者模式是并发编程中最经典的设计模式之一：

生产者：负责生产数据（任务），放入队列
消费者：从队列中取出数据（任务）进行处理
队列：作为生产者和消费者之间的缓冲区

┌──────────┐     ┌──────────┐     ┌──────────┐
│ 生产者1  │     │ 生产者2  │     │ 生产者3  │
└─────┬────┘     └────┬─────┘     └────┬─────┘
      │               │                │
      └───────────────┼────────────────┘
                      ▼
              ┌──────────────┐
              │   任务队列    │
              │ [ ][ ][ ][ ] │
              └──────┬───────┘
                     │
      ┌──────────────┼──────────────┐
      ▼              ▼              ▼
┌──────────┐   ┌──────────┐   ┌──────────┐
│ 消费者1  │   │ 消费者2  │   │ 消费者3  │
└──────────┘   └──────────┘   └──────────┘

优点：

解耦：生产者和消费者不需要知道对方的存在，只依赖队列
削峰填谷：生产者速度快时，数据可以在队列中缓存；消费者慢慢处理
并发：生产者和消费者可以并发执行，提高系统吞吐量
可扩展：可以灵活增减生产者和消费者的数量

二、多线程版本：queue.Queue

Python 标准库的 queue.Queue 是线程安全的队列，适合多线程的生产者消费者模式。

python

import threading
import queue
import time
import random

# 创建队列，最大容量 10
q = queue.Queue(maxsize=10)

def producer(name):
    for i in range(5):
        item = f"{name}-product-{i}"
        q.put(item)  # 队列满时自动阻塞
        print(f"Producer {name} produced: {item}")
        time.sleep(random.uniform(0.1, 0.5))

def consumer(name):
    while True:
        item = q.get()  # 队列空时自动阻塞
        if item is None:  # 哨兵值，收到 None 表示结束
            q.task_done()
            break
        print(f"Consumer {name} consumed: {item}")
        time.sleep(random.uniform(0.2, 0.6))
        q.task_done()  # 标记任务完成

# 创建生产者和消费者
producers = [threading.Thread(target=producer, args=(f"P{i}",)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(f"C{i}",)) for i in range(3)]

# 启动
for p in producers:
    p.start()
for c in consumers:
    c.start()

# 等待生产者生产完
for p in producers:
    p.join()

# 发送结束信号（每个消费者一个 None）
for _ in consumers:
    q.put(None)

# 等待所有任务处理完
q.join()

# 等待消费者结束
for c in consumers:
    c.join()

print("All done!")

Queue 的常用方法：

方法	说明
`put(item, block=True, timeout=None)`	放入元素，满则阻塞
`get(block=True, timeout=None)`	取出元素，空则阻塞
`put_nowait(item)`	不阻塞放入，满则抛 Full 异常
`get_nowait()`	不阻塞取出，空则抛 Empty 异常
`task_done()`	标记任务完成
`join()`	阻塞，直到所有任务都被标记完成
`qsize()`	当前队列大小（近似值）
`empty()`	是否为空（近似值）
`full()`	是否为满（近似值）

三、多进程版本：multiprocessing.Queue

多进程场景下使用 multiprocessing.Queue，接口和 queue.Queue 类似。

python

from multiprocessing import Process, Queue
import time
import random

def producer(q, name):
    for i in range(5):
        item = f"{name}-product-{i}"
        q.put(item)
        print(f"Producer {name} produced: {item}")
        time.sleep(random.uniform(0.1, 0.5))

def consumer(q, name):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"Consumer {name} consumed: {item}")
        time.sleep(random.uniform(0.2, 0.6))

if __name__ == "__main__":
    q = Queue(maxsize=10)
    
    producers = [Process(target=producer, args=(q, f"P{i}")) for i in range(2)]
    consumers = [Process(target=consumer, args=(q, f"C{i}")) for i in range(3)]
    
    for p in producers:
        p.start()
    for c in consumers:
        c.start()
    
    for p in producers:
        p.join()
    
    # 发送结束信号
    for _ in consumers:
        q.put(None)
    
    for c in consumers:
        c.join()
    
    print("All done!")

注意：

multiprocessing.Queue 没有 task_done() 和 join() 方法
数据需要经过 pickle 序列化，有性能开销
可以使用 JoinableQueue 来获得 task_done() 和 join() 功能

四、异步版本：asyncio.Queue

异步编程场景下使用 asyncio.Queue。

python

import asyncio
import random

async def producer(q, name):
    for i in range(5):
        item = f"{name}-product-{i}"
        await q.put(item)  # 队列满时挂起协程
        print(f"Producer {name} produced: {item}")
        await asyncio.sleep(random.uniform(0.1, 0.5))

async def consumer(q, name):
    while True:
        item = await q.get()  # 队列空时挂起协程
        if item is None:
            q.task_done()
            break
        print(f"Consumer {name} consumed: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.6))
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=10)
    
    # 创建生产者和消费者任务
    producers = [asyncio.create_task(producer(q, f"P{i}")) for i in range(2)]
    consumers = [asyncio.create_task(consumer(q, f"C{i}")) for i in range(3)]
    
    # 等待生产者完成
    await asyncio.gather(*producers)
    
    # 发送结束信号
    for _ in consumers:
        await q.put(None)
    
    # 等待所有任务处理完
    await q.join()
    
    # 等待消费者结束
    await asyncio.gather(*consumers)
    
    print("All done!")

asyncio.run(main())

五、三种实现对比

对比	多线程 queue.Queue	多进程 multiprocessing.Queue	异步 asyncio.Queue
适用场景	IO 密集型	CPU 密集型	高并发 IO 密集型
线程/进程安全	线程安全	进程安全	协程安全
阻塞方式	阻塞线程	阻塞进程	挂起协程
通信速度	快（共享内存）	较慢（IPC + 序列化）	最快（同一线程）
task_done/join	支持	JoinableQueue 支持	支持
最大并发数	几千	几十到几百	几万到几十万

六、结束信号的处理方式

生产者消费者模式中，如何告诉消费者"没有更多数据了"？

方式一：哨兵值（Sentinel）

用一个特殊值（如 None）表示结束，每个消费者收到一个哨兵值就退出。

python

# 有 N 个消费者，就放 N 个 None
for _ in consumers:
    q.put(None)

优点：简单直观缺点：如果消费者数量不确定，不适用

方式二：join + task_done

生产者完成后，调用 q.join() 等待所有任务处理完，然后通过其他方式（如事件）通知消费者退出。

方式三：事件信号（Event）

用一个 Event 来通知所有消费者停止：

python

stop_event = threading.Event()

def consumer(q):
    while not stop_event.is_set():
        try:
            item = q.get(timeout=0.1)
            # 处理 item
            q.task_done()
        except queue.Empty:
            continue

七、实际应用场景

Web 服务器：请求队列 + 工作线程/进程
消息队列：Kafka、RabbitMQ 本质上就是分布式的生产者消费者
日志系统：业务线程生产日志，专门的线程消费写入文件
爬虫系统：URL 生产者 + 页面下载消费者 + 数据解析消费者
任务处理系统：任务提交者 + Worker 池

追问延伸：

生产者消费者模式的优点是什么？解决了什么问题？
队列满了怎么办？有哪些处理策略？
如何优雅地停止消费者线程？
有多个消费者时，任务是怎么分配的？能指定消费者吗？
你在项目中用过生产者消费者模式吗？具体是什么场景？
什么是背压（Backpressure）？生产者消费者模式中如何处理？

Q13: 什么是死锁？怎么避免？Python中如何检测？「🟡 中级」

考察点：考察对死锁概念的理解、预防策略，以及排查死锁的实战经验，筛选缺乏并发编程实战经验的候选人。

参考答案：

一、什么是死锁

死锁（Deadlock）是指两个或多个线程/进程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。

二、死锁的四个必要条件

死锁的产生必须同时满足以下四个条件（缺一不可）：

条件	说明
互斥条件	资源在同一时间只能被一个线程占有
请求与保持条件	线程已经持有至少一个资源，又请求其他被占有的资源
不剥夺条件	已获得的资源不能被其他线程强行剥夺，只能自己释放
循环等待条件	多个线程之间形成头尾相接的循环等待资源关系

死锁示例：

python

import threading
import time

lock_a = threading.Lock()
lock_b = threading.Lock()

def thread1():
    with lock_a:
        print("Thread 1 got lock A")
        time.sleep(0.1)  # 确保 thread2 拿到 lock B
        print("Thread 1 waiting for lock B")
        with lock_b:  # 等待 lock B
            print("Thread 1 got lock B")

def thread2():
    with lock_b:
        print("Thread 2 got lock B")
        time.sleep(0.1)
        print("Thread 2 waiting for lock A")
        with lock_a:  # 等待 lock A
            print("Thread 2 got lock A")

t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)
t1.start()
t2.start()
t1.join()
t2.join()
# 程序会卡死！

分析：

线程1 持有 lock A，等待 lock B
线程2 持有 lock B，等待 lock A
形成循环等待 → 死锁

三、如何避免死锁

只要打破四个必要条件中的任意一个，就能避免死锁。

1. 打破互斥条件 → 不太可行

互斥是资源本身的属性（如锁就是互斥的），一般无法改变。

2. 打破请求与保持条件 → 一次性申请所有资源

线程在开始执行前，一次性申请所有需要的资源，如果有一个申请不到，就都不申请。

python

# 不好的做法：逐步申请
def bad():
    with lock_a:
        # 做一些事
        with lock_b:
            # 做另一些事

# 好的做法：一次性申请（需要设计"同时获取多把锁"的机制）
# 但 Python 标准库没有直接支持，可以自定义

缺点：资源利用率低，可能导致饥饿。

3. 打破不剥夺条件 → 超时放弃

如果尝试获取锁超时，就放弃并释放已经持有的锁。

python

def thread1():
    while True:
        if lock_a.acquire(timeout=1):
            print("Thread 1 got lock A")
            time.sleep(0.1)
            if lock_b.acquire(timeout=1):
                print("Thread 1 got lock B")
                # 业务逻辑
                lock_b.release()
                lock_a.release()
                break
            else:
                print("Thread 1 timeout, releasing lock A")
                lock_a.release()
                time.sleep(0.1)

Python 的 threading.Lock.acquire(timeout=) 支持超时（Python 3.2+）。

缺点：可能活锁（两个线程都不断重试，还是抢不到）。

4. 打破循环等待条件 → 按固定顺序加锁（最常用）

给所有锁编号，规定线程必须按编号从小到大（或从大到小）的顺序获取锁。

python

# 规定：总是先获取编号小的锁，再获取编号大的锁

def thread1():
    # 先 A 后 B（正确顺序）
    with lock_a:
        print("Thread 1 got lock A")
        time.sleep(0.1)
        with lock_b:
            print("Thread 1 got lock B")

def thread2():
    # 也先 A 后 B（和 thread1 一样的顺序）
    with lock_a:  # 先拿 A，而不是 B
        print("Thread 2 got lock A")
        time.sleep(0.1)
        with lock_b:
            print("Thread 2 got lock B")

这样就不会形成循环等待，因为所有线程都按相同顺序获取锁。

更通用的实现：

python

def acquire_locks(*locks):
    """按 id 排序获取多把锁，避免死锁"""
    # 按锁的 id 排序，保证所有线程按相同顺序获取
    sorted_locks = sorted(locks, key=lambda x: id(x))
    for lock in sorted_locks:
        lock.acquire()

def release_locks(*locks):
    for lock in locks:
        lock.release()

# 使用
def thread1():
    acquire_locks(lock_a, lock_b)
    try:
        # 业务逻辑
        pass
    finally:
        release_locks(lock_a, lock_b)

5. 其他方法

减少锁的使用：尽量用无锁数据结构或消息传递代替锁
减少锁粒度：用细粒度的锁代替粗粒度的锁
使用 tryLock + 回退：尝试获取所有锁，不成功就释放已获取的，稍后重试

四、死锁检测

死锁有时候难以避免，需要能够检测和排查。

1. 现象判断

程序卡住不动，CPU 占用率低（说明线程在等待，而不是在计算）
某些功能突然不工作了

2. 工具检测

方法一：py-spy（推荐）

py-spy 是一个 Python 性能分析工具，可以采样查看每个线程在做什么。

bash

# 安装
pip install py-spy

# 查看 Python 进程的线程调用栈
py-spy dump --pid <进程ID>

如果发现多个线程都停在 acquire 上，且互相等待，很可能是死锁。

方法二：自定义锁包装类

可以自定义一个 Lock 包装类，记录锁的持有者和等待者，用于检测死锁。

python

import threading
import traceback

class DeadlockDetectingLock:
    def __init__(self):
        self._lock = threading.Lock()
        self._owner = None
        self._waiters = []
        self._owner_stack = None
    
    def acquire(self):
        tid = threading.get_ident()
        if not self._lock.acquire(blocking=False):
            # 记录等待信息
            self._waiters.append((tid, traceback.format_stack()))
            self._lock.acquire()
            self._waiters.remove((tid, traceback.format_stack()))
        
        self._owner = tid
        self._owner_stack = traceback.format_stack()
    
    def release(self):
        self._owner = None
        self._owner_stack = None
        self._lock.release()
    
    def __enter__(self):
        self.acquire()
        return self
    
    def __exit__(self, *args):
        self.release()

方法三：sys.settrace

使用 sys.settrace 跟踪函数调用，检测死锁。但这种方法性能开销大，不适合生产环境。

方法四：pyrasite

pyrasite 可以附加到运行中的 Python 进程，注入代码来检查状态。

五、死锁排查步骤

确认死锁：程序卡住，CPU 占用低，相关线程不工作
获取线程栈：用 py-spy 或其他工具 dump 线程调用栈
分析锁等待关系：看哪些线程在等什么锁，谁持有这些锁
找到循环等待：确认是否形成了循环等待链
修复代码：根据原因选择合适的避免策略

六、死锁 vs 活锁 vs 饥饿

概念	说明
死锁	线程都在等待对方释放资源，谁也无法继续
活锁	线程都在运行，但都在不断重试，无法推进（类似两个人在走廊里互相让路但都过不去）
饥饿	某个线程一直得不到所需资源，无法执行（如优先级低的线程一直抢不到锁）

追问延伸：

死锁的四个必要条件是什么？为什么缺一不可？
你遇到过死锁吗？怎么排查和解决的？
按顺序加锁为什么能避免死锁？有什么缺点？
什么是活锁？和死锁有什么区别？
Python 中有什么工具可以检测死锁？
数据库中的死锁和编程语言中的死锁有什么异同？
如何设计一个死锁检测工具？

Q14: 协程和线程的区别？为什么协程更高效？「🟡 中级」

考察点：考察对协程和线程本质区别的理解，是否清楚协程高效的根本原因，筛选只会用语法而不理解原理的候选人。

参考答案：

协程和线程都是并发执行的单元，但在调度方式、切换开销、并发数量等方面有本质区别。

一、核心区别对比

维度	线程（Thread）	协程（Coroutine）
调度者	操作系统内核调度	用户态程序（事件循环）调度
切换开销	大（内核态切换，几微秒）	极小（用户态切换，几纳秒）
切换时机	抢占式（操作系统随时可能切换）	协作式（只有 await 时才切换）
数量上限	几千（受内存和调度开销限制）	几万到几十万（非常轻量）
内存占用	每个线程栈约 1MB	每个协程栈几 KB 到几十 KB
数据共享	共享进程内存（需锁）	共享线程内存（通常不需要锁）
锁	重量级锁（阻塞线程）	轻量级锁（挂起协程）
多核利用	可以（但 Python 受 GIL 限制）	不可以（单线程内）
调试难度	相对简单	较难（调用栈不直观）
阻塞影响	阻塞一个线程，其他线程不受影响	阻塞操作阻塞整个事件循环

二、为什么协程更高效

1. 用户态切换 vs 内核态切换

这是最核心的区别：

线程切换（内核态）：

线程由操作系统调度，切换需要陷入内核态
保存和恢复大量寄存器上下文（通用寄存器、程序计数器、栈指针等）
切换页表、刷新 TLB（Translation Lookaside Buffer）
开销：约 1-5 微秒

用户态 → 内核态 → 保存上下文 → 调度 → 恢复上下文 → 用户态

协程切换（用户态）：

协程由程序自己调度，不需要内核参与
只需要保存和恢复少量寄存器（如程序计数器、栈指针）
本质上就是函数调用级别的切换
开销：约几十纳秒（比线程快 100 倍以上）

保存少量寄存器 → 切换栈 → 恢复寄存器

2. 协作式调度 vs 抢占式调度

线程（抢占式）：

操作系统可以在任何时候抢占当前线程，切换到另一个线程
优点：不会有某个线程独占 CPU
缺点：切换时机不可控，可能在不合适的时候切换；需要更多同步机制

协程（协作式）：

只有在 await 点才会切换，切换时机是可控的
优点：切换时机明确，减少不必要的切换；很多时候不需要锁
缺点：一个协程如果不主动让出 CPU，会一直占着（阻塞操作会拖垮整个事件循环）

3. 内存占用差异

线程：每个线程的栈空间通常是 1MB（Linux 默认），几千个线程就占几个 GB
协程：每个协程的栈非常小，初始可能只有几 KB，可以动态增长
10 万个线程需要约 100GB 内存（不可能），10 万个协程只需要几百 MB

4. 调度开销

线程：线程太多时，操作系统的调度器负担很重，频繁的上下文切换消耗大量 CPU
协程：协程调度是用户态的，切换开销极小，几十万协程也能轻松调度

三、代码层面的对比

python

# 线程版：1000 个线程
import threading
import time

def worker():
    time.sleep(1)

start = time.time()
threads = [threading.Thread(target=worker) for _ in range(1000)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"1000 threads: {time.time() - start:.2f}s")

python

# 协程版：100000 个协程
import asyncio
import time

async def worker():
    await asyncio.sleep(1)

async def main():
    start = time.time()
    tasks = [asyncio.create_task(worker()) for _ in range(100000)]
    await asyncio.gather(*tasks)
    print(f"100000 coroutines: {time.time() - start:.2f}s")

asyncio.run(main())

协程可以轻松支持 10 万级并发，而线程 1000 个就开始吃力了。

四、协程的缺点

协程不是银弹，也有缺点：

需要异步生态：所有 IO 操作都需要对应的异步库，同步库会阻塞整个事件循环
不能利用多核：协程在单线程内运行，无法利用多核 CPU（需要配合多进程）
阻塞代码是灾难：一个协程中的阻塞操作会导致所有协程都卡住
调试困难：异常堆栈不直观，不像多线程那样容易调试
心智负担：需要理解事件循环、async/await、Future 等概念

五、适用场景对比

场景	推荐方案	原因
高并发 IO 密集型（Web 服务、爬虫）	协程	并发量高，切换开销小
CPU 密集型	多进程	需要利用多核
混合任务（IO + CPU）	多进程 + 协程	进程利用多核，协程处理 IO
简单的并发任务	多线程	编程简单，生态成熟
需要高稳定性	多进程	进程隔离，互不影响

六、总结

协程之所以高效，根本原因是：

用户态切换：不需要内核参与，切换开销极小
协作式调度：切换时机可控，减少不必要的切换
轻量内存：每个协程占用内存极小，可以创建大量协程

但协程也有局限性，不能完全替代线程和进程，需要根据场景选择合适的并发模型。

追问延伸：

协程切换具体比线程切换快在哪里？能量化吗？
为什么协程是单线程的还能处理高并发？
协程和 goroutine（Go）有什么区别？
有了协程，线程还有存在的必要吗？
什么情况下用协程反而不如多线程？
你在项目中是做技术选型的？什么时候用协程，什么时候用线程？

Q1: Python中实现并发的方式有哪些？各有什么优缺点？ 「🟡 中级」 ​

1. 多线程（threading） ​

2. 多进程（multiprocessing） ​

3. 协程（asyncio） ​

对比表 ​

选型建议 ​

Q2: 多线程和多进程的区别？怎么选择？ 「🟡 中级」 ​

核心区别 ​

深入理解 ​

1. 内存共享的利与弊 ​

2. GIL 的影响 ​

3. 稳定性差异 ​

选型原则 ​

混合使用 ​

Q3: threading模块的常用方法？线程同步方式有哪些？ 「🟢 校招/初级」 ​

一、Thread 类的常用方法 ​

守护线程（daemon） ​

二、线程同步方式 ​

1. Lock（互斥锁） ​

2. RLock（可重入锁） ​

3. Condition（条件变量） ​

4. Semaphore（信号量） ​

5. Event（事件） ​

6. Barrier（屏障） ​

同步原语对比 ​

Q4: 进程间通信（IPC）的方式有哪些？Python中怎么实现？ 「🟡 中级」 ​

Python 中常见的 IPC 方式 ​

1. Queue（队列） ​

2. Pipe（管道） ​

3. 共享内存（Shared Memory） ​

4. Manager（管理器） ​

5. 其他同步原语 ​

6. Socket / RPC（跨机器） ​

各种 IPC 方式对比 ​

选型建议 ​

Q5: concurrent.futures 模块是什么？ThreadPoolExecutor 和 ProcessPoolExecutor？ 「🟡 中级」 ​

核心概念 ​

1. Executor（执行器） ​

2. Future（未来对象） ​

基本用法 ​

ThreadPoolExecutor 示例 ​

ProcessPoolExecutor 示例 ​

重要 API 详解 ​

1. submit() vs map() ​

2. as_completed() — 按完成顺序获取结果 ​

3. wait() — 等待任务完成 ​

4. Future 对象的方法 ​

线程池/进程池的优势 ​

注意事项 ​

ProcessPoolExecutor 的限制 ​

线程池大小的选择 ​

对比：底层 API vs concurrent.futures ​

Q6: asyncio 的事件循环原理？核心概念有哪些？ 「🔴 高级」 ​

一、事件循环的工作原理 ​

二、核心概念 ​

1. 协程（Coroutine） ​

2. Task（任务） ​

3. Future（期程） ​

4. 事件循环（Event Loop） ​

三、事件循环的执行流程 ​

四、底层实现：selectors 模块 ​

五、事件循环的类型 ​

六、为什么是单线程？ ​

Q7: async/await 的底层原理是什么？ 「🔴 高级」 ​

一、从生成器到协程 ​

1. 生成器基础 ​

2. yield from：生成器的委托 ​

3. 基于生成器的协程 ​

二、async/await 的本质 ​

1. async def：定义协程函数 ​

2. await：挂起和恢复 ​

三、协程的状态 ​

四、事件循环如何驱动协程 ​

五、Task 的作用 ​

六、awaitable 对象 ​

七、与生成器的区别 ​

八、总结 ​

Q8: 异步编程中有哪些常见的陷阱？ 「🔴 高级」 ​

1. 阻塞调用阻塞整个事件循环 ​

2. 未 await 的协程不会执行 ​

Q1: Python中实现并发的方式有哪些？各有什么优缺点？「🟡 中级」

1. 多线程（threading）

2. 多进程（multiprocessing）

3. 协程（asyncio）

对比表

选型建议

Q2: 多线程和多进程的区别？怎么选择？「🟡 中级」

核心区别

深入理解

1. 内存共享的利与弊

2. GIL 的影响

3. 稳定性差异

选型原则

混合使用

Q3: threading模块的常用方法？线程同步方式有哪些？「🟢 校招/初级」

一、Thread 类的常用方法

守护线程（daemon）

二、线程同步方式

1. Lock（互斥锁）

2. RLock（可重入锁）

3. Condition（条件变量）

4. Semaphore（信号量）

5. Event（事件）

6. Barrier（屏障）

同步原语对比

Q4: 进程间通信（IPC）的方式有哪些？Python中怎么实现？「🟡 中级」

Python 中常见的 IPC 方式

1. Queue（队列）

2. Pipe（管道）

3. 共享内存（Shared Memory）

4. Manager（管理器）

5. 其他同步原语

6. Socket / RPC（跨机器）

各种 IPC 方式对比

选型建议

Q5: concurrent.futures 模块是什么？ThreadPoolExecutor 和 ProcessPoolExecutor？「🟡 中级」

核心概念

1. Executor（执行器）

2. Future（未来对象）

基本用法

ThreadPoolExecutor 示例

ProcessPoolExecutor 示例

重要 API 详解

1. submit() vs map()

2. as_completed() — 按完成顺序获取结果

3. wait() — 等待任务完成

4. Future 对象的方法

线程池/进程池的优势

注意事项

ProcessPoolExecutor 的限制

线程池大小的选择

对比：底层 API vs concurrent.futures

Q6: asyncio 的事件循环原理？核心概念有哪些？「🔴 高级」

一、事件循环的工作原理

二、核心概念

1. 协程（Coroutine）

2. Task（任务）

3. Future（期程）

4. 事件循环（Event Loop）

三、事件循环的执行流程

四、底层实现：selectors 模块

五、事件循环的类型

六、为什么是单线程？

Q7: async/await 的底层原理是什么？「🔴 高级」

一、从生成器到协程

1. 生成器基础

2. yield from：生成器的委托

3. 基于生成器的协程

二、async/await 的本质

1. async def：定义协程函数

2. await：挂起和恢复

三、协程的状态

四、事件循环如何驱动协程

五、Task 的作用

六、awaitable 对象

七、与生成器的区别

八、总结

Q8: 异步编程中有哪些常见的陷阱？「🔴 高级」

1. 阻塞调用阻塞整个事件循环

2. 未 await 的协程不会执行