深入理解 Python 异步编程：async、await 与同步函数详解 | 极客日志

Python

深入理解 Python 异步编程：async、await 与同步函数详解

详细解析了 Python 异步编程的核心概念，对比了同步与异步函数的区别及性能差异。重点讲解了 async 关键字定义协程函数、await 关键字暂停协程让出控制权以及 asyncio.create_task 实现并发任务调用的机制。通过实际场景如 Web API、文件处理、数据库操作展示了异步应用，并纠正了常见误区如 await 非阻塞、异步非自动并发等。最后提供了最佳实践建议，包括何时使用异步、错误处理、资源管理及性能优化技巧，帮助开发者编写高效可扩展的应用程序。

莫名其妙发布于 2026/3/27更新于 2026/4/186 浏览

引言

在现代 Python 开发中，异步编程已经成为处理 I/O 密集型任务的标准方式。理解 async、await 和同步函数的区别，对于编写高效、可扩展的应用程序至关重要。

本文将通过理论解释、代码示例和实际场景，帮助你全面理解 Python 异步编程的核心概念。

同步函数 vs 异步函数

什么是同步函数？

同步函数是传统的函数调用方式，代码按顺序执行，每一步都必须等待前一步完成。

 time

 ():
    
    ()
    time.sleep()  
    ()
     


result = sync_function()
(result)

import asyncio

async def async_function():
    """异步函数示例"""
    print("开始执行")
    await asyncio.sleep(2)  # 异步等待 2 秒
    print("执行完成")
    return "结果"

# 调用异步函数
async def main():
    result = await async_function()
    print(result)

# 运行异步函数
asyncio.run(main())

import time
import asyncio
import aiohttp

# ========== 同步方式 ==========
def sync_fetch(url):
    """同步获取网页内容"""
    time.sleep(1)  # 模拟网络延迟
    return f"Response from {url}"

def sync_process_100_requests():
    """同步处理 100 个请求"""
    start = time.time()
    results = []
    for i in range(100):
        result = sync_fetch(f"http://example.com/{i}")
        results.append(result)
    end = time.time()
    print(f"同步方式耗时：{end - start:.2f} 秒")
    return results

# ========== 异步方式 ==========
async def async_fetch(session, url):
    """异步获取网页内容"""
    await asyncio.sleep(1)  # 模拟网络延迟
    return f"Response from {url}"

async def async_process_100_requests():
    """异步处理 100 个请求"""
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [async_fetch(session, f"http://example.com/{i}") for i in range(100)]
        results = await asyncio.gather(*tasks)
    end = time.time()
    print(f"异步方式耗时：{end - start:.2f} 秒")
    return results

# 运行对比
if __name__ == "__main__":
    # 同步方式：约 100 秒
    sync_process_100_requests()
    # 异步方式：约 1 秒（并发执行）
    asyncio.run(async_process_100_requests())

async def my_async_function():
    return "Hello"

# 调用异步函数
coro = my_async_function()
print(type(coro))  # <class 'coroutine'>

# ❌ 错误：直接调用不会执行
result = my_async_function()  # 这只是创建了协程对象，没有执行

# ✅ 正确：需要使用 await 或 asyncio.run()
result = await my_async_function()  # 在异步上下文中
# 或
result = asyncio.run(my_async_function())  # 在同步上下文中

async def fetch_data():
    # 可以 await 其他异步函数
    data = await some_async_operation()
    return data

# 方式 1：使用 await（在异步函数中）
async def main():
    result = await func()
    print(result)  # 42

# 方式 2：使用 asyncio.run()
result = asyncio.run(func())
print(result)  # 42

async def func():
    return 42

coro = func()  # 协程对象，尚未执行

# ❌ 错误 1：忘记 await
async def get_data():
    return "data"

def main():
    result = get_data()  # 错误！result 是协程对象，不是字符串
    print(result)  # <coroutine object get_data at 0x...>

# ✅ 正确
async def main():
    result = await get_data()  # 正确
    print(result)  # "data"

# ❌ 错误 2：在同步函数中直接调用异步函数
def sync_func():
    result = await async_func()  # 错误！await 只能在 async 函数中使用

# ✅ 正确
async def async_func():
    result = await async_func()  # 正确

# 或在同步函数中使用 asyncio.run()
def sync_func():
    result = asyncio.run(async_func())  # 正确

