别再用 for 循环串行执行了！Python asyncio 异步编程实战（运维自用版）

凌晨 4 点，我盯着屏幕上那个跑了快 20 分钟的 Python 脚本，咖啡喝了三杯，困得要死。脚本干啥呢？检查 80 台服务器的网络连通性，每台超时设 5 秒，串行跑的。

当时我就想，这要是哪天服务器数量翻到 200 台，我不得跑到天亮？

后来同事看了我的代码，就说了句：你咋不用 asyncio？我一愣，asyncio 是啥？查了一晚上资料，代码重构之后，同样的活，40 秒干完了。

提速 30 倍。

从那之后，我就把异步编程当成自己写脚本的"标配"了。不管是批量查 API、并发执行 SSH 命令、还是拉一堆监控数据，只要涉及到"等待"，我第一反应就是能不能异步。

今天这篇文章不讲那些虚头巴脑的理论（什么事件循环原理、底层 epoll 机制），就讲怎么用、用在哪、踩过什么坑。看完你也能把自己脚本的效率拉起来，少加几个小时的班。

运维写脚本的痛，谁写谁知道

搞运维的，估计都干过这事：写个脚本批量处理一堆任务，逻辑也不复杂，就是循环里调个 API 或者连一下服务器，结果等得花儿都谢了。

常见的场景我列一下，你看看中了几条：

• 批量 ping 一堆机器看通不通
• 调云厂商 API 拉几十台 ECS 的监控数据
• 并发执行 SSH 命令收集日志
• 定时任务里要调 N 个外部接口拿数据
• 爬内部系统页面抓信息

这些场景有个共同点：每个任务都要"等"。等网络返回、等 IO 完成、等对端响应。而 Python 默认的同步代码，就是一个任务完成再做下一个。

我之前那个检查连通性的脚本，每台 5 秒超时，80 台理论最差要等 400 秒。但实际上很多机器不通要等满 5 秒，所以最后跑了将近 20 分钟。

如果用异步呢？所有请求"同时"发出去，谁先回来先处理谁，总时间就取决于最慢的那一个，而不是所有加起来。

这就是异步编程的核心价值：把"等待"的时间利用起来。

异步不是新概念，只是 Python 终于有了"正经"语法

在说代码之前，先聊点背景，避免你看着看着懵。

异步这个概念早就有了，从最早的 Node.js 到 Go 语言，异步都是标配。Python 因为历史原因（GIL、全局解释器锁），多线程一直被诟病，所以异步这个特性迟迟没搞起来。

直到 Python 3.4 出了 asyncio 模块，3.5 加入了 async/await 语法，异步编程才在 Python 这边真正能用。

简单理解 asyncio 的几个关键概念：

协程（Coroutine）：可以暂停的函数。用 async def 定义的函数就是协程函数，调用它不会立即执行，而是返回一个协程对象，得用 await 或者 asyncio.run() 之类的方式驱动它。

事件循环（Event Loop）：asyncio 的核心，负责调度所有协程。你可以理解成"大管家"，所有协程都注册到它上面，它来决定什么时候执行哪个。

任务（Task）：对协程的进一步封装，让协程能被事件循环调度执行。asyncio.create_task() 就是干这个的。

Future：表示一个还没完成的结果，通常我们不用直接操作它。

这玩意有点像餐厅服务员。同步就是你点完菜干等着，菜上完才点下一道；异步就是你点完菜服务员给你个号牌，你接着点下一道，菜好了服务员喊你。

听起来挺玄乎，下面看代码就明白了。

async/await 基础语法

最基础的异步函数长这样：

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)  # 模拟 IO 等待
    print("World")

就这么简单。async def 定义异步函数，await 后面接"可等待对象"（通常是另一个协程）。

调用的时候不能直接 hello()，那只会返回一个协程对象，函数体不会执行。得这么干：

asyncio.run(hello())  # Python 3.7+ 推荐写法

asyncio.run() 会创建事件循环，运行协程，然后关闭循环。一句话搞定。

