Python asyncio 异步编程完全指南：从 async/await 到并发实战 | 2026最新入门教程

1
import requests
2
import time
3

4
def fetch_urls(urls):
5
    results = []
6
    for url in urls:
7
        response = requests.get(url)  # 每次请求阻塞等待
8
        results.append(response.text)
9
    return results
10

11
# 10 个 URL，每个耗时 1 秒 → 总耗时约 10 秒
12
urls = [f"https://httpbin.org/delay/1?id={i}" for i in range(10)]
13
start = time.time()
14
fetch_urls(urls)
15
print(f"同步耗时: {time.time() - start:.2f}s")  # 约 10s

1
import aiohttp
2
import asyncio
3
import time
4

5
async def fetch_urls_async(urls):
6
    async with aiohttp.ClientSession() as session:
7
        tasks = [fetch_one(session, url) for url in urls]
8
        results = await asyncio.gather(*tasks)
9
    return results
10

11
async def fetch_one(session, url):
12
    async with session.get(url) as response:
13
        return await response.text()
14

15
# 10 个 URL，并发执行 → 总耗时约 1 秒
16
urls = [f"https://httpbin.org/delay/1?id={i}" for i in range(10)]
17
start = time.time()
18
asyncio.run(fetch_urls_async(urls))
19
print(f"异步耗时: {time.time() - start:.2f}s")  # 约 1s

1
async def my_coroutine():
2
    print("开始执行")
3
    await asyncio.sleep(1)  # 模拟 I/O 操作，暂停 1 秒
4
    print("执行完成")
5
    return "结果"
6

7
# 协程调用方式
8
result = asyncio.run(my_coroutine())
9
print(result)  # "结果"

1
# 1. 协程（Coroutine）
2
async def coro():
3
    await asyncio.sleep(1)
4
    return "协程结果"
5

6
result = await coro()
7

8
# 2. 任务（Task）- 协程的调度包装
9
task = asyncio.create_task(coro())
10
result = await task
11

12
# 3. Future - 低层级异步操作的结果表示
13
future = asyncio.Future()
14
future.set_result("Future结果")
15
result = await future

1
import asyncio
2

3
async def say_hello():
4
    await asyncio.sleep(1)
5
    return "Hello"
6

7
# ❌ 直接调用协程不会执行
8
coro = say_hello()  # 返回协程对象，不执行
9
# coro 是一个 coroutine 对象，需要被 await 或调度
10

11
# ✅ 正确方式 1: await 协程
12
async def main1():
13
    result = await say_hello()  # 直接 await
14
    print(result)
15

16
# ✅ 正确方式 2: 创建 Task
17
async def main2():
18
    task = asyncio.create_task(say_hello())  # 立即调度
19
    result = await task  # 等待完成
20
    print(result)
21

22
# ✅ 正确方式 3: asyncio.run()
23
asyncio.run(main1())

1
import asyncio
2

3
async def main():
4
    print("Hello asyncio")
5
    await asyncio.sleep(1)
6
    print("Done")
7

8
# 创建事件循环，运行协程，关闭循环
9
asyncio.run(main())
10

11
# 注意：一个程序只能调用一次 asyncio.run()
12
# 不要在已有事件循环中再次调用

1
import asyncio
2

3
async def task(name, delay):
4
    await asyncio.sleep(delay)
5
    print(f"{name} 完成")
6
    return f"{name} 结果"
7

8
async def main():
9
    # 并发运行 3 个任务
10
    results = await asyncio.gather(
11
        task("A", 1),
12
        task("B", 2),
13
        task("C", 1),
14
    )
15
    print(results)  # ['A 结果', 'B 结果', 'C 结果']
16

17
asyncio.run(main())

1
# return_exceptions=True: 异常作为结果返回，不中断其他任务
2
results = await asyncio.gather(
3
    task("A", 1),
4
    failing_task(),  # 这个会抛异常
5
    task("C", 1),
6
    return_exceptions=True
7
)
8
# results = ['A 结果', ValueError(...), 'C 结果']
9

10
# return_exceptions=False (默认): 第一个异常会中断 gather

1
import asyncio
2

3
async def background_task():
4
    while True:
5
        print("后台任务运行中...")
6
        await asyncio.sleep(1)
7

8
async def main():
9
    # 创建后台任务，立即开始执行
10
    task = asyncio.create_task(background_task())
11

12
    # 主任务执行其他工作
13
    await asyncio.sleep(3)
14
    print("主任务完成")
15

16
    # 取消后台任务
17
    task.cancel()
18
    try:
19
        await task
20
    except asyncio.CancelledError:
21
        print("后台任务已取消")
22

23
asyncio.run(main())

1
import asyncio
2

3
async def task(name, delay):
4
    await asyncio.sleep(delay)
5
    return name
6

7
async def main():
8
    tasks = [asyncio.create_task(task(f"T{i}", i)) for i in range(5)]
9

10
    # FIRST_COMPLETED: 任一完成即返回
11
    done, pending = await asyncio.wait(
12
        tasks,
13
        return_when=asyncio.FIRST_COMPLETED
14
    )
15
    print(f"已完成: {done}")
16
    print(f"待处理: {pending}")
17

18
    # 取消剩余任务
19
    for t in pending:
20
        t.cancel()
21

22
asyncio.run(main())

1
# 异步等待，不阻塞线程
2
await asyncio.sleep(1)  # 1 秒
3

4
# 等待 0 秒：让事件循环有机会调度其他任务
5
await asyncio.sleep(0)  # 立即切换，类似 yield

1
import aiohttp
2
import asyncio
3

4
async def fetch_url(session, url):
5
    try:
6
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:
7
            if response.status == 200:
8
                return await response.text()
9
            else:
10
                return f"Error: {response.status}"
11
    except asyncio.TimeoutError:
12
        return "Timeout"
13
    except Exception as e:
14
        return f"Exception: {e}"
15

16
async def fetch_all(urls):
17
    async with aiohttp.ClientSession() as session:
18
        tasks = [fetch_url(session, url) for url in urls]
19
        results = await asyncio.gather(*tasks)
20
        return results
21

22
# 示例
23
urls = [
24
    "https://httpbin.org/get",
25
    "https://httpbin.org/ip",
26
    "https://httpbin.org/headers",
27
]
28

29
results = asyncio.run(fetch_all(urls))
30
for r in results[:100]:  # 只显示前 100 字符
31
    print(r[:100])

1
import aiohttp
2
import asyncio
3

4
async def post_json():
5
    async with aiohttp.ClientSession() as session:
6
        payload = {"name": "test", "value": 123}
7

8
        async with session.post(
9
            "https://httpbin.org/post",
10
            json=payload,
11
            headers={"Content-Type": "application/json"}
12
        ) as response:
13
            data = await response.json()
14
            print(data["json"])  # {"name": "test", "value": 123}
15

16
asyncio.run(post_json())

1
import aiohttp
2
import asyncio
3

4
# 全局会话复用（推荐）
5
class HttpClient:
6
    _session = None
7

8
    @classmethod
9
    async def get_session(cls):
10
        if cls._session is None or cls._session.closed:
11
            cls._session = aiohttp.ClientSession(
12
                timeout=aiohttp.ClientTimeout(total=30),
13
                connector=aiohttp.TCPConnector(limit=100)  # 连接池限制
14
            )
15
        return cls._session
16

17
    @classmethod
18
    async def close(cls):
19
        if cls._session:
20
            await cls._session.close()
21

22
async def fetch(url):
23
    session = await HttpClient.get_session()
24
    async with session.get(url) as response:
25
        return await response.text()
26

27
async def main():
28
    urls = [f"https://httpbin.org/get?id={i}" for i in range(50)]
29
    results = await asyncio.gather(*[fetch(url) for url in urls])
30
    await HttpClient.close()
31
    print(f"获取了 {len(results)} 个响应")
32

33
asyncio.run(main())

1
import aiofiles
2
import asyncio
3

4
async def write_file(path, content):
5
    async with aiofiles.open(path, mode='w', encoding='utf-8') as f:
6
        await f.write(content)
7

8
async def read_file(path):
9
    async with aiofiles.open(path, mode='r', encoding='utf-8') as f:
10
        content = await f.read()
11
        return content
12

13
async def main():
14
    await write_file("test.txt", "Hello aiofiles!")
15
    content = await read_file("test.txt")
16
    print(content)
17

18
asyncio.run(main())

1
import aiofiles
2
import asyncio
3
import os
4

5
async def process_file(filepath):
6
    async with aiofiles.open(filepath, 'r', encoding='utf-8') as f:
7
        content = await f.read()
8
    # 处理内容
9
    lines = content.split('\n')
10
    return filepath, len(lines)
11

12
async def process_directory(directory):
13
    files = [f for f in os.listdir(directory) if f.endswith('.txt')]
14
    tasks = [process_file(os.path.join(directory, f)) for f in files]
15
    results = await asyncio.gather(*tasks)
16

17
    for filepath, count in results:
18
        print(f"{filepath}: {count} 行")
19

20
asyncio.run(process_directory("./data"))

1
import asyncio
2

3
async def task(name, delay):
4
    await asyncio.sleep(delay)
5
    print(f"{name} 完成")
6
    return name
7

8
async def main():
9
    async with asyncio.TaskGroup() as tg:
10
        t1 = tg.create_task(task("A", 1))
11
        t2 = tg.create_task(task("B", 2))
12
        t3 = tg.create_task(task("C", 1))
13

14
    # TaskGroup 退出时，所有任务已完成
15
    print(f"结果: {t1.result()}, {t2.result()}, {t3.result()}")
16

17
asyncio.run(main())

1
import asyncio
2

3
async def failing_task():
4
    await asyncio.sleep(0.5)
5
    raise ValueError("任务失败")
6

7
async def normal_task():
8
    await asyncio.sleep(1)
9
    return "正常完成"
10

11
async def main():
12
    try:
13
        async with asyncio.TaskGroup() as tg:
14
            tg.create_task(normal_task())
15
            tg.create_task(failing_task())
16
            tg.create_task(normal_task())
17
    except ExceptionGroup as eg:
18
        # ExceptionGroup 包含所有异常
19
        print(f"捕获异常组: {eg}")
20
        for exc in eg.exceptions:
21
            print(f"  - {type(exc).__name__}: {exc}")
22

23
asyncio.run(main())

1
import aiohttp
2
import asyncio
3

4
async def fetch_with_limit(session, url, semaphore):
5
    async with semaphore:  # 获取信号量，限制并发
6
        async with session.get(url) as response:
7
            return await response.text()
8

9
async def fetch_all_limited(urls, max_concurrent=10):
10
    semaphore = asyncio.Semaphore(max_concurrent)
11

12
    async with aiohttp.ClientSession() as session:
13
        tasks = [
14
            fetch_with_limit(session, url, semaphore)
15
            for url in urls
16
        ]
17
        results = await asyncio.gather(*tasks)
18
        return results
19

20
# 限制最多 5 个并发请求
21
urls = [f"https://httpbin.org/delay/1?id={i}" for i in range(20)]
22
asyncio.run(fetch_all_limited(urls, max_concurrent=5))

1
import asyncio
2
import time
3

4
class RateLimiter:
5
    def __init__(self, rate_per_second):
6
        self.interval = 1.0 / rate_per_second
7
        self.last_time = 0
8

9
    async def acquire(self):
10
        current = time.monotonic()
11
        wait_time = self.last_time + self.interval - current
12
        if wait_time > 0:
13
            await asyncio.sleep(wait_time)
14
        self.last_time = time.monotonic()
15

16
async def fetch_rate_limited(urls, rate=2):
17
    limiter = RateLimiter(rate)
18
    results = []
19

20
    async with aiohttp.ClientSession() as session:
21
        for url in urls:
22
            await limiter.acquire()  # 限制速率
23
            async with session.get(url) as response:
24
                results.append(await response.text())
25

26
    return results
27

28
# 每秒最多 2 个请求
29
asyncio.run(fetch_rate_limited(urls, rate=2))

1
import aiosqlite
2
import asyncio
3

4
async def init_db():
5
    async with aiosqlite.connect("app.db") as db:
6
        await db.execute("""
7
            CREATE TABLE IF NOT EXISTS users (
8
                id INTEGER PRIMARY KEY,
9
                name TEXT NOT NULL,
10
                email TEXT UNIQUE
11
            )
12
        """)
13
        await db.commit()
14

15
async def insert_user(name, email):
16
    async with aiosqlite.connect("app.db") as db:
17
        await db.execute(
18
            "INSERT INTO users (name, email) VALUES (?, ?)",
19
            (name, email)
20
        )
21
        await db.commit()
22

23
async def get_all_users():
24
    async with aiosqlite.connect("app.db") as db:
25
        async with db.execute("SELECT * FROM users") as cursor:
26
            rows = await cursor.fetchall()
27
            return rows
28

29
async def main():
30
    await init_db()
31
    await insert_user("Alice", "alice@example.com")
32
    await insert_user("Bob", "bob@example.com")
33
    users = await get_all_users()
34
    print(users)
35

36
asyncio.run(main())

1
import asyncpg
2
import asyncio
3

4
async def query_postgres():
5
    conn = await asyncpg.connect(
6
        host="localhost",
7
        port=5432,
8
        user="postgres",
9
        password="password",
10
        database="mydb"
11
    )
12

13
    # 执行查询
14
    rows = await conn.fetch("SELECT * FROM users LIMIT 10")
15
    for row in rows:
16
        print(row)
17

18
    # 执行插入
19
    await conn.execute(
20
        "INSERT INTO users (name, email) VALUES ($1, $2)",
21
        "Charlie", "charlie@example.com"
22
    )
23

24
    await conn.close()
25

26
asyncio.run(query_postgres())

1
import asyncio
2

3
async def long_running_task():
4
    try:
5
        print("任务开始")
6
        for i in range(10):
7
            await asyncio.sleep(1)
8
            print(f"进度: {i+1}/10")
9
        print("任务完成")
10
    except asyncio.CancelledError:
11
        print("任务被取消，执行清理...")
12
        # 执行清理操作
13
        raise  # 重新抛出，让调用者知道被取消
14

15
async def main():
16
    task = asyncio.create_task(long_running_task())
17

18
    await asyncio.sleep(3)
19
    task.cancel()  # 发送取消请求
20

21
    try:
22
        await task
23
    except asyncio.CancelledError:
24
        print("确认任务已取消")
25

26
asyncio.run(main())

1
import asyncio
2

3
async def slow_operation():
4
    await asyncio.sleep(10)
5
    return "完成"
6

7
async def main():
8
    try:
9
        # asyncio.wait_for 设置超时
10
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
11
        print(result)
12
    except asyncio.TimeoutError:
13
        print("操作超时")
14

15
asyncio.run(main())

1
import asyncio
2

3
async def main():
4
    async with asyncio.timeout(3.0):
5
        await slow_operation()  # 超时后自动取消
6

7
asyncio.run(main())

1
import asyncio
2
import aiohttp
3
import requests
4
import time
5

6
# 同步版本
7
def sync_fetch(urls):
8
    start = time.time()
9
    results = []
10
    for url in urls:
11
        results.append(requests.get(url).text)
12
    elapsed = time.time() - start
13
    return results, elapsed
14

15
# 异步版本
16
async def async_fetch(urls):
17
    start = time.time()
18
    async with aiohttp.ClientSession() as session:
19
        tasks = [session.get(url) for url in urls]
20
        responses = await asyncio.gather(*tasks)
21
        results = [await r.text() for r in responses]
22
    elapsed = time.time() - start
23
    return results, elapsed
24

25
# 测试
26
urls = [f"https://httpbin.org/delay/1?id={i}" for i in range(10)]
27

28
# 同步: 约 10 秒
29
_, sync_time = sync_fetch(urls)
30
print(f"同步耗时: {sync_time:.2f}s")
31

32
# 异步: 约 1 秒
33
_, async_time = asyncio.run(async_fetch(urls))
34
print(f"异步耗时: {async_time:.2f}s")
35

36
# 性能提升约 10 倍
37
print(f"性能提升: {sync_time/async_time:.1f}x")

1
# ❌ 错误：直接调用协程不执行
2
async def my_coro():
3
    return "结果"
4

5
def wrong_call():
6
    result = my_coro()  # 返回协程对象，不是结果
7
    print(result)  # <coroutine object my_coro at ...>
8

9
# ✅ 正确
10
def correct_call():
11
    result = asyncio.run(my_coro())
12
    print(result)  # "结果"

1
# ❌ 错误：在协程中使用阻塞操作
2
async def bad_example():
3
    time.sleep(5)  # 阻塞整个事件循环！
4
    return "完成"
5

6
# ✅ 正确：使用异步版本
7
async def good_example():
8
    await asyncio.sleep(5)  # 不阻塞，允许其他任务运行
9
    return "完成"
10

11
# ✅ 正确：在线程池中运行阻塞代码
12
async def run_blocking():
13
    result = await asyncio.to_thread(blocking_function)
14
    return result

1
# ❌ 错误：不关闭会话导致资源泄漏
2
async def bad():
3
    session = aiohttp.ClientSession()
4
    await session.get("https://example.com")
5
    # 没有 close！
6

7
# ✅ 正确：使用 async with
8
async def good():
9
    async with aiohttp.ClientSession() as session:
10
        await session.get("https://example.com")
11
    # 自动关闭

1
# ❌ 错误：无限制并发
2
async def bad():
3
    tasks = [fetch(url) for url in urls]  # 可能数千个
4
    await asyncio.gather(*tasks)  # 可能耗尽内存或触发服务器限流
5

6
# ✅ 正确：使用信号量限制
7
async def good():
8
    semaphore = asyncio.Semaphore(100)
9
    tasks = [fetch_with_limit(url, semaphore) for url in urls]
10
    await asyncio.gather(*tasks)

1
import asyncio
2
import time
3

4
def blocking_io():
5
    time.sleep(2)  # 阻塞操作
6
    return "阻塞函数结果"
7

8
async def main():
9
    # 在线程池中运行阻塞函数，不阻塞事件循环
10
    result = await asyncio.to_thread(blocking_io)
11
    print(result)
12

13
    # 同时可以运行其他异步任务
14
    await asyncio.sleep(1)
15

16
asyncio.run(main())

1
import asyncio
2
from concurrent.futures import ThreadPoolExecutor
3

4
def cpu_intensive_task(n):
5
    return sum(range(n))
6

7
async def main():
8
    executor = ThreadPoolExecutor(max_workers=4)
9
    loop = asyncio.get_event_loop()
10

11
    # 在线程池中运行 CPU 密集任务
12
    result = await loop.run_in_executor(
13
        executor,
14
        cpu_intensive_task,
15
        10_000_000
16
    )
17
    print(f"计算结果: {result}")
18

19
asyncio.run(main())