Python 异步编程：完整指南

当您的 Python 脚本耐心等待 API 响应、数据库查询或文件操作完成时，这段时间往往被闲置。借助 Python 异步编程，您的代码可以同时处理多项任务。这样一来，当某个操作在等待时，其他操作仍可向前推进，把空档变为产出，常常能把几分钟的等待缩短到几秒。

在本指南中，我将通过小型项目带您学习 Python 中的异步编程要点。您将看到协程、事件循环和异步 I/O 如何让代码响应更敏捷。

如果您想学习如何构建异步 Web API，务必查看这门关于 FastAPI 的课程。

什么是 Python 异步编程？

在传统的同步 Python 中，代码按行顺序执行。比如调用一个 API 时，程序会停下来等待响应。如果这需要两秒，那么整段程序就空等两秒。异步编程允许您的代码发起 API 调用后，继续处理其他任务。

当响应到达后，代码会从暂停处继续执行。与其等待每个操作完成，您可以同时运行多个操作。这在代码需要等待外部系统（如数据库、API 或文件系统）响应时尤为重要。 

为实现这一点，Python 的异步系统使用了几个核心概念：

• 协程（Coroutine）：用 async def 而不是 def 定义的函数。它们可以暂停并恢复执行，非常适合涉及等待的操作。

• await：该关键字告诉 Python：“在此操作完成前暂停此协程，同时让其他代码先运行。”

• 事件循环（Event loop）：管理所有协程的引擎，决定运行哪个、何时在它们之间切换。

• 任务（Task）：为并发执行而封装的协程。使用 asyncio.create_task() 创建，以同时运行多个操作。

为避免对异步能做什么（以及不能做什么）产生误解，请记住：

• 异步最适合 I/O 密集型工作，如 HTTP 请求、数据库查询和文件操作，这些场景下您的代码在等待外部系统。

• 异步不适用于 CPU 密集型工作，例如复杂计算或数据处理，此时代码主要在“算”而不是“等”。

理解这些概念的最佳方式就是编写真实的异步代码。下一节您将创建第一个异步函数，直观了解协程与事件循环如何配合。

您的第一个 Python 异步函数

在写异步代码前，先看一个在执行前等待的常规同步函数：

import time

def greet_after_delay():
    print("Starting...")
    time.sleep(2)  # Blocks for 2 seconds
    print("Hello!")

greet_after_delay()技术支持

Starting...
Hello!技术支持

这个函数可用，但 time.sleep(2) 会阻塞整个程序。在这两秒里别的东西都跑不了。

下面是异步版本：

import asyncio

async def greet_after_delay():
    print("Starting...")
    await asyncio.sleep(2)  # Pauses, but doesn't block
    print("Hello!")

asyncio.run(greet_after_delay())技术支持

Starting...
Hello!技术支持

输出看起来一样，但底层发生了变化。三处改动让它变成了异步：

1. 用 async def 代替 def 将其声明为协程。

2. 用 await asyncio.sleep(2) 代替 time.sleep(2)，实现“暂停不阻塞”。

3. asyncio.run() 启动事件循环并运行协程。

注意 asyncio.sleep() 本身是异步函数，因此需要 await。这是关键规则：每个异步函数都必须用 await 调用。无论是内置的 asyncio.sleep()，还是您自己编写的，忘记 await 就不会真正执行。

目前这个异步版本看起来并不更快。这是因为只有一个任务。真正的优势体现在同时运行多个协程时，下一节会介绍。

还有一点很重要：不能像普通函数那样直接调用异步函数。我们来试试：

result = greet_after_delay()
print(result)
print(type(result))技术支持

<coroutine object greet_after_delay at 0x...>
<class 'coroutine'>技术支持

调用 greet_after_delay() 返回的是协程对象，而不是结果。函数并未真正运行。您需要用 asyncio.run() 或在另一个函数内用 await 才能执行它。

事件循环如何工作

事件循环是异步编程背后的引擎。它管理协程并决定何时运行谁。运行异步版 greet_after_delay() 时，会按以下步骤进行：