但这有啥用？单个协程和同步没啥区别。真正的威力在并发。

真正的提速：gather 和 create_task

asyncio 里两个最常用的并发工具：asyncio.gather 和 asyncio.create_task。

先说 gather，最直观的并发方式：

import asyncio
import aiohttp  # 异步 HTTP 库

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

async def main():
    urls = [
        "https://api1.example.com",
        "https://api2.example.com",
        "https://api3.example.com",
    ]
    results = await asyncio.gather(*[fetch(url) for url in urls])
    print(results)

asyncio.gather 接受一堆协程，并发执行，等所有都完成返回一个结果列表。

这代码如果同步执行，每个请求假设 0.5 秒，三个就 1.5 秒。异步并发，几乎就是 0.5 秒多一点。

再说 create_task，这个更灵活：

async def main():
    task1 = asyncio.create_task(fetch("https://api1.example.com"))
    task2 = asyncio.create_task(fetch("https://api2.example.com"))
    task3 = asyncio.create_task(fetch("https://api3.example.com"))
  
    result1 = await task1
    result2 = await task2
    result3 = await task3

create_task 把协程包装成任务，立即扔到事件循环里调度。这样你可以在等待的时候干别的事。

gather vs create_task 的区别：

• gather 更简单，自动收集所有结果
• create_task 更灵活，可以单独控制某个任务（取消、查状态）

实际写脚本我 90% 的场景用 gather，剩下 10% 用 create_task。够用了。

实战案例：80 台服务器健康检查

光说不练假把式，我拿自己用过的真实场景讲讲。

需求是这样的：检查 80 台服务器的网络连通性，每台用 ping 或者 TCP 端口检测，要支持超时控制，最后输出哪些不通。

同步版本（不推荐）：

import socket

def check_server(host, port, timeout=3):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        sock.connect((host, port))
        sock.close()
        return True
    except:
        return False

def main():
    servers = [...]  # 80 台服务器
    failed = []
    for host, port in servers:
        if not check_server(host, port):
            failed.append((host, port))
    print(f"不通的: {failed}")

这版本最坏情况要等 80 * 3 = 240 秒。

异步版本：

import asyncio

async def check_server(host, port, timeout=3):
    try:
        # asyncio.open_connection 是异步的
        reader, writer = await asyncio.wait_for(
            asyncio.open_connection(host, port), 
            timeout=timeout
        )
        writer.close()
        await writer.wait_closed()
        return True
    except:
        return False

async def main():
    servers = [...]  # 80 台服务器
    tasks = [check_server(host, port) for host, port in servers]
    results = await asyncio.gather(*tasks)
  
    failed = [servers[i] for i, r in enumerate(results) if not r]
    print(f"不通的: {failed}")

asyncio.run(main())

这版本总时间接近 3 秒（最慢的那台的超时时间），不是 240 秒。

asyncio.wait_for 是干嘛的？ 给协程加超时。如果超时还没完成，就取消掉，避免一直卡着。

asyncio.open_connection 干啥？ 异步建立 TCP 连接。底层用的是非阻塞 IO，所以不会卡住事件循环。

看到区别了吧？同步是"排队买票"，异步是"同时开 N 个窗口一起办"。

异步里最常踩的 5 个坑

这玩意看着简单，但用起来一堆坑。我把自己踩过的列出来，你绕着走。

第一个坑：在 async 函数里调用同步的阻塞 IO

这是最常见的错误。举个例子：

import asyncio
import requests  # 同步 HTTP 库

async def fetch(url):
    response = requests.get(url)  # 错误！这会阻塞整个事件循环
    return response.text

requests 是同步库，会阻塞。阻塞期间，事件循环啥也干不了，等于异步失效。

解决办法：换成异步 HTTP 库，比如 aiohttp、httpx。

import httpx  # 支持异步的 HTTP 库

async def fetch(url):
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

第二个坑：忘记 await