async def task1():
    print("Task 1 开始")
    await asyncio.sleep(2)  # 暂停，让出控制权
    print("Task 1 完成")

async def task2():
    print("Task 2 开始")
    await asyncio.sleep(1)  # 暂停，让出控制权
    print("Task 2 完成")

async def main():
    # 并发执行两个任务
    await asyncio.gather(task1(), task2())

asyncio.run(main())
# 输出：
# Task 1 开始
# Task 2 开始
# Task 2 完成 (1 秒后)
# Task 1 完成 (2 秒后)
# 总耗时：约 2 秒（不是 3 秒）

import time
import asyncio

# ========== 同步阻塞 ==========
def sync_blocking():
    print("开始阻塞")
    time.sleep(2)  # 阻塞整个线程
    print("阻塞结束")
    # 在这 2 秒内，整个线程无法做任何其他事情

# ========== 异步等待 ==========
async def async_waiting():
    print("开始等待")
    await asyncio.sleep(2)  # 暂停协程，但不阻塞线程
    print("等待结束")
    # 在这 2 秒内，其他协程可以运行

# 对比演示
async def demo():
    print("=== 同步阻塞演示 ===")
    start = time.time()
    sync_blocking()
    sync_blocking()
    print(f"耗时：{time.time()- start:.2f} 秒\n")  # 约 4 秒

    print("=== 异步等待演示 ===")
    start = time.time()
    await asyncio.gather(async_waiting(), async_waiting())
    print(f"耗时：{time.time()- start:.2f} 秒")  # 约 2 秒

asyncio.run(demo())

# 1. 其他协程对象
async def func1():
    return "result"

async def func2():
    result = await func1()  # ✅ 可以
    return result

# 2. 可等待对象（Awaitable）
# - 协程（Coroutine）
# - 任务（Task）
# - Future 对象

# 3. 异步上下文管理器
async def example():
    async with aiofiles.open('file.txt') as f:
        content = await f.read()  # ✅ 可以

# 4. 异步迭代器
async def example():
    async for line in async_file_reader():
        print(line)  # ✅ 可以

import asyncio

async def my_coroutine():
    await asyncio.sleep(1)
    return "完成"

# create_task 将协程包装成 Task
task = asyncio.create_task(my_coroutine())
# Task 立即开始执行（不等待完成）

import asyncio
import time

async def task(name: str, duration: int):
    print(f"[{time.time():.2f}] 任务 {name} 开始")
    await asyncio.sleep(duration)
    print(f"[{time.time():.2f}] 任务 {name} 完成")
    return f"任务 {name} 的结果"

async def main():
    start = time.time()
    # 创建多个 Task，它们会并发执行
    task1 = asyncio.create_task(task("A", 2))
    task2 = asyncio.create_task(task("B", 1))
    task3 = asyncio.create_task(task("C", 3))
    
    # 等待所有任务完成
    results = await asyncio.gather(task1, task2, task3)
    end = time.time()
    print(f"\n所有任务完成，耗时：{end - start:.2f} 秒")
    print(f"结果：{results}")

asyncio.run(main())
# 输出示例：
# [12345.67] 任务 A 开始
# [12345.67] 任务 B 开始
# [12345.67] 任务 C 开始
# [12346.67] 任务 B 完成
# [12347.67] 任务 A 完成
# [12348.67] 任务 C 完成
#
# 所有任务完成，耗时：3.00 秒（不是 6 秒！）
# 结果：['任务 A 的结果', '任务 B 的结果', '任务 C 的结果']

import asyncio

async def slow_task(name: str):
    print(f"{name} 开始")
    await asyncio.sleep(2)
    print(f"{name} 完成")
    return f"{name} 的结果"

# ========== 方式 1：顺序执行（使用 await）==========
async def sequential():
    print("=== 顺序执行 ===")
    result1 = await slow_task("任务 1")  # 等待完成
    result2 = await slow_task("任务 2")  # 等待完成
    result3 = await slow_task("任务 3")  # 等待完成
    # 总耗时：约 6 秒
    return [result1, result2, result3]

# ========== 方式 2：并发执行（使用 create_task）==========
async def concurrent():
    print("=== 并发执行 ===")
    task1 = asyncio.create_task(slow_task("任务 1"))  # 立即开始
    task2 = asyncio.create_task(slow_task("任务 2"))  # 立即开始
    task3 = asyncio.create_task(slow_task("任务 3"))  # 立即开始
    
    # 等待所有任务完成
    result1 = await task1
    result2 = await task2
    result3 = await task3
    # 总耗时：约 2 秒（并发执行）
    return [result1, result2, result3]