1. asyncio.run() 创建事件循环。

2. 事件循环启动 greet_after_delay()。

3. 打印“Starting...”。

4. 遇到 await asyncio.sleep(2) → 协程暂停。

5. 事件循环检查：“还有其他任务吗？”（此刻没有）。

6. 两秒过去，sleep 完成。

7. 事件循环恢复 greet_after_delay()。

8. 打印“Hello!”。

9. 函数结束 → 事件循环退出。

第 5 步是异步变得有趣的地方。只有一个协程时，没别的可做。但当有多个协程时，事件循环会在一个协程等待时切换去做其他工作。与其在两秒的 sleep 里闲着，不如运行别的代码。

把事件循环想象成交通指挥。它不会让每辆车开得更快，但能在某辆车停下时让其他车先走，让整体更通畅。

常见异步错误：忘记 await

初学者常见错误是在另一个异步函数里调用协程时忘记 await：

import asyncio

async def get_message():
    await asyncio.sleep(1)
    return "Hello!"

async def main():
    message = get_message()  # Missing await!
    print(message)

asyncio.run(main())技术支持

<coroutine object get_message at 0x...>
RuntimeWarning: coroutine 'get_message' was never awaited技术支持

没有 await，拿到的是协程对象而不是返回值。Python 还会警告该协程从未被等待执行。

修复很简单：

async def main():
    message = await get_message()  # Added await
    print(message)

asyncio.run(main())技术支持

Hello!技术支持

当您看到关于协程未被等待的 RuntimeWarning，请检查是否对每个异步函数调用都使用了 await。

并发的 Python 异步任务 

上一节我们把同步函数改成了异步，但并没更快。这是因为只运行了一个协程。异步的真正威力体现在同时运行多个协程时。

为何顺序 await 仍然是顺序执行

您可能以为调用多个异步函数会自动并发运行。但看看我们把 greet_after_delay() 调用三次会发生什么：

import asyncio
import time

async def greet_after_delay(name):
    print(f"Starting {name}...")
    await asyncio.sleep(2)
    print(f"Hello, {name}!")

async def main():
    start = time.perf_counter()
    
    await greet_after_delay("Alice")
    await greet_after_delay("Bob")
    await greet_after_delay("Charlie")
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())技术支持

Starting Alice...
Hello, Alice!
Starting Bob...
Hello, Bob!
Starting Charlie...
Hello, Charlie!
Total time: 6.01 seconds技术支持

三个两秒的任务却花了六秒。每个 await 都会等对应协程完成才执行下一行。代码是异步的，但仍按顺序运行。

使用 asyncio.gather() 并发运行异步任务 

要同时运行多个协程，使用 asyncio.gather()。它接收多个协程并并发执行：

async def main():
    start = time.perf_counter()
    
    await asyncio.gather(
        greet_after_delay("Alice"),
        greet_after_delay("Bob"),
        greet_after_delay("Charlie"),
    )
    
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f} seconds")

asyncio.run(main())技术支持

Starting Alice...
Starting Bob...
Starting Charlie...
Hello, Alice!
Hello, Bob!
Hello, Charlie!
Total time: 2.00 seconds技术支持

两秒而不是六秒。三个协程一起启动，同时 sleep，并一同结束。只改一处就快了 3 倍。

注意输出顺序：三个 “Starting...” 都先打印，之后才是 “Hello...”。这说明三个协程处在同一两秒窗口内同时运行，而不是互相等待。

asyncio.gather() 会按照您传入协程的顺序返回结果列表。如果协程有返回值，可以这样接收：

async def fetch_number(n):
    await asyncio.sleep(1)
    return n * 10

async def main():
    results = await asyncio.gather(
        fetch_number(1),
        fetch_number(2),
        fetch_number(3),
    )
    print(results)

asyncio.run(main())技术支持

[10, 20, 30]技术支持

结果顺序为 [10, 20, 30]，与传给 gather() 的协程顺序一致。

使用 aiohttp 进行 Python 异步 HTTP 请求