async 函数返回的是协程对象，必须 await 才会真正执行。少写一个 await，程序就静悄悄不工作了。

async def main():
    result = fetch("https://api.example.com")  # 没 await，协程没执行
    print(result)  # 输出的是协程对象，不是结果

新手经常栽这上面，代码跑起来没报错，但结果不对。

第三个坑：asyncio.run() 只能调用一次

asyncio.run() 会创建新的事件循环，结束后关闭。所以一个进程里只能调一次。如果你在 Jupyter 或者已经有事件循环的环境里用，会报 RuntimeError。

解决办法：

• 普通脚本：用 asyncio.run()
• Jupyter：用 await main()（IPython 内核已经有事件循环了）
• 已有事件循环：用 asyncio.create_task()

第四个坑：CPU 密集型任务别用 asyncio

asyncio 适合 IO 密集型（网络请求、文件读写），不适合 CPU 密集型（计算、加密、压缩）。

为啥？因为异步的本质是"等待的时候干别的"，如果任务是纯计算，根本没有"等待"，异步就退化成了同步。

CPU 密集型用 multiprocessing（多进程）或者 concurrent.furures.ProcessPoolExecutor。

第五个坑：超时控制别用 sleep

有些同学想"等 N 秒"，这么写：

await asyncio.sleep(N)  # 正确
time.sleep(N)  # 错误！会阻塞事件循环

time.sleep 是同步的，会卡住整个循环，所有其他任务都得等。

进阶用法：信号量控制并发度

实战中还有个常见问题：并发数太高会把对方服务打挂。

比如批量请求某个 API，对方限流 100 QPS，你一下发 1000 个请求过去，轻则被封 IP，重则把人家服务搞挂。

asyncio 提供 Semaphore（信号量）来控制并发度：

import asyncio

semaphore = asyncio.Semaphore(10)  # 最多 10 个并发

async def fetch(url):
    async with semaphore:  # 获取信号量
        async with httpx.AsyncClient() as client:
            response = await client.get(url)
            return response.text

Semaphore(10) 意味着最多同时 10 个协程在跑，其他都得等。简单粗暴但有效。

类似的还有 asyncio.Queue，用于生产者-消费者模式。这个我后面专门写一篇文章讲，太长了今天塞不下。

异步文件 IO 怎么搞

有人问，文件操作能用异步吗？能。

Python 3.4+ 的 aiofiles 库：

import aiofiles

async def read_file(path):
    async with aiofiles.open(path, 'r') as f:
        content = await f.read()
        return content

但说实话，文件 IO 异步化收益不大。因为：

1. 本地磁盘 IO 本身就快
2. 异步文件库生态不完善，很多库还是同步的
3. 如果是网络文件（NFS、对象存储），那是另一回事

所以运维场景里，文件操作我一般还是同步。除非你读写的是 S3 之类的远程存储，那可以用 aioboto3（异步版 AWS SDK）。

协程 vs 线程 vs 进程，一张表说清楚

这三种并发方式经常被搞混，我做了个对比表，方便你根据场景选：

简单说：

• 网络请求多 → asyncio
• 等 IO + 需要兼容老库 → threading
• 大量计算 → multiprocessing

我们运维写脚本，90% 是网络 IO，所以 asyncio 几乎是首选。

我常用的异步库清单

分享下我自己常用的库，你按需取用：

HTTP 客户端：

• httpx（推荐，同时支持同步和异步）
• aiohttp（纯异步，老牌库）

Web 框架：

• FastAPI（异步框架，写接口特别爽）
• aiohttp（也能写服务端）

数据库：

• asyncpg（PostgreSQL 异步）
• aiomysql（MySQL 异步）
• motor（MongoDB 异步）
• sqlalchemy[asyncio]（ORM 异步版）

Redis：

• aioredis（已合并到 redis-py 4.0+）

SSH：

• asyncssh（异步 SSH 客户端）
• paramiko（同步，需要线程包装）

文件：

• aiofiles（异步文件 IO）