# ========== 方式 3：使用 gather（推荐）==========
async def using_gather():
    print("=== 使用 gather ===")
    tasks = [
        asyncio.create_task(slow_task("任务 1")),
        asyncio.create_task(slow_task("任务 2")),
        asyncio.create_task(slow_task("任务 3"))
    ]
    results = await asyncio.gather(*tasks)
    # 总耗时：约 2 秒
    return results

方式	执行方式	耗时	适用场景
`await coro()`	顺序执行	累加	需要顺序执行
`create_task()` + `await`	并发执行	最大值	需要并发，且需要单独处理每个任务
`asyncio.gather()`	并发执行	最大值	需要并发，且统一处理结果（推荐）

import asyncio

async def fetch_data(url: str):
    await asyncio.sleep(1)  # 模拟网络请求
    return f"数据来自 {url}"

# ========== 使用 create_task：需要单独控制每个任务 ==========
async def with_create_task():
    task1 = asyncio.create_task(fetch_data("url1"))
    task2 = asyncio.create_task(fetch_data("url2"))
    
    # 可以单独等待某个任务
    result1 = await task1
    print(f"第一个结果：{result1}")
    
    # 可以取消任务
    task2.cancel()
    try:
        result2 = await task2
    except asyncio.CancelledError:
        print("任务 2 被取消")
    
    # 可以检查任务状态
    print(f"任务 1 完成：{task1.done()}")
    print(f"任务 2 完成：{task2.done()}")

# ========== 使用 gather：统一处理所有任务 ==========
async def with_gather():
    # 更简洁，统一处理结果
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
        fetch_data("url3")
    )
    # results 是一个列表，包含所有结果
    return results

# ========== gather 的错误处理 ==========
async def with_gather_error_handling():
    async def fetch_with_error(url: str):
        if url == "error":
            raise ValueError("模拟错误")
        await asyncio.sleep(1)
        return f"数据来自 {url}"
    
    # return_exceptions=True：不抛出异常，返回异常对象
    results = await asyncio.gather(
        fetch_with_error("url1"),
        fetch_with_error("error"),
        fetch_with_error("url3"),
        return_exceptions=True
    )
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务 {i} 失败：{result}")
        else:
            print(f"任务 {i} 成功：{result}")

import asyncio

async def send_notification(user_id: int, message: str):
    """发送通知（不需要等待完成）"""
    await asyncio.sleep(1)  # 模拟发送
    print(f"通知已发送给用户 {user_id}: {message}")

async def process_order(order_id: int):
    """处理订单"""
    print(f"开始处理订单 {order_id}")
    await asyncio.sleep(2)  # 模拟处理
    print(f"订单 {order_id} 处理完成")
    
    # 发送通知（后台任务，不阻塞主流程）
    asyncio.create_task(send_notification(order_id, "订单处理完成"))
    # 注意：不 await，立即返回
    return f"订单 {order_id} 已处理"

async def main():
    order = await process_order(123)
    print(f"主流程完成：{order}")
    # 等待一下，让后台任务完成
    await asyncio.sleep(2)

asyncio.run(main())
# 输出：
# 开始处理订单 123
# 订单 123 处理完成
# 主流程完成：订单 已处理
# 通知已发送给用户 123: 订单处理完成

import asyncio

async def fetch_with_timeout(url: str, timeout: float):
    """带超时的请求"""
    task = asyncio.create_task(fetch_data(url))
    try:
        result = await asyncio.wait_for(task, timeout=timeout)
        return result
    except asyncio.TimeoutError:
        task.cancel()  # 取消任务
        return f"请求 {url} 超时"

async def fetch_data(url: str):
    await asyncio.sleep(5)  # 模拟慢请求
    return f"数据来自 {url}"

async def main():
    result = await fetch_with_timeout("http://example.com", timeout=2.0)
    print(result)  # 请求 http://example.com 超时

asyncio.run(main())

import asyncio

async def long_running_task(name: str, duration: int):
    try:
        print(f"任务 {name} 开始")
        await asyncio.sleep(duration)
        print(f"任务 {name} 完成")
        return f"任务 {name} 的结果"
    except asyncio.CancelledError:
        print(f"任务 {name} 被取消")
        raise