到目前为止我们用 asyncio.sleep() 模拟了延迟。现在来发真实的 HTTP 请求。您可能会想到 requests 库，但在这里不行。requests 是同步的，会阻塞事件循环，与异步初衷相悖。

请改用 aiohttp，这是专为此目的构建的异步 HTTP 客户端。

aiohttp 简介

以下是使用 aiohttp 获取 URL 的方法：

import aiohttp
import asyncio

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

async def main():
    html = await fetch("https://example.com")
    print(f"Fetched {len(html)} characters")

asyncio.run(main())技术支持

Fetched 513 characters技术支持

注意嵌套的两个 async with。它们各自管理不同的资源，理解其作用是正确使用 aiohttp 的关键。

ClientSession 在 aiohttp 中如何工作

使用 aiohttp 发起请求时，会按以下步骤进行：

1. aiohttp.ClientSession() 创建一个连接池（初始为空）。

2. session.get(url) 检查连接池：“是否已有到该主机（网站服务器）的打开连接？”

3. 如果没有，则创建新的 TCP 连接（互联网传输数据的基础协议）和 SSL 握手（HTTPS 的加密建立）。

4. 发送 HTTP 请求，等待响应头。

5. 响应对象持有该连接。

6. await response.text() 从网络读取响应体数据。

7. 退出内层 async with 块：连接回收到连接池（保持打开）。

8. 下一次请求相同主机时，复用连接池里的连接（跳过第 3 步）。

9. 退出外层 async with 块：关闭所有池中连接。

第 7、8 步是关键洞见。连接池会在请求之间保持连接活跃。再次请求同一主机时，完全跳过 TCP 和 SSL 握手。

这很重要，因为建立新连接很慢。一次 TCP 握手需要往返一次；一次 SSL 握手还要两次。取决于延迟，这在发送首个字节数据前就要 100–300ms。

为所有请求复用同一个会话

现在您明白了，为每次请求都创建新会话会出问题：

# Wrong: new session for each request
async def fetch_bad(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    results = await asyncio.gather(*[fetch_bad(url) for url in urls])技术支持

每次调用 fetch_bad() 都会创建一个新会话，其连接池为空。即使请求都指向同一主机，每次都要承担完整的握手成本。

修正方法是创建一个会话并把它传给获取函数：

# Right: reuse a single session
async def fetch_good(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch_good(session, url) for url in urls])技术支持

使用共享会话时，第一次请求建立连接，随后九个请求都能复用。一场握手代替十场。

异步 HTTP 请求示例：抓取 Hacker News

用 Hacker News API 实践一下。该 API 很适合展示异步特性，因为获取故事需要多次请求。如果您是第一次在 Python 中使用 REST API，请参考 Python APIs: A Guide to Building and Using APIs 了解基础概念。

Hacker News API 结构：

• https://hacker-news.firebaseio.com/v0/topstories.json 返回故事 ID 列表（纯数字）

• https://hacker-news.firebaseio.com/v0/item/{id}.json 返回单个故事详情

要获取 10 篇故事，需要 11 次请求：一次拿 ID 列表，再对每个故事各一次。这正是异步大显身手的场景。

首先，看看尝试获取第一篇故事时 API 返回了什么：

import aiohttp
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def main():
    async with aiohttp.ClientSession() as session:
        # Get top story IDs
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        print(f"Found {len(story_ids)} stories")
        print(f"First 5 IDs: {story_ids[:5]}")
        
        # Fetch first story details
        first_id = story_ids[0]
        async with session.get(f"{HN_API}/item/{first_id}.json") as response:
            story = await response.json()
        
        print(f"\nStory structure:")
        for key, value in story.items():
            print(f"  {key}: {repr(value)[:50]}")

asyncio.run(main())技术支持

Found 500 stories
First 5 IDs: [46051449, 46055298, 46021577, 46053566, 45984864]

