pythonasync函数asyncdef怎么定义

题图：Python 原生协程函数的生命周期与事件循环协作全景

假设你在编写一个网络数据采集脚本或接口服务，随着请求数量增加，传统的 def 函数让整个程序在等待网络响应时陷入卡顿与串行阻塞。为了让程序在等待 I/O 时能够处理其他任务，Python 提供了基于协程的原生异步机制，其核心起点就是使用 async def 关键字。初学者在刚接触时，往往容易直接套用常规函数的调用习惯，结果不仅没能实现并发，还遇到了协程未等待等警告与报错。本文将系统讲解 async def 的语法定义、参数及返回值传递、调用运行的完整步骤与底层运行机制，帮助读者写出规范且高效的 Python 异步函数。

一、async def 的基本定义语法与核心组成

在 Python 3.5 及更高版本中，定义一个原生异步协程函数的标准语法是在普通的 def 声明前加上 async 关键字。它的基本结构如下：

async def 函数名(参数1: 类型, 参数2: 类型 = 默认值) -> 返回类型:
    # 函数体逻辑
    return 返回值

使用 async def 定义的函数被称为原生协程函数（Native Coroutine Function）。它在形式上完全继承了 Python 普通函数的灵活性，同时在执行机制上具备异步特性：

• 参数完全通用：可以像普通函数一样声明位置参数、默认参数、可变参数 *args 和关键字参数 **kwargs。
• 支持类型提示（Type Hints）：完全兼容 PEP 484 规范，可以为形参及返回值标注类型，提高代码的可读性与静态检查体验。
• 支持常规返回值：函数体内可以使用 return 语句返回任何 Python 数据结构（如字典、列表、自定义类实例等）。如果没有显式写 return，函数结束时默认返回 None。

下面是一个带有参数与类型注解的基础异步函数定义示例：

async def calculate_summary(title: str, scores: list[float]) -> dict:
    """计算成绩平均值并返回字典结果的异步函数"""
    if not scores:
        return {"title": title, "average": 0.0, "count": 0}
    
    total = sum(scores)
    avg = total / len(scores)
    return {
        "title": title,
        "average": round(avg, 2),
        "count": len(scores)
    }

图 1：async def 语法结构拆解与各核心组成要素解析

二、直接调用与协程对象：初学者的常见误区

很多初学者在写好 async def 函数后，会习惯性地像普通函数一样直接执行 result = calculate_summary("测试", [90, 85])，但随后会发现函数并没有正常计算，或者控制台输出了异常的信息。

我们通过一段具体的演示代码来观察这个现象：

import asyncio

async def demo_task(name: str):
    print(f"任务 [{name}] 开始执行内部代码")
    return f"任务 [{name}] 完成"

# 错误做法：像普通函数一样直接调用
result = demo_task("A")

print("返回值类型:", type(result))
print("返回值内容:", result)

运行上述脚本后，控制台得到的实际输出如下（示例）：

返回值类型: <class 'coroutine'>
返回值内容: <coroutine object demo_task at 0x104b2a8d0>
sys:1: RuntimeWarning: coroutine 'demo_task' was never awaited

从输出中可以清晰地看到两点关键事实：

1. 函数体内的 print("任务 [A] 开始执行内部代码") 根本没有被打印出来，说明函数体逻辑压根没有被执行。
2. 直接调用 demo_task("A") 返回的不是最终的字符串结果，而是一个 coroutine object（协程对象）。随后 Python 解释器在程序退出时抛出了 RuntimeWarning: coroutine 'demo_task' was never awaited 警告。

三、如何正确执行与驱动 async def 函数

要让 async def 定义的异步函数真正运转并拿到返回结果，在 Python 中主要有两种标准操作路径：在程序的最外层作为主入口驱动，或者在另一个异步函数内部协作等待。

图 2：async def 函数从声明到执行的规范四步法流程

1. 顶层入口驱动：使用 asyncio.run()

在 Python 3.7+ 中，运行一个顶层异步函数的官方推荐方法是 asyncio.run()。它会自动完成一系列复杂的基础设施工作：创建全新的事件循环、将传入的顶层协程注册为主任务执行、运行直至该协程返回结果，最后自动关闭事件循环并回收线程与网络连接等底层资源。

import asyncio

async def main():
    # 在协程内部等待并获取子任务结果
    res = await demo_task("主流程")
    print("拿到执行结果:", res)

if __name__ == "__main__":
    # 正确的顶层启动方式
    asyncio.run(main())

执行该脚本后，控制台将正常输出：

任务 [主流程] 开始执行内部代码
拿到执行结果: 任务 [主流程] 完成