async def main():
    # 创建任务
    task1 = asyncio.create_task(long_running_task("A", 5))
    task2 = asyncio.create_task(long_running_task("B", 3))
    
    # 等待 2 秒
    await asyncio.sleep(2)
    
    # 检查任务状态
    print(f"任务 A 完成：{task1.done()}")
    print(f"任务 B 完成：{task2.done()}")
    
    # 取消任务 A
    task1.cancel()
    
    # 等待任务完成（或取消）
    try:
        result1 = await task1
    except asyncio.CancelledError:
        print("任务 A 已被取消")
    
    result2 = await task2
    print(f"任务 B 结果：{result2}")

asyncio.run(main())

# ❌ 错误：创建任务后不等待
async def main():
    task = asyncio.create_task(some_async_function())
    # 没有 await，主函数可能提前结束
    return "完成"

# ✅ 正确：确保任务完成
async def main():
    task = asyncio.create_task(some_async_function())
    result = await task  # 等待任务完成
    return result

# 或者使用 gather
async def main():
    await asyncio.gather(some_async_function())
    return "完成"

# ❌ 错误：没有运行事件循环
def sync_function():
    task = asyncio.create_task(async_function())  # 错误！

# ✅ 正确：在异步上下文中使用
async def async_function():
    task = asyncio.create_task(another_async_function())
    await task

# 或者在同步函数中使用 asyncio.run()
def sync_function():
    asyncio.run(async_function())

# ❌ 错误：混用导致顺序执行
async def main():
    await slow_task("A")  # 等待完成
    task = asyncio.create_task(slow_task("B"))  # 创建任务
    await slow_task("C")  # 等待完成
    await task  # 等待任务 B
    # 实际执行顺序：A -> C -> B（不是并发）

# ✅ 正确：统一使用 create_task
async def main():
    task1 = asyncio.create_task(slow_task("A"))
    task2 = asyncio.create_task(slow_task("B"))
    task3 = asyncio.create_task(slow_task("C"))
    await asyncio.gather(task1, task2, task3)

import asyncio
from typing import List

class BackgroundTaskManager:
    """后台任务管理器"""
    def __init__(self):
        self.tasks: List[asyncio.Task] = []

    def add_task(self, coro):
        """添加后台任务"""
        task = asyncio.create_task(coro)
        self.tasks.append(task)
        return task

    async def wait_all(self):
        """等待所有任务完成"""
        if self.tasks:
            await asyncio.gather(*self.tasks, return_exceptions=True)

    def cancel_all(self):
        """取消所有任务"""
        for task in self.tasks:
            if not task.done():
                task.cancel()

# 使用示例
async def main():
    manager = BackgroundTaskManager()
    # 添加多个后台任务
    manager.add_task(send_notification(1, "消息 1"))
    manager.add_task(send_notification(2, "消息 2"))
    
    # 主流程继续
    print("主流程执行中...")
    await asyncio.sleep(1)
    
    # 等待所有后台任务完成
    await manager.wait_all()
    print("所有任务完成")

import asyncio
import logging

async def safe_task(coro, task_name: str):
    """安全的任务包装器，自动处理异常"""
    try:
        result = await coro
        logging.info(f"任务 {task_name} 完成")
        return result
    except Exception as e:
        logging.error(f"任务 {task_name} 失败：{e}")
        return None

async def main():
    # 创建安全的任务
    task1 = asyncio.create_task(safe_task(risky_operation(), "任务 1"))
    task2 = asyncio.create_task(safe_task(another_operation(), "任务 2"))
    results = await asyncio.gather(task1, task2)
    return results

import asyncio

class TaskLimiter:
    """限制并发任务数量"""
    def __init__(self, max_concurrent: int):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.tasks = []

    async def add_task(self, coro):
        """添加任务（受并发限制）"""
        async with self.semaphore:
            return await coro

    async def run_tasks(self, coros):
        """运行多个任务"""
        tasks = [asyncio.create_task(self.add_task(coro)) for coro in coros]
        return await asyncio.gather(*tasks)

# 使用示例：最多同时运行 3 个任务
async def main():
    limiter = TaskLimiter(max_concurrent=3)
    coros = [fetch_data(f"url{i}") for i in range(10)]
    results = await limiter.run_tasks(coros)
    return results

from fastapi import FastAPI
import asyncio

app = FastAPI()

# ❌ 同步方式：阻塞服务器
@app.get("/sync")
def sync_endpoint():
    # 模拟数据库查询
    time.sleep(1)  # 阻塞！其他请求无法处理
    return {"status": "ok"}