Story structure:
  by: 'mikeayles'
  descendants: 22
  id: 46051449
  kids: [46054027, 46053889, 46053275, 46053515, 46053002,
  score: 217
  text: 'I got DOOM running in KiCad by rendering it with 
  time: 1764108815
  title: 'Show HN: KiDoom – Running DOOM on PCB Traces'
  type: 'story'
  url: 'https://www.mikeayles.com/#kidoom'技术支持

该 API 返回了 500 个故事 ID，每个故事包含 title、url、score、by（作者）等字段。 

顺序获取与并发获取对比

现在顺序获取 10 篇故事：

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        stories = []
        for story_id in story_ids[:10]:
            story = await fetch_story(session, story_id)
            stories.append(story)
        elapsed = time.perf_counter() - start
        
        print(f"Sequential: Fetched {len(stories)} stories in {elapsed:.2f} seconds")

asyncio.run(main())技术支持

Sequential: Fetched 10 stories in 2.41 seconds技术支持

接下来并发获取相同的故事：

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()
        
        start = time.perf_counter()
        tasks = [fetch_story(session, story_id) for story_id in story_ids[:10]]
        stories = await asyncio.gather(*tasks)
        elapsed = time.perf_counter() - start
        
        print(f"Concurrent: Fetched {len(stories)} stories in {elapsed:.2f} seconds")
        print("\nTop 3 stories:")
        for story in stories[:3]:
            print(f"  - {story.get('title', 'No title')}")

asyncio.run(main())技术支持

Concurrent: Fetched 10 stories in 0.69 seconds

Top 3 stories:
  - Show HN: KiDoom – Running DOOM on PCB Traces
  - AWS is 10x slower than a dedicated server for the same price [video]
  - Surprisingly, Emacs on Android is pretty good技术支持

并发版本快了 3.5 倍。不再等待每个请求完成才开始下一个，而是 10 个请求同时发起。这正是异步在真实网络 I/O 中带来的收益。

Python 异步错误处理与限流

并发获取数据时，可能会出现多种问题。您可能会因为请求过多而压垮服务器；有些请求可能会一直挂起；还有的会直接失败。而一旦失败，您需要恢复策略。

本节按发生顺序讲解各项问题：控制并发请求数量、设置超时、处理失败，以及在合适的时候重试。如果您需要回顾 Python 异常处理基础，请参阅 Python 中的异常与错误处理。下面这段基础代码会贯穿使用：

import aiohttp
import asyncio
import time

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()技术支持

用信号量进行限流

上一节我们一次性发了 10 个请求，运行良好。但如果需要获取 500 篇故事呢？或者抓取 10,000 个页面？

大多数 API 都有速率限制。比如每秒 10 次请求，或最多 100 个并发连接。超出限制，您会被阻止、限速或封禁。即使 API 不限速，同时发出上千请求也可能压垮您自己的系统或对方服务器。

您需要一种方法控制任意时刻“在途”的请求数量。这正是信号量（semaphore）的作用。

信号量像许可制度。假设您有三个许可。任何要发请求的任务必须先取得许可。完成后归还许可，新的请求才能使用。如果没有可用许可，任务就等待直到有空出。

在 3 个许可、4 个及以上任务的情况下，会这样进行：

1. 有三个可用许可。

2. 任务 A 取走一个许可（剩 2），开始请求。

3. 任务 B 取走一个许可（剩 1），开始请求。

4. 任务 C 取走一个许可（剩 0），开始请求。

5. 任务 D 需要许可，但没有可用——它等待。

6. 任务 A 完成，归还许可（可用 1）。

7. 任务 D 取得该许可并开始请求。

8. 如此循环，直到所有任务完成。

第 5 步中的等待是高效的。任务不会空转检查“有许可了吗？”，而是挂起自己让其他代码运行。只有当许可可用时，事件循环才会唤醒它。

代码实现如下。在 asyncio 中，使用 asyncio.Semaphore(n) 创建信号量，n 为许可数量。使用时将代码包在 async with semaphore: 中。进入时获取许可，退出时自动释放：

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:  # Acquire permit (or wait if none available)
        async with session.get(f"{HN_API}/item/{story_id}.json") as response:
            return await response.json()
    # Permit automatically released here技术支持

对比在有无信号量的情况下获取 30 篇故事：

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = (await response.json())[:30]

        # Without rate limiting: all 30 at once
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story(session, sid) for sid in story_ids])
        print(f"No limit: {time.perf_counter() - start:.2f}s (30 concurrent)")

        # With Semaphore(5): max 5 at a time
        semaphore = asyncio.Semaphore(5)
        start = time.perf_counter()
        await asyncio.gather(*[fetch_story_limited(session, sid, semaphore) for sid in story_ids])
        print(f"Semaphore(5): {time.perf_counter() - start:.2f}s (5 concurrent)")

