Python 异步编程及多线程学习总结

Wangzy

发布于 2026-06-22 18:20:33

1400

01

— Python解释器及GIL（全局解释器锁）介绍

1、GIL（全局解释器锁）

在介绍Python #异步编程及 #多线程之前，有必要先介绍下GIL（全局解释器锁）。

GIL 是 CPython 解释器（Python 最广泛使用的官方实现）中使用的一种机制。它本质上是一个互斥锁（mutex），它要求在任何时刻，只有一个线程可以执行 Python 字节码。这意味着即使在多核 CPU 上，使用 CPython 的多线程程序，在执行纯 Python 代码时，也无法实现真正的并行计算（多个线程同时利用多个 CPU 核心进行计算）。

GIL 的存在主要是为了解决 CPython 解释器内部的内存管理问题（尤其是引用计数机制）在多线程环境下的线程安全问题。

GIL对CPU密集型的任务会有较大影响（例如科学计算、视频编码、大规模数据处理），你试图使用多线程来加速，性能提升会非常有限，甚至可能比单线程更慢（因为线程切换也有开销）。

如果你的程序主要涉及等待 I/O 操作（如网络请求、磁盘读写、数据库查询），GIL 的影响相对较小。原因是可以通过多线程或异步编程的协程方式来释放GIL，让其他线程接管GIL并执行Python代码。

2、GIL是在Python的CPython 解释器中使用的机制，所以Python都有哪些解释器呢？

a、CPython解释器

Flask、Django、Tornado 等流行的 Python Web 框架都是在 CPython 解释器上运行的。大部分的机器学习框架也都是用的CPython解释器

CPython 是 Python 语言的官方参考实现，由 Python 软件基金会维护。它是目前最广泛使用、兼容性最好、生态系统最丰富的 Python 解释器。

绝大多数 Python 库（包括 Web 框架、数据库驱动、科学计算库等）都是首要为 CPython 开发和测试的。它们的文档、社区支持和常见问题解答都基于 CPython 环境。

b、PyPy解释器

定位：使用JIT（即时编译）加速的Python实现。

优势：纯Python代码运行速度可提升数倍（对计算密集型循环有效）。

局限性：对C扩展支持较弱（如NumPy、TensorFlow兼容性不佳），机器学习中较少使用。

适用场景：纯Python算法加速（非主流库依赖场景）。

c、Jython解释器

运行在 JVM 上

d、IronPython

运行在 .NET CLR 上

—

进程、线程、协程概念介绍

1、进程（Process）

在这就不科普操作系统os中进程的相关概念，主要讲下进程在Python异步框架中所处的角色。

Pyhont Web框架通常都是单进程运行的，但是在生产环境实际部署中，会通过多种方式实现多进程处理以支持高并发。具体有如下几种：

a、WSGI服务器多进程

使用Gunicorn/Uvicorn启动多个Worker进程（gunicorn -w 4 myapp:app），如Flask/Django等同步框架。

b、ASGI服务器多进程

Uvicorn + Gunicorn运行ASGI应用（gunicorn -k uvicorn.workers.UvicornWorker），如FastAPI/Starlette等异步框架。

c、uWSGI多进程多线程

配置uWSGI的processes和threads参数。

d、编程式多进程启动

使用multiprocessing.Process启动多个实例。

多进程部署的形式，可以突破CPU限制和GIL限制。一个主进程管理多个工作进程。工作进程是独立的Python解释器实例，运行着各自的事件循环和协程。可以根据服务器CPU核心数启动相应数量的工作进程，每个CPU核心一个进程，互相独立。

多进程的缺点是：创建进程、进程间通信（内存隔离，需要通过IPC）开销远大于线程或协程。

2、线程（Thread）

这一节也是主要介绍在于异步框架中所处的角色。

事件循环主线程：是主线程，运行着唯一（或主要）的事件循环。这个进程可以调度所有协程去执行异步任务。