# ✅ 异步方式：非阻塞
@app.get("/async")
async def async_endpoint():
    # 模拟异步数据库查询
    await asyncio.sleep(1)  # 不阻塞，其他请求可以处理
    return {"status": "ok"}

import aiofiles
import asyncio
from pathlib import Path

# ❌ 同步方式
def sync_process_files(file_paths):
    results = []
    for path in file_paths:
        with open(path, 'r') as f:
            content = f.read()  # 阻塞
        results.append(process(content))
    return results

# ✅ 异步方式
async def async_process_files(file_paths):
    async def process_file(path):
        async with aiofiles.open(path, 'r') as f:
            content = await f.read()  # 非阻塞
        return process(content)
    
    tasks = [process_file(path) for path in file_paths]
    results = await asyncio.gather(*tasks)  # 并发处理
    return results

import asyncpg
import asyncio

# ✅ 异步数据库操作
async def fetch_users():
    conn = await asyncpg.connect('postgresql://...')
    try:
        users = await conn.fetch('SELECT * FROM users')
        return users
    finally:
        await conn.close()

# 并发查询多个表
async def fetch_all_data():
    users, posts, comments = await asyncio.gather(
        fetch_users(),
        fetch_posts(),
        fetch_comments()
    )
    return users, posts, comments

import asyncio

async def background_task(task_id: str):
    """后台任务：不需要等待完成"""
    print(f"任务 {task_id} 开始")
    await asyncio.sleep(5)  # 模拟耗时操作
    print(f"任务 {task_id} 完成")

async def critical_operation():
    """关键操作：需要等待完成"""
    print("关键操作开始")
    await asyncio.sleep(1)
    print("关键操作完成")
    return "结果"

async def main():
    # 关键操作：必须等待
    result = await critical_operation()
    print(f"得到结果：{result}")
    
    # 后台任务：使用 create_task，不需要等待，立即返回
    task1 = asyncio.create_task(background_task("task-1"))
    task2 = asyncio.create_task(background_task("task-2"))
    
    print("主流程继续执行...")
    # 后台任务在后台运行，不影响主流程
    # 如果需要，可以稍后等待任务完成
    await asyncio.sleep(6)  # 等待后台任务完成
    print("所有任务完成")

asyncio.run(main())

# 认为 await 就是同步阻塞
await some_async_function()  # 认为这会阻塞整个程序

# await 是异步等待，会暂停当前协程，但允许其他协程运行
async def task1():
    await asyncio.sleep(2)  # 暂停 task1

async def task2():
    await asyncio.sleep(1)  # 暂停 task2

# 两个任务并发执行，总耗时约 2 秒（不是 3 秒）
await asyncio.gather(task1(), task2())

async def main():
    # 认为这样会自动并发
    await func1()
    await func2()
    await func3()

# ❌ 顺序执行（不是并发）
async def main():
    await func1()  # 等待完成
    await func2()  # 等待完成
    await func3()  # 等待完成
    # 总耗时 = func1 + func2 + func3

# ✅ 并发执行方式 1：使用 create_task
async def main():
    task1 = asyncio.create_task(func1())
    task2 = asyncio.create_task(func2())
    task3 = asyncio.create_task(func3())
    await asyncio.gather(task1, task2, task3)
    # 总耗时 = max(func1, func2, func3)

# ✅ 并发执行方式 2：使用 gather（推荐）
async def main():
    await asyncio.gather(func1(), func2(), func3())
    # 总耗时 = max(func1, func2, func3)

# 认为所有函数都应该改成 async
async def add(a, b):
    return a + b  # CPU 密集型操作，不需要 async

# CPU 密集型：使用同步函数
def add(a, b):
    return a + b

# I/O 密集型：使用异步函数
async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def get_data():
    return "data"

def main():
    result = get_data()  # ❌ 错误！result 是协程对象
    print(result)  # <coroutine object ...>

async def get_data():
    return "data"

async def main():
    result = await get_data()  # ✅ 正确
    print(result)  # "data"

async def main():
    # ❌ 错误：创建任务后不等待，主函数可能提前结束
    asyncio.create_task(long_running_task())
    return "完成"  # 任务可能还没完成就返回了