asyncio.run(main())技术支持

No limit: 0.62s (30 concurrent)
Semaphore(5): 1.50s (5 concurrent)技术支持

使用信号量的版本更慢，因为每次只处理 5 个请求。但这就是权衡：以速度换取可控、对服务器友好的行为。

需要注意：信号量限制的是并发请求数，而非单位时间的请求数。Semaphore(10) 意味着“同时在途请求不超过 10 个”，而不是“每秒 10 个”。若需要严格的时间限速（例如精确每秒 10 个），可以把信号量与分批延迟结合，或使用 aiolimiter 等库。

使用 asyncio.wait_for() 设置超时

即使控制了并发，个别请求也可能会挂起。服务器可能接受了连接却一直不响应。没有超时，程序会无限等待。

asyncio.wait_for() 可为任意协程设置截止时间。传入协程和以秒为单位的 timeout。若未能及时完成，就会抛出 asyncio.TimeoutError：

async def slow_operation():
    print("Starting slow operation...")
    await asyncio.sleep(5)
    return "Done"

async def main():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
        print(f"Success: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out after 2 seconds")

asyncio.run(main())技术支持

Starting slow operation...
Operation timed out after 2 seconds技术支持

超时到期时，wait_for() 会取消该协程。您可以捕获 TimeoutError 决定后续处理：跳过、返回默认值，或进行重试。

对于并发请求，请分别为每个请求包裹超时。下面这个辅助函数会在超时时返回错误字典，而非抛错：

async def fetch_story_with_timeout(session, story_id, timeout=5.0):
    try:
        coro = fetch_story(session, story_id)
        return await asyncio.wait_for(coro, timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": f"Story {story_id} timed out"}技术支持

当协程被取消（因超时或其他原因）时，Python 会在其中抛出 asyncio.CancelledError。如果您的协程持有资源（如文件句柄或连接），请使用 try/finally 确保即使在取消时也能完成清理：

async def fetch_with_cleanup(session, url):
    print("Starting fetch...")
    try:
        async with session.get(url) as response:
            return await response.text()
    finally:
        print("Cleanup complete")  # Runs even on cancellation技术支持

与 asyncio.gather() 的错误处理

超时用于捕捉慢请求。但有些请求会立刻以错误结束。来看当一批请求中有一个失败时会发生什么。

首先，我们需要一个在 ID 无效时抛出异常的 fetch_story() 版本：

async def fetch_story_strict(session, story_id):
    story = await fetch_story(session, story_id)
    if story is None:
        raise ValueError(f"Story not found: {story_id}")
    return story技术支持

现在获取四个有效故事加一个无效 ID：

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]  # 4 valid + 1 invalid

        try:
            stories = await asyncio.gather(
                *[fetch_story_strict(session, sid) for sid in ids_to_fetch]
            )
            print(f"Got {len(stories)} stories")
        except ValueError as e:
            print(f"ERROR: {e}")

asyncio.run(main())技术支持

ERROR: Story not found: 99999999999技术支持

有一个无效 ID 时，我们丢失了所有四个成功结果。默认情况下，gather() 采用快速失败：任一异常会取消所有任务并向上抛出。