工作线程池：为了执行阻塞的、非异步的或 CPU 密集型的操作。异步框架（如 asyncio 本身）通常会提供一个线程池执行器（ThreadPoolExecutor）。当协程需要执行这类阻塞操作时，它会将阻塞函数提交给线程池，挂起自身（await 线程池返回一个 Future），线程池中的某个工作线程去执行该阻塞函数，函数执行完毕后，结果通过 Future 通知事件循环，事件循环唤醒挂起的协程，并传递结果。

使用线程池是为了兼容阻塞代码，避免它们阻塞事件循环线程。工作线程本身是受操作系统调度的，会阻塞和切换，有开销。

3、协程（Coroutine）

协程是一种用户态的轻量级线程。它由用户代码控制调度（通过 async/await 语法），而不是由操作系统内核调度。协程拥有自己的栈和上下文，但共享进程的堆内存。

创建、销毁、切换协程的开销远低于操作系统线程（可能是 KB 级 vs MB 级，纳秒级 vs 微秒级）。

协程主动 yield（通过 await）让出控制权给事件循环，而不是被操作系统强制抢占。这避免了复杂的锁机制（在单事件循环线程内），但也要求协程不能长时间阻塞。

注意：一个协程如果不主动 await，它会一直运行直到完成或遇到下一个 await。如果它执行了 CPU 密集型计算，会阻塞整个事件循环线程！

在异步框架中，处理一个请求的逻辑通常被写成一个或多个协程。例如，一个 FastAPI 的路径操作函数用 async def 定义，它就是一个协程。当请求到来时，框架创建一个协程实例来处理它。

—

Python异步框架及异步编程

1、Python 异步编程

async/await语法是异步编程概念发展到一定阶段后被诸多编程语言采用的一种语法。

Python语言从3.4开始引入asyncio库，从3.5开始显式地引入async/await关键字，并持续改进发展至今。

asyncio库当前是python标准库的一部分，默认内置。使用async/await语法时往往需要和asyncio库中的一些功能配合使用，一般会import asyncio库。

如下为使用asyncio库实现的一个异步编码例子。

在异步编程中，我们调用一个异步函数时，如果没有使用await关键字，得到的会是一个协程对象，而不是函数的返回值。得到这个协程对象后程序会直接向后运行，而不是等待这个函数运行完后给予我们返回值。如果我们需要告诉程序，我们需要得到异步函数执行的结果后才应该往后执行，那么就可以在调用异步函数时，在前面加上await关键字。这样程序会等待该异步函数执行完后才会往后执行，且能够正常地得到异步函数的返回值。加上await关键字是在告诉程序：我需要等待这个异步操作才应该往后执行代码逻辑，等待该异步操作时你可以找点别的工作做。

import asyncio

async def heavy_task(task_name,n):
   print(f"Task begin:{task_name}")
   await asyncio.sleep(n)
   print(f"After {n} seconds, Task {task_name} Done!")

async def main():
   tasks = [heavy_task(f"task{i}", 4 - i) for i in range(1, 4)]
   await asyncio.gather(*tasks)

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

在执行一些高级方法，如rununtilcomplete()时，我们首先需要得到asyncio运行的事件循环。

我们可以asyncio.neweventloop()来创建一个事件循环；也可以使用asyncio.geteventloop()来获取当前运行的事件循环，若当前没有正在运行的事件循环，则创建一个。如下是具体代码样例：

import asyncio

async def heavy_task(task_name,n):
   print(f"Task begin:{task_name}")
   await asyncio.sleep(n)
   print(f"After {n} seconds, Task {task_name} Done!")

if __name__=='__main__':
   tasks=[heavy_task(f"task{i}",4-i) for i in range(1,4)]
   task_list=asyncio.gather(*tasks)
   loop=asyncio.get_event_loop()
   loop.run_until_complete(task_list)

2、异步框架的心脏：事件循环

上一节中有提到异步编程高级函数需要先引入事件循环。

Fastapi、Tornado等异步框架，本身就是一个单线程的事件循环，事件循环是异步框架的心脏。

事件循环是一个无限循环的核心调度器。它持续运行，负责监听和分发事件（Event）。

如下详细介绍下事件循环：

a、事件来源：