消息队列：

• aio-pika（RabbitMQ）
• aiokafka（Kafka）

把这些库装上，基本能满足运维 90% 的脚本需求。

真实案例：监控数据采集脚本

最后来个综合点的案例。需求：每隔 30 秒从云厂商 API 拉取 50 台 ECS 的 CPU、内存、磁盘数据，写到本地文件。

import asyncio
import httpx
import json
from datetime import datetime

# 信号量控制并发
semaphore = asyncio.Semaphore(20)

async def fetch_metric(instance_id, client):
    async with semaphore:
        try:
            url = f"https://api.cloud.example.com/metric/{instance_id}"
            response = await client.get(url, timeout=5)
            data = response.json()
            return {
                "instance_id": instance_id,
                "cpu": data.get("cpu"),
                "memory": data.get("memory"),
                "disk": data.get("disk"),
                "timestamp": datetime.now().isoformat()
            }
        except Exception as e:
            return {
                "instance_id": instance_id,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

async def main():
    # 读机器列表
    with open("instances.txt") as f:
        instances = [line.strip() for line in f if line.strip()]
  
    async with httpx.AsyncClient() as client:
        tasks = [fetch_metric(iid, client) for iid in instances]
        results = await asyncio.gather(*tasks, return_exceptions=True)
  
    # 写结果
    with open(f"metrics_{datetime.now():%Y%m%d_%H%M%S}.json", "w") as f:
        json.dump(results, f, indent=2)
  
    print(f"采集完成: {len(results)} 台")

while True:
    asyncio.run(main())
    asyncio.sleep(30)  # 注意：这里该用 asyncio.sleep，但外层是同步

这个脚本同步版本要跑 2-3 分钟（50 台 * 单台 3 秒），异步版本 5-10 秒搞定。

注意几个细节：

1. asyncio.gather(..., return_exceptions=True) 让单个任务失败不影响整体
2. 用 Semaphore 控制并发，避免打挂云厂商 API
3. httpx.AsyncClient 用 async with 管理，自动关闭连接

调试异步代码的小技巧

异步代码调试比同步麻烦，因为栈帧不是线性的。几个小技巧：

1. 用 logging 而不是 print

import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def fetch(url):
    logger.info(f"开始请求 {url}")
    ...

2. 启用 asyncio debug 模式

asyncio.run(main(), debug=True)

debug 模式会打印慢回调、未等待的协程、未处理的异常等警告。

3. 用 aiomonitor 看运行时状态

import aiomonitor

async def main():
    monitor = aiomonitor.start_aiomonitor()
    ...

aiomonitor 起一个交互式 shell，可以查看当前所有任务、事件循环状态，还能执行 Python 代码。调试神器。

写在最后

写到最后，回头看看，其实 asyncio 没那么神秘。它就是 Python 给我们的一个工具，让我们写 IO 密集型脚本时能更高效。

我个人建议的学习路径：

1. 先用 async/await 写个简单的并发请求
2. 学会 gather 和 wait_for
3. 理解 Semaphore 控制并发
4. 接触异步库（aiohttp、httpx）
5. 进阶：异步上下文、异步生成器

这玩意 80% 的场景用 20% 的特性就够，别一上来就啃源码、看实现。

少加点班，从写好脚本开始。

如果这篇文章对你有帮助，别忘了点赞、转发，让更多运维兄弟看到。

下期我打算写写 asyncssh 批量执行 SSH 命令的实战，以及怎么用异步改造我们组那套老掉牙的部署脚本。感兴趣的话，关注公众号「耕云躬行录」，第一时间收到更新。

需要文中代码合集的，公众号后台回复 asyncio 领取。

我的个人博客上也整理了一份完整的 asyncio 速查表，地址放下面了，欢迎常来逛逛：

👉 躬行笔记

关注 @耕云躬行录，回复关键词拿资料，踩过的坑、趟过的雷，都给你掰开揉碎讲清楚。咱们下篇见。