要保留部分结果，可添加 return_exceptions=True。这会改变 gather() 的行为：不再抛出异常，而是把异常与成功值一起作为列表项返回：

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        ids_to_fetch = story_ids[:4] + [99999999999]

        results = await asyncio.gather(
            *[fetch_story_strict(session, sid) for sid in ids_to_fetch],
            return_exceptions=True  # Don't raise, return exceptions in list
        )

        # Separate successes from failures using isinstance()
        stories = [r for r in results if not isinstance(r, Exception)]
        errors = [r for r in results if isinstance(r, Exception)]

        print(f"Got {len(stories)} stories, {len(errors)} failed")

asyncio.run(main())技术支持

Got 4 stories, 1 failed技术支持

isinstance(result, Exception) 的检查可将成功结果与错误分离。随后您可以处理成功的，记录或重试失败的。

指数退避的重试逻辑

有些失败是暂时的。服务器可能短时过载，或者网络抖动导致连接中断。在这些情况下，重试是合理的。

但立刻重试可能加剧问题。如果服务器本就吃力，重试会进一步施压。指数退避通过逐次拉长等待时间来解决。

该模式用 2 ** attempt 计算等待时长：第 0 次尝试等一秒（2⁰），第 1 次等两秒（2¹），第 2 次等四秒（2²），以此类推。这样能给服务器越来越多的恢复时间：

async def fetch_with_retry(session, story_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            story = await fetch_story(session, story_id)
            if story is None:
                raise ValueError(f"Story {story_id} not found")
            return story
        except (aiohttp.ClientError, ValueError):  # Catch specific exceptions
            if attempt == max_retries - 1:
                print(f"Story {story_id}: Failed after {max_retries} attempts")
                return None

            backoff = 2 ** attempt  # 1s, 2s, 4s...
            print(f"Story {story_id}: Attempt {attempt + 1} failed, retrying in {backoff}s...")
            await asyncio.sleep(backoff)技术支持

注意我们捕获的是特定异常（aiohttp.ClientError、ValueError），而不是裸 except。这确保只在可能是暂时性的问题上重试。代码错误导致的 KeyError 不应触发重试。

用一组有效与无效 ID 测试一下：

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{HN_API}/topstories.json") as response:
            story_ids = await response.json()

        test_ids = [story_ids[0], 99999999999, story_ids[1], 88888888888, story_ids[2]]

        results = await asyncio.gather(*[fetch_with_retry(session, sid) for sid in test_ids])

        successful = [r for r in results if r is not None]
        print(f"\nSuccessful: {len(successful)}, Failed: {len(test_ids) - len(successful)}")

asyncio.run(main())技术支持

Story 99999999999: Attempt 1 failed, retrying in 1s...
Story 88888888888: Attempt 1 failed, retrying in 1s...
Story 99999999999: Attempt 2 failed, retrying in 2s...
Story 88888888888: Attempt 2 failed, retrying in 2s...
Story 99999999999: Failed after 3 attempts
Story 88888888888: Failed after 3 attempts

Successful: 3, Failed: 2技术支持

在生产中，您还应加入抖动（在等待时间上加少量随机延迟），以避免多个失败请求在同一时刻同时重试。此外，应仅对暂时性错误（如 503 的服务器端网络问题）进行重试，对永久性错误（如 404 或 401）应立即放弃。

使用 aiosqlite 进行 Python 异步数据库存储

我们已经在合适的限流、超时与错误处理下获取了 Hacker News 的故事。现在把它们存入数据库。

使用常规的同步数据库库（如 sqlite3）会在查询时阻塞事件循环，背离异步编程的目的。代码等待数据库时，其他协程无法运行。对于异步应用，您需要异步数据库库。

aiosqlite 为 Python 内置的 sqlite3 提供了异步接口。它在一个线程池中执行数据库操作，从而不阻塞事件循环。SQLite 无需服务器配置——它只是一个文件——因此您可以立即运行这段代码。如果您是第一次在 Python 中使用数据库，Introduction to Databases in Python 课程涵盖了 aiosqlite 所基于的同步基础。

设置数据库

模式应该很熟悉。与 aiohttp.ClientSession 类似，您用 async with 管理连接：

import aiosqlite

async def init_db(db_path):
    async with aiosqlite.connect(db_path) as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        await db.commit()

asyncio.run(init_db("stories.db"))技术支持

关键函数：

• aiosqlite.connect(path) 打开（或创建）数据库文件。

• await db.execute(sql) 执行一条 SQL 语句。

• await db.commit() 将更改保存到磁盘。

保存故事

下面是保存单个故事的函数：

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )技术支持