I/O 操作就绪：网络 socket 有数据到达、连接建立完成、文件读写完成、数据库查询结果返回等。

定时器到期：设定的延时任务或周期性任务到时间了。

其他信号：系统信号、线程间通信等。

b、工作原理：

轮询：使用高效的 I/O 多路复用机制（如 Linux 的 epoll, macOS 的 kqueue, Windows 的 IOCP）来监控大量的文件描述符（主要是网络 socket），找出哪些已经就绪（有数据可读、可写或发生错误）。

调度协程：当检测到一个事件就绪（例如，一个 socket 有数据可读），事件循环会找到与该事件关联的挂起的协程（Coroutine），将其唤醒并放入可执行队列。

执行协程：事件循环从可执行队列中取出一个协程（通常是按某种公平策略，如轮询），恢复其执行（执行到下一个 await 点或完成）。

挂起协程：当协程执行遇到 await 一个非立即完成的 I/O 操作（如 await response.read()）时，它会主动让出控制权给事件循环，并将自己与该 I/O 操作关联起来，注册到事件循环的监听列表中。此时该协程状态变为“挂起”（Pending）。

c、关键点：事件循环运行在单线程（或少量线程）中。它高效地管理着成千上万的协程，让它们在 I/O 等待时挂起，在 I/O 就绪时恢复，避免了线程阻塞和频繁的线程上下文切换开销。

3、异步框架的异步编程实践简介（Tornado框架）

在 Tornado 框架中进行异步编程时，遵循最佳实践可显著提升性能、可维护性和可靠性。以下是关键实践总结及示例：

a. 优先使用原生协程（async/await）

替代方案：弃用 @gen.coroutine 和 yield
优势：代码更简洁，兼容 Python 生态（如 asyncio）

`` async def get(self):
``     result = await self.async_task()
``     self.write(result)

b. 避免阻塞事件循环

常见阻塞操作：

CPU 密集型计算
同步 I/O（文件、网络、DB）

解决方案：

异步库：使用 aiofiles（文件）、aiomysql（MySQL）、asyncpg（PostgreSQL）
线程池：对无法异步的操作使用 IOLoop.runinexecutor

`` from concurrent.futures import ThreadPoolExecutor
`` import time
`` 
`` async def get(self):
``     result = await IOLoop.current().run_in_executor(
``         None,  # 使用默认线程池
``         time.sleep, 5  # 阻塞函数
``     )
``     self.write("Done")

c. 并发执行异步任务

并行处理：使用 asyncio.gather 或 tornado.gen.multi

`` async def get(self):
``     # 同时发起两个异步请求
``     res1, res2 = await asyncio.gather(
``         http_client.fetch(url1),
``         http_client.fetch(url2)
``     )
``     self.write(f"Size: {len(res1.body) + len(res2.body)}")

d. 使用异步 HTTP 客户端

内置客户端：tornado.httpclient.AsyncHTTPClient

`` async def get(self):
``     client = AsyncHTTPClient()
``     response = await client.fetch("https://api.example.com")
``     self.write(response.body)

e、常见陷阱

意外阻塞：在协程中调用同步 I/O
未处理异常：协程内异常需显式捕获
资源共享冲突：使用异步锁（tornado.locks.Lock）
事件循环嵌套：避免在 Tornado 中启动新的事件循环

—

Python异步编程应对高并发

1、核心原则：保持事件循环畅通

严禁阻塞操作
任何在事件循环线程中的同步 I/O 或 CPU 密集型任务都会导致所有请求卡顿。
解决方案：

使用 asyncio.tothread() 或 loop.runinexecutor() 将阻塞操作（如文件读写、CPU 计算）移交线程池。

`` # FastAPI 示例：将阻塞函数移交线程池
`` from fastapi import FastAPI
`` import asyncio
`` import time
`` 
`` app = FastAPI()
`` 
`` def blocking_operation():
``     time.sleep(2)  # 模拟阻塞操作
``     return "Done"
`` 
`` @app.get("/")
`` async def endpoint():
``     result = await asyncio.to_thread(blocking_operation)
``     return {"result": result}

2、全链路异步化