Python 编程基础与工程实践（八）：线程、进程、asyncio 与 GIL

并发问题最容易被一句“Python 有 GIL”讲坏：有人因此认为线程毫无用处，也有人误以为 GIL 会自动保证数据安全。Python 3.14 又让情况多了一层——自由线程构建已经获得官方支持，但仍不是默认构建。

本文以 CPython 3.14 为基线，先按任务性质选择工具，再解释线程、进程、asyncio、取消和 GIL 的准确边界。

1. 并发、并行和异步不是同义词

并发（Concurrency）：多个任务的生命周期重叠，可以交替推进；
并行（Parallelism）：多个任务在同一时刻由不同计算资源执行；
异步（Asynchrony）：调用方发起操作后不必阻塞等待，以事件或可等待对象继续协调。

选择方案先看瓶颈：

任务	常见首选	原因
阻塞式文件、网络、设备 I/O	`threading` / `ThreadPoolExecutor`	能复用同步库，等待期间可推进其他线程
高并发异步网络 I/O	`asyncio`	单线程事件循环可管理大量协作式任务
纯 Python CPU 密集计算	`multiprocessing` / `ProcessPoolExecutor`	默认 CPython 中绕开进程内 GIL，利用多核
可隔离、可序列化的 CPU 任务	`InterpreterPoolExecutor`	每个工作线程使用独立解释器和独立 GIL，可在单进程内利用多核
支持自由线程的 CPU 工作负载	自由线程 CPython + `threading`，经实测决定	可并行执行 Python 代码，但生态与同步设计需验证

这只是起点。数据复制成本、第三方库是否释放 GIL、部署平台和故障隔离也会改变结论，最终要用真实工作负载测量。

2. 默认 CPython 的 GIL 到底保证什么

全局解释器锁（Global Interpreter Lock，GIL）在默认 CPython 构建中，通常只允许一个线程在同一时刻执行 Python 字节码并操作 Python 对象。线程在阻塞 I/O 时会释放 GIL，部分 C 扩展也会在长时间本地计算时主动释放它。

GIL 不等于：

整个 Python 语句都是原子的；
多步业务操作天然线程安全；
容器复合操作不会竞态；
内存可见性和事务一致性已经得到业务层保证。

下面的“先检查再写入”由多个步骤组成：

1
2
if key not in cache:
    cache[key] = load_value(key)

两个线程可能都发现键不存在，然后重复加载并覆盖。正确方案取决于语义：可以用锁保护复合不变量，也可以使用允许重复计算但结果幂等的设计。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
from threading import Lock


cache: dict[str, str] = {}
cache_lock = Lock()


def load_value(key: str) -> str:
    return key.upper()


def get_or_load(key: str) -> str:
    with cache_lock:
        if key not in cache:
            cache[key] = load_value(key)
        return cache[key]

这个版本在锁内执行加载，能避免重复加载，却也让慢 I/O 阻塞其他键。生产代码可能使用每键锁、Future 占位或锁外加载后再合并，必须围绕业务不变量权衡。

3. Python 3.14 的自由线程构建

CPython 3.13 开始提供可禁用 GIL 的自由线程构建；Python 3.14 根据 PEP 779 进入“官方支持但可选”的阶段。它并没有取代默认构建，也不意味着所有扩展和程序自动获得线性加速。

可以检查当前运行时：

1
2
3
4
5
6
7
8
import sys
import sysconfig


supports_free_threading = sysconfig.get_config_var("Py_GIL_DISABLED") == 1
gil_enabled = sys._is_gil_enabled()

print(supports_free_threading, gil_enabled)

sys._is_gil_enabled() 在 Python 3.14 可报告当前进程是否启用了 GIL。自由线程构建仍可在运行时启用 GIL；导入未声明兼容自由线程的 C 扩展时，也可能自动重新启用 GIL 并给出警告。

自由线程不取消同步需求，恰恰会让真实并行下的数据竞争更容易暴露。迁移前要检查：

第三方二进制扩展是否提供兼容构建；
共享可变状态是否有明确所有权或锁；
代码是否依赖“看起来原子”的 CPython 偶然行为；
单线程性能、内存开销和扩展兼容性是否可接受；
基准是否覆盖真实请求、数据规模和硬件。

4. 线程：适合阻塞 I/O 和同步库