async def main():
    # ✅ 正确方式 1：等待任务完成
    task = asyncio.create_task(long_running_task())
    result = await task
    return result

    # ✅ 正确方式 2：使用 gather
    await asyncio.gather(long_running_task())
    return "完成"

    # ✅ 正确方式 3：后台任务（明确不需要等待）
    task = asyncio.create_task(background_task())
    # 确保事件循环运行足够长时间
    await asyncio.sleep(10)  # 等待后台任务完成
    return "完成"

# ✅ 好的异步函数设计
async def fetch_user_data(user_id: int):
    """清晰的异步函数"""
    async with aiohttp.ClientSession() as session:
        async with session.get(f'/api/users/{user_id}') as response:
            return await response.json()

# ❌ 不好的设计：混合同步和异步
async def bad_function():
    time.sleep(1)  # ❌ 同步阻塞，破坏了异步的优势
    await asyncio.sleep(1)  # ✅ 异步等待

async def robust_async_function():
    try:
        result = await some_async_operation()
        return result
    except Exception as e:
        logger.error(f"操作失败：{e}")
        raise

# 批量操作时的错误处理
async def batch_operations():
    tasks = [operation(i) for i in range(10)]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务 {i} 失败：{result}")
        else:
            print(f"任务 {i} 成功：{result}")

# ✅ 使用异步上下文管理器
async def process_file():
    async with aiofiles.open('file.txt', 'r') as f:
        content = await f.read()  # 文件自动关闭

# ✅ 数据库连接管理
async def query_database():
    async with asyncpg.create_pool('postgresql://...') as pool:
        async with pool.acquire() as conn:
            result = await conn.fetch('SELECT * FROM users')  # 连接自动释放

# ✅ 使用 asyncio.gather 并发执行
async def fetch_multiple_urls(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

# ✅ 使用 create_task 实现更细粒度的控制
async def fetch_with_control(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch_url(session, url)) for url in urls]
        # 可以单独处理每个任务
        results = []
        for task in tasks:
            try:
                result = await task
                results.append(result)
            except Exception as e:
                print(f"任务失败：{e}")
                results.append(None)
        return results

# ✅ 限制并发数量（避免过多并发）
async def fetch_with_limit(urls, limit=10):
    semaphore = asyncio.Semaphore(limit)
    
    async def fetch_with_semaphore(url):
        async with semaphore:
            return await fetch_url(url)
    
    tasks = [asyncio.create_task(fetch_with_semaphore(url)) for url in urls]
    return await asyncio.gather(*tasks)

特性	同步函数	异步函数 + await	create_task
执行方式	顺序执行	顺序执行（单个协程）	并发执行（多个任务）
阻塞性	阻塞线程	暂停协程，不阻塞线程	不阻塞，立即返回
适用场景	CPU 密集型	I/O 密集型（单个）	I/O 密集型（多个并发）
并发能力	无	无（单个）	高（多个）
代码复杂度	低	中	中高
返回类型	直接返回值	协程对象	Task 对象

深入理解 Python 异步编程：async、await 与同步函数详解

引言

同步函数 vs 异步函数

什么是同步函数？

什么是异步函数？

对比示例：处理 100 个网络请求

async 关键字详解

async 的作用

async 函数的特征

常见错误示例

await 关键字详解

await 的作用机制

await 的执行流程

await vs 同步阻塞

await 可以等待什么？

asyncio.create_task 详解

什么是 create_task？

create_task 的核心作用

create_task vs await

create_task vs asyncio.gather

create_task 的实际应用场景

场景 1：后台任务（Fire-and-Forget）

场景 2：超时控制

场景 3：任务取消和状态检查

create_task 的常见错误

错误 1：忘记 await Task

错误 2：在事件循环外使用

错误 3：混用 create_task 和直接 await

create_task 最佳实践

1. 后台任务管理

2. 任务包装器

3. 限制并发数量

总结：create_task 的核心要点

实际应用场景

场景 1：Web API 请求处理

场景 2：批量文件处理

场景 3：数据库操作

场景 4：任务调度与后台任务（使用 create_task）

常见误区

误区 1：await 就是同步等待

误区 2：异步函数会自动并发

误区 3：所有函数都应该用 async

误区 4：忘记 await 异步函数

误区 5：create_task 后忘记等待

最佳实践

1. 何时使用异步？

2. 异步函数设计原则

3. 错误处理

4. 资源管理

5. 性能优化技巧

总结

核心概念回顾

关键区别总结

记忆要点

实践建议

延伸阅读

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具