2. 协程内部协作：使用 await 挂起与获取结果

当已经处于一个 async def 函数内部时，驱动另一个异步函数不需要也不能再次调用 asyncio.run()，而是必须使用 await 关键字。

await 表达式后只能跟一个可等待对象（Awaitable），最典型的就是另一个 async def 函数调用所产生的协程对象。它的运行机制是：暂停当前协程的执行，将 CPU 控制权交还给事件循环；等到被等待的异步任务完成后，事件循环唤醒当前协程，并将被等待任务的返回值赋给左侧的变量。

import asyncio

async def fetch_data_from_db(query_id: int) -> str:
    print(f"开始查询数据库 ID={query_id}...")
    # 模拟异步网络 I/O 耗时 1 秒，不能使用 time.sleep
    await asyncio.sleep(1)
    return f"数据记录 #{query_id}"

async def process_pipeline():
    print("Pipeline 启动")
    # 通过 await 驱动 fetch_data_from_db 并获取其 return 值
    record = await fetch_data_from_db(101)
    print(f"处理完成，数据内容为: {record}")

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

3. 多任务并发调度：使用 asyncio.gather()

异步函数最大的优势在于并发处理多个独立 I/O 操作。如果你有多个 async def 定义的任务需要同时执行，可以使用 asyncio.gather() 将它们并发分发，而不需要串行依次等待：

import asyncio
import time

async def download_file(file_id: int):
    print(f"开始下载文件 {file_id}")
    await asyncio.sleep(1)  # 模拟 1 秒的网络下载延时
    print(f"文件 {file_id} 下载完成")
    return f"content_{file_id}"

async def batch_runner():
    start_time = time.time()
    # 同时发起 3 个下载任务并发执行
    results = await asyncio.gather(
        download_file(1),
        download_file(2),
        download_file(3)
    )
    duration = time.time() - start_time
    print(f"全部下载结果: {results}")
    print(f"总耗时: {duration:.2f} 秒")

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

如果是传统的同步代码，顺序执行 3 个耗时 1 秒的操作需要累计 3 秒；而通过上面的 async def 和 asyncio.gather()，3 个任务在事件循环中并发推进，总耗时仅需约 1 秒。

四、async def 与普通 def 的本质机制对比

为了在工程开发中准确决定何时使用 async def、何时使用普通 def，我们需要理解两者在调用栈、调度权和时间线模型上的差异。

图 3：同步阻塞模型与异步协程并发调度的时间轴对比

五、编写 async def 必须牢记的注意事项与避坑指南

在编写异步函数时，代码编写逻辑虽然与常规 Python 代码很像，但由于其异步事件驱动特性，有几项关键规则必须严格遵守：

1. 绝不能在 async def 内部使用同步阻塞库

这是编写异步代码最容易犯的错误。由于 asyncio 默认是在单个线程内运行事件循环，如果在一个 async def 函数里调用了 time.sleep(5) 或同步网络库 requests.get()，整个线程都会被操作系统冻结，导致事件循环无法处理任何其他协程任务。

2. 语法规则约束：await 不能脱离 async def

await 是针对协程的挂起语法，Python 语法解析器规定 await 只能写在用 async def 定义的函数体之内。如果试图在普通的 def 函数中直接编写 await，代码在编译解析阶段就会抛出报错：

SyntaxError: 'await' outside async function

3. 正确区分“函数对象”与“协程对象”

在传参和调度时，务必注意函数名后是否带有括号：

• my_coro_func：是函数本身（可调用对象），通常用于注册回调或工厂引用。
• my_coro_func()：才是通过调用生成的协程对象，用于传递给 await、asyncio.run() 或 asyncio.create_task()。

4. 异常捕获机制与同步代码完全一致

很多开发者担心异步代码的异常难以捕获。事实上，在 async def 函数内部，异常处理机制与普通函数完全通用。你可以使用标准的 try...except...finally 块精确拦截被 await 的任务抛出的异常：

async def safe_fetch(url: str):
    try:
        response = await simulated_network_fetch(url)
        return response
    except TimeoutError:
        print(f"请求地址超时: {url}，执行降级逻辑")
        return "default_cache"
    finally:
        print("清理连接资源")

图 4：async def 核心规则、调用驱动与避坑排查全景脑图

综上所述，定义 Python 异步函数只需要在函数声明前加上 async def 关键字，但其核心价值在于它与 await 关键字及 asyncio 事件循环的深度协作。掌握从语法定义到协程驱动的这套标准规范，就能在编写网络请求、API 服务和并发处理时充分发挥 Python 的异步性能优势。