高层接口 ThreadPoolExecutor 比手工管理线程更适合一批独立任务：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.request import urlopen


def fetch_length(url: str) -> tuple[str, int]:
    with urlopen(url, timeout=5) as response:
        return url, len(response.read())


urls = [
    "https://www.python.org/",
    "https://docs.python.org/3/",
]

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(fetch_length, url) for url in urls]
    for future in as_completed(futures):
        try:
            url, length = future.result()
        except OSError as error:
            print(f"请求失败：{error}")
        else:
            print(url, length)

注意边界：

future.result() 会重新抛出工作线程异常；
超时要传给实际 I/O API，不能只给等待 Future 的调用加超时；
线程池大小不是越大越好，受远端限流、连接数、内存和调度影响；
关闭执行器默认会等待任务结束，进程不会靠“忘记引用”可靠退出。

共享数据使用 Lock、RLock、Condition、Semaphore、Event 或线程安全队列协调。优先通过队列传递数据、缩小共享状态，而不是给每个字段随手加锁。

5. 进程：隔离地址空间并利用多核

ProcessPoolExecutor 把可序列化任务提交给子进程：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from concurrent.futures import ProcessPoolExecutor


def count_primes(limit: int) -> int:
    def is_prime(value: int) -> bool:
        if value < 2:
            return False
        divisor = 2
        while divisor * divisor <= value:
            if value % divisor == 0:
                return False
            divisor += 1
        return True

    return sum(1 for value in range(limit) if is_prime(value))


def main() -> None:
    limits = [50_000, 55_000, 60_000]
    with ProcessPoolExecutor() as executor:
        print(list(executor.map(count_primes, limits)))


if __name__ == "__main__":
    main()

入口保护是跨平台进程代码的必要习惯。提交的函数、参数和返回值通常需要可 pickle；REPL 中定义的局部函数、lambda、打开的连接和锁通常不适合直接传递。

Python 3.14 之后，fork 不再是任何平台的默认启动方式。支持通过 Unix 管道传递文件描述符的 POSIX 平台（如 Linux）默认使用 forkserver，macOS 和 Windows 默认使用 spawn。不要假设所有系统都使用同一种方式，库也不应擅自全局设置启动方式。确需指定时，由应用入口通过上下文显式选择，并验证第三方库兼容性。

进程有序列化、启动和数据复制成本。很小的任务可能比单进程更慢；把工作分成足够大的批次，并避免在进程间来回搬运巨量对象。

6. Python 3.14 的子解释器执行器

Python 3.14 新增 InterpreterPoolExecutor。它是 ThreadPoolExecutor 的子类，但每个工作线程运行在独立解释器中；每个解释器拥有独立 GIL，因此默认构建也能让纯 Python 任务在不同 CPU 核上并行：

1
2
3
4
5
6
7
8
9
from concurrent.futures import InterpreterPoolExecutor


def square(value: int) -> int:
    return value * value


with InterpreterPoolExecutor() as executor:
    print(list(executor.map(square, range(5))))

隔离是它的核心边界：不同解释器拥有各自的模块、sys、内置对象和运行时状态，不能直接共享普通可变对象。执行器会使用 pickle 传递初始化器、任务参数和返回值，因此函数必须可导入、数据必须可序列化，传输成本也不会消失。

与进程池相比，子解释器仍处于同一操作系统进程，故障隔离和扩展兼容性不同；与自由线程相比，它通过隔离减少共享状态，而不是让多个线程共同操作同一套对象。它是 Python 3.14 值得评估的新选择，不是对进程池或自由线程的全面替代。

7. `asyncio`：协作式并发

协程遇到 await 才把控制权交回事件循环：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import asyncio


