Python中如何实现协程？async/await语法详解-Python教程-PHP中文网

Python中如何实现协程？async/await语法详解

看不見的法師

发布： 2025-07-04 20:24:02

原创

875人浏览过

python中协程通过async/await实现高效并发，适合i/o密集型任务。1. 使用async def定义协程函数，调用后返回协程对象；2. 用await等待其他协程完成，但只能在async函数内使用；3. 通过asyncio.run启动事件循环执行协程；4. 用asyncio.create_task将多个协程封装为任务实现并发；5. 异步库如aiohttp可提升网络请求效率；6. 注意避免混用阻塞代码并正确管理事件循环。

Python中如何实现协程？async/await语法详解

在 Python 中，协程（Coroutine）是一种高效的并发编程方式，尤其适合处理 I/O 密集型任务。实现协程最常见的方式是使用 async/await 语法，这是从 Python 3.5 开始引入的特性。

协程的基本概念：不是线程也不是生成器

协程听起来像线程，但它不是操作系统层面的线程，而是用户态的轻量级“线程”。它可以在执行过程中暂停（yield）并在合适的时候恢复。Python 的协程基于生成器发展而来，但和普通生成器不同的是，协程可以接收外部传入的数据，并且支持异步操作。

简单来说，一个用 async def 定义的函数就是一个协程函数，调用它会返回一个协程对象，而不是立即执行。

立即学习“Python免费学习笔记（深入）”；

async def hello():
    print("Hello")

coro = hello()  # 这只是一个协程对象，还没运行

登录后复制

要真正运行这个协程，必须把它放进事件循环里执行。

async/await 的基本用法：定义和调用协程

async 和 await 是协程的核心关键词：

async def 定义一个协程函数；
await 用于等待另一个协程完成。

举个例子：

import asyncio

async def say_after(delay, text):
    await asyncio.sleep(delay)
    print(text)

async def main():
    print("Start")
    await say_after(1, "Hello")
    await say_after(2, "World")
    print("Done")

asyncio.run(main())

登录后复制

这段代码中：

say_after 是一个协程函数；
main() 函数中用 await 调用了两个协程；
使用 asyncio.run() 启动事件循环并运行主协程。

输出顺序是：

Start
Hello
World
Done

登录后复制

注意：await 只能在 async def 函数内部使用。

协程并发执行：用 asyncio.create_task 创建任务

上面的例子是串行执行两个协程，如果想并发执行怎么办？可以用 asyncio.create_task() 把多个协程包装成任务加入事件循环。

修改上面的例子：

async def main():
    task1 = asyncio.create_task(say_after(1, "Hello"))
    task2 = asyncio.create_task(say_after(2, "World"))

    print("Started")
    await task1
    await task2
    print("Finished")

登录后复制

这次输出会是：

Started
Hello
World
Finished

登录后复制

可以看到，“Started”立刻打印出来，两个协程同时运行，整体耗时约 2 秒，而不是串行的 3 秒。

关键点：

多个协程封装成任务后可并发执行；
通过 await task 来确保主协程等待所有子任务完成；
不需要手动管理线程或进程，事件循环帮你调度。

实际应用场景：网络请求、文件读写等 I/O 操作

协程最适合处理 I/O 密集型任务，比如发起多个 HTTP 请求、读写文件、数据库查询等。

比如用 aiohttp 发起异步 HTTP 请求：

import aiohttp
import asyncio

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

async def main():
    urls = [
        'https://example.com',
        'https://example.org',
        'https://example.net'
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(len(result))

asyncio.run(main())

登录后复制

在这个例子里：