跳到主要内容

异步编程基础

asyncio 提供单线程协作式并发：在等待 I/O 时切换协程，适合大量 HTTP/API 调用且希望控制连接数。配合 httpx 的异步客户端，可替代线程池处理千级 URL 探活，内存开销更低。

核心概念​

概念说明
coroutineasync def 定义的协程函数
await挂起当前协程，让出事件循环
event loop调度协程运行的核心
Task被调度执行的协程包装

第一个异步探活脚本​

#!/usr/bin/env python3
"""asyncio + httpx 并发探活多个 URL。"""

import asyncio
import time

import httpx


async def probe(client: httpx.AsyncClient, url: str) -> dict:
    start = time.perf_counter()
    try:
        resp = await client.get(url, timeout=3.0)
        ok = resp.status_code < 400
        err = None
    except httpx.HTTPError as e:
        ok = False
        err = str(e)
    elapsed = round((time.perf_counter() - start) * 1000, 1)
    return {"url": url, "ok": ok, "elapsed_ms": elapsed, "error": err}


async def main() -> None:
    urls = [
        "https://gitlab.example.com/health",
        "https://harbor.example.com/api/v2.0/health",
        "https://nexus.example.com/service/rest/v1/status",
    ]

    # 共享 AsyncClient，复用连接池
    async with httpx.AsyncClient(follow_redirects=True) as client:
        # asyncio.gather 并发执行多个协程
        results = await asyncio.gather(*[probe(client, u) for u in urls])

    for r in results:
        flag = "OK" if r["ok"] else "FAIL"
        print(f"[{flag}] {r['url']}  {r['elapsed_ms']}ms")


if __name__ == "__main__":
    asyncio.run(main())  # Python 3.7+ 推荐入口

依赖：uv add httpx

Semaphore 限制并发​

无限制 gather 可能压垮目标或本地文件描述符：

async def probe_limited(
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore,
    url: str,
) -> dict:
    async with sem:  # 同时最多 N 个在飞请求
        return await probe(client, url)


async def batch_probe(urls: list[str], concurrency: int = 20) -> list[dict]:
    sem = asyncio.Semaphore(concurrency)
    async with httpx.AsyncClient() as client:
        tasks = [probe_limited(client, sem, u) for u in urls]
        return await asyncio.gather(*tasks)

异常处理与 return_exceptions​

async def safe_gather(urls: list[str]) -> None:
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(
            *[probe(client, u) for u in urls],
            return_exceptions=True,  # 异常作为结果返回，不中断其他任务
        )
    for url, res in zip(urls, results):
        if isinstance(res, Exception):
            print(f"FAIL {url}: {res}")
        else:
            print(f"OK   {url}: {res['elapsed_ms']}ms")

与阻塞库混用​

阻塞调用（SSH、subprocess）不能 await，用 loop.run_in_executor(None, blocking_fn, *args) 丢进线程池执行。

选型对比​

方案适用
单线程顺序主机数 < 10，逻辑简单
ThreadPoolExecutorSSH、混合阻塞 I/O、团队更熟悉线程
asyncio + httpx大量 HTTP/API、需精细控制连接数
注意
  • asyncio 脚本全程 async，避免在协程内调用 time.sleep()，应使用 await asyncio.sleep()
  • Jupyter 等环境可能已有 event loop，入口方式不同，运维脚本用 asyncio.run() 即可
  • 异步代码调试较难，先用小批量验证再放大

小结​

  • async def + await 实现协作式并发
  • httpx.AsyncClient 复用连接，适合 API 批量调用
  • Semaphore 控制并发上限
  • 阻塞库用 run_in_executor 桥接

第六阶段完结。下一模块学习 FastAPI，将巡检能力暴露为 HTTP 服务。