async def worker(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"{name} done"


async def main() -> None:
    async with asyncio.TaskGroup() as group:
        first = group.create_task(worker("A", 0.2))
        second = group.create_task(worker("B", 0.1))

    print(first.result(), second.result())


if __name__ == "__main__":
    asyncio.run(main())

TaskGroup 提供结构化并发：离开上下文前等待所有任务；某个任务以普通异常失败时，会取消其余任务，并在清理后以异常组传播失败。它比创建一批无人持有、无人等待的后台任务更容易管理生命周期。

async 函数调用只创建协程对象，不会自动执行：

1
coroutine = worker("A", 1.0)

需要 await coroutine、创建 Task，或由 asyncio.run() 驱动顶层协程。

8. 不要阻塞事件循环

下面的同步休眠会阻塞整个事件循环线程：

1
2
3
4
# 错误示例
async def bad() -> None:
    import time
    time.sleep(1)

异步等待应使用：

1
2
async def good() -> None:
    await asyncio.sleep(1)

无法替换的短期阻塞 I/O 可以交给线程：

1
2
3
4
5
6
import asyncio
from pathlib import Path


async def read_text(path: Path) -> str:
    return await asyncio.to_thread(path.read_text, encoding="utf-8")

to_thread() 主要用于不会长时间占用 GIL 的阻塞调用。默认 GIL 构建下，把纯 Python CPU 密集函数扔给线程通常不会获得多核并行。

9. 超时、取消和清理

1
2
3
4
5
6
import asyncio


async def load_with_timeout() -> bytes:
    async with asyncio.timeout(2.0):
        return await load_bytes()

这里的 load_bytes() 代表项目提供的异步 I/O 函数；超时必须包住真正的等待操作。

超时通过取消当前任务实现。协程应使用 try/finally 释放资源，并在捕获 asyncio.CancelledError 后完成必要清理再重新抛出。这里的 acquire() 与 resource 代表项目中的异步资源获取与句柄操作：

1
2
3
4
5
6
7
8
9
async def consumer() -> None:
    resource = await acquire()
    try:
        await resource.run()
    except asyncio.CancelledError:
        await resource.stop()
        raise
    finally:
        await resource.close()

CancelledError 是 BaseException 的直接子类。结构化并发组件依赖取消工作；吞掉取消又不调用相应的取消状态处理，会造成任务组和超时行为异常。

超时也不等于底层操作一定已停止。线程中的阻塞函数不能被 Python 强行安全终止，远端请求也可能已经产生副作用。幂等性、底层超时和补偿策略仍需单独设计。

10. 背压与有界并发

一次创建百万个任务会消耗大量内存并压垮下游。使用有界队列或信号量限制在途任务：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
import asyncio
from collections.abc import Iterable


async def fetch_all(urls: Iterable[str], limit: int = 20) -> list[bytes]:
    semaphore = asyncio.Semaphore(limit)

    async def bounded_fetch(url: str) -> bytes:
        async with semaphore:
            return await fetch(url)

    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(bounded_fetch(url)) for url in urls]

    return [task.result() for task in tasks]

这里的 fetch() 代表异步 HTTP 客户端提供的请求函数。这个版本限制同时执行的 fetch 数量，但仍一次创建所有 Task。输入规模不受控时应进一步使用固定数量消费者和有界 asyncio.Queue，让生产者在队列满时等待，形成真正背压。

11. 如何做选择

可以按以下顺序判断：

先确认瓶颈是 CPU、I/O、锁竞争还是外部限流；
已有同步库且并发量适中，优先线程池；
整条调用链已有异步 API 且需要大量在途 I/O，考虑 asyncio；
默认 CPython 下的纯 Python CPU 密集任务，考虑进程池；
任务可序列化且适合解释器隔离时，评估 3.14 的 InterpreterPoolExecutor；
自由线程构建只有在依赖兼容、共享状态正确且基准受益时才采用；
无论哪种模型，都设置有界并发、超时、取消、错误聚合和关闭流程。

并发模型不是架构身份。一个服务可以在事件循环中管理网络连接，把阻塞调用交给线程，把大块 CPU 计算交给进程，但每跨一层都增加调度、序列化和故障处理成本。

12. 小结

GIL 限制默认 CPython 中 Python 字节码的线程并行，不保证复合业务操作线程安全。
Python 3.14 的自由线程构建已受支持但仍可选；扩展兼容和显式同步仍是前提。
线程适合同步阻塞 I/O，进程适合较粗粒度 CPU 任务，子解释器提供隔离的单进程多核方案，asyncio 适合协作式高并发 I/O。
Python 3.14 不再在任何平台默认使用 fork；Linux 等平台通常使用 forkserver，macOS 和 Windows 使用 spawn，跨平台代码不要依赖隐含默认值。
取消、超时、有界并发和资源关闭是正确性的组成部分，不是上线前再补的优化。

下一篇将把代码组织成可测试、可构建的项目，统一 pyproject.toml、依赖和 CI。