? 占位符可以防止 SQL 注入——切勿使用 f-string 直接拼接 SQL 值。INSERT OR REPLACE 能在重复获取时更新已存在的故事。

完整的 Python 异步流水线：获取与存储

现在把本教程中的所有内容整合为一个完整流水线。我们将以限流的方式获取 20 篇 Hacker News 故事，并将其存入数据库：

import aiohttp
import aiosqlite
import asyncio

HN_API = "https://hacker-news.firebaseio.com/v0"

async def fetch_story(session, story_id):
    async with session.get(f"{HN_API}/item/{story_id}.json") as response:
        return await response.json()

async def fetch_story_limited(session, story_id, semaphore):
    async with semaphore:
        story = await fetch_story(session, story_id)
        if story:
            return story
        return None

async def save_story(db, story):
    await db.execute(
        "INSERT OR REPLACE INTO stories (id, title, url, score) VALUES (?, ?, ?, ?)",
        (story["id"], story.get("title", ""), story.get("url", ""), story.get("score", 0))
    )

async def main():
    # Initialize database
    async with aiosqlite.connect("hn_stories.db") as db:
        await db.execute("""
            CREATE TABLE IF NOT EXISTS stories (
                id INTEGER PRIMARY KEY,
                title TEXT,
                url TEXT,
                score INTEGER,
                fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        
        # Fetch stories
        async with aiohttp.ClientSession() as session:
            async with session.get(f"{HN_API}/topstories.json") as response:
                story_ids = await response.json()
            
            semaphore = asyncio.Semaphore(5)
            tasks = [fetch_story_limited(session, sid, semaphore) for sid in story_ids[:20]]
            stories = await asyncio.gather(*tasks)
        
        # Save to database
        for story in stories:
            if story:
                await save_story(db, story)
        await db.commit()
        
        # Query and display
        cursor = await db.execute("SELECT id, title, score FROM stories ORDER BY score DESC LIMIT 5")
        rows = await cursor.fetchall()
        
        print(f"Saved {len([s for s in stories if s])} stories. Top 5 by score:")
        for row in rows:
            print(f"  [{row[2]}] {row[1][:50]}")

asyncio.run(main())技术支持

Saved 20 stories. Top 5 by score:
  [671] Google Antigravity exfiltrates data via indirect p
  [453] Trillions spent and big software projects are stil
  [319] Ilya Sutskever: We're moving from the age of scali
  [311] Show HN: We built an open source, zero webhooks pa
  [306] FLUX.2: Frontier Visual Intelligence技术支持

该流水线运用了各节中的模式：用 ClientSession 进行连接池复用，用 Semaphore(5) 做限流，用 gather() 并发抓取，再用 aiosqlite 进行异步存储。各组件各司其职，互不阻塞。

每次运行该流程，您都会获取当天的热门故事。

总结

本教程从基础的 async/await 语法出发，带您完成了一个完整的数据流水线。您学习了协程如何暂停与恢复，事件循环如何管理并发任务，以及 asyncio.gather() 如何同时运行多项操作。您还用 aiohttp 发起真实 HTTP 请求，用信号量控制并发，用超时与重试处理失败，并用 aiosqlite 将结果存入数据库。

当代码需要等待外部系统（HTTP API、数据库、文件 I/O 或网络套接字）时，请使用异步。对于数据处理或数值计算等 CPU 密集型工作，异步无益——请考虑 multiprocessing 或 concurrent.futures。想要进一步学习，请查阅 asyncio 文档，并考虑使用 FastAPI 构建异步 Web API。 

如果您希望在此基础上学习如何设计智能应用，务必查看 Associate AI Engineer for Developers 职业路径。