​
你的爬虫程序又卡住了。

不是代码写错了，是请求发得太慢。一万条数据要爬，一条一秒，两个多小时才能跑完。你试着用多线程，结果IP被封得更快了。线程开太多，系统资源也扛不住。你听说过异步，听说它能用单线程搞定高并发，但看着async和await这两个关键字，总觉得像在写另一门语言。

这种感觉不怪你。异步编程确实和Python其他部分不太一样，但一旦跨过那道坎，你会发现它没有想象中那么难。

异步到底在解决什么问题

先看一个很简单的程序：

import time

def task(name, seconds):
print(f"{name} 开始")
time.sleep(seconds)
print(f"{name} 结束")

start = time.time()
task("任务1", 2)
task("任务2", 2)
print(f"总耗时: {time.time() - start}")

运行结果是：任务1开始、等待两秒、任务1结束、任务2开始、再等两秒、任务2结束，总耗时4秒。

问题出在time.sleep这里。当程序执行到sleep时，CPU其实什么事都没干，就傻等着时间过去。这段时间本可以用来处理其他任务，但同步代码不允许这样做——它是一条路走到黑的，没执行完当前函数，绝不会跳到下一行。

异步要解决的就是这个“等待浪费”的问题。当某个操作需要等待时，让程序暂时离开，去做别的事，等那个操作准备好了再回来继续。

async和await到底在干什么

Python的异步编程基于一个叫做“事件循环”的东西。你可以把它理解成一个调度中心，管理着所有待执行的任务。当一个任务遇到需要等待的操作时，它会告诉事件循环：“我先去忙别的，好了叫我”，然后事件循环就会切换到下一个可以执行的任务。

async def用来定义一个异步函数，也叫协程。它和普通函数的区别在于，调用它并不会立即执行，而是返回一个协程对象。

async def hello():
print("Hello")
await asyncio.sleep(1)
print("World")

await用来等待一个异步操作完成。当程序执行到await时，它会暂时离开这个函数，让事件循环去处理其他任务。等asyncio.sleep(1)这个操作完成了，事件循环才会回来继续执行后面的print("World")。

来看一个完整的例子：

import asyncio

async def task(name, seconds):
print(f"{name} 开始")
await asyncio.sleep(seconds)
print(f"{name} 结束")

async def main():

# 创建三个任务，但还没有执行
tasks = [
    asyncio.create_task(task("任务1", 2)),
    asyncio.create_task(task("任务2", 1)),
    asyncio.create_task(task("任务3", 3))
]
# 等待所有任务完成
await asyncio.gather(*tasks)

asyncio.run(main())

运行结果：

任务1 开始
任务2 开始
任务3 开始
任务2 结束
任务1 结束
任务3 结束
总耗时约3秒

三个任务几乎是同时开始的，总共只用了最慢那个任务的时间。这就是异步的核心价值：在等待的时间里做别的事。

事件循环是怎么运转的

理解事件循环的工作机制，能帮你避开大部分异步编程的坑。

事件循环本质上是一个无限循环，它维护着两个队列：就绪队列和等待队列。就绪队列里放着可以立即执行的任务，等待队列里放着正在等待某个事件（比如网络响应、定时器到期）的任务。

每一次循环，事件循环会从就绪队列里取出一个任务执行。当任务执行到await时，它会把自己挂起，并告诉事件循环它正在等待什么。事件循环就把这个任务放到等待队列里，然后继续处理就绪队列中的下一个任务。

当等待队列里的某个任务等的事件发生了（比如定时器到期），事件循环就会把它移回就绪队列，等待下一次被调度执行。

这就是为什么异步程序能在一个线程里实现并发：它不是同时做多件事，而是在一件事等待的时候，去做另一件事。

新手最容易踩的坑

坑一：在异步函数里用了同步阻塞操作

async def bad_example():

# 错误：用了 time.sleep 而不是 asyncio.sleep
time.sleep(5)  # 这会阻塞整个事件循环
return "done"

time.sleep是同步阻塞的，它会卡住当前线程，事件循环在这5秒内完全无法工作，所有任务都会被堵住。正确的做法是用await asyncio.sleep(5)。

同样的道理，如果你在异步代码里使用了requests库发HTTP请求，它也会阻塞事件循环。应该用aiohttp这类异步HTTP客户端。

坑二：忘记加await

async def wrong():
coro = some_async_function() # 这返回的是协程对象，不是执行结果
print(coro) # 打印

# 协程对象没有被 await，它永远不会执行

协程对象被创建后，必须被await或者被asyncio.create_task()调度，否则它不会执行。这是异步编程新手最容易忽略的地方。

坑三：在同步代码里调用异步函数

def sync_function():

# 错误：不能在同步函数里直接 await
result = await async_function()  # SyntaxError

await只能在async def定义的函数内部使用。如果你想在同步代码里执行异步函数，需要用asyncio.run()或者asyncio.create_task()配合事件循环。

坑四：创建了太多任务导致资源耗尽

错误：一下子创建几万个任务

for i in range(100000):
asyncio.create_task(heavy_task(i))
await asyncio.gather(*tasks) # 可能内存爆炸

虽然协程比线程轻量很多，但也不是无限创建的。几万个协程同时存在，内存和调度开销还是会很大。可以用信号量（Semaphore）来控制并发数量。

控制并发数的正确姿势

在实际项目中，你通常不会把所有任务一次性丢进事件循环。以爬虫为例，同时发几百个请求，目标网站可能直接把你IP封了，你自己的网络也可能扛不住。

用asyncio.Semaphore可以轻松控制并发数量：

import asyncio
import aiohttp

async def fetch(session, url, semaphore):
async with semaphore: # 获取信号量，控制并发数
async with session.get(url) as response:
return await response.text()

async def main():
urls = ["http://example.com"] * 100 # 100个URL
semaphore = asyncio.Semaphore(10) # 限制最多同时10个请求

async with aiohttp.ClientSession() as session:
    tasks = [fetch(session, url, semaphore) for url in urls]
    results = await asyncio.gather(*tasks)

print(f"抓取完成，共{len(results)}个页面")

asyncio.run(main())

信号量的工作原理很简单：它内部维护一个计数器。每次async with semaphore会尝试减少计数器，如果计数器大于0，就允许进入；如果等于0，就等待直到有任务释放信号量。这样就能精确控制并发数，既充分利用资源，又不至于把对方服务器打趴下。

超时处理是保命技能

异步程序里，某个任务卡住会影响整个事件循环吗？不会，因为await是让出控制权的。但如果一个任务内部有死循环或者一直没遇到await，它确实会霸占事件循环，导致其他任务无法执行。

更常见的问题是网络请求一直不返回。这时候就需要超时控制：

import asyncio

async def fetch_with_timeout():
try:

    # 设置5秒超时
    result = await asyncio.wait_for(slow_operation(), timeout=5)
    return result
except asyncio.TimeoutError:
    print("操作超时")
    return None

async def slow_operation():
await asyncio.sleep(10) # 模拟慢操作
return "数据"

asyncio.wait_for会给一个协程加上时间限制。超时后它会抛出asyncio.TimeoutError，你可以捕获它做降级处理。

还有一个更灵活的工具是asyncio.gather的return_exceptions参数：

results = await asyncio.gather(
task1(),
task2(),
task3(),
return_exceptions=True # 异常不会中断，而是作为结果返回
)

for result in results:
if isinstance(result, Exception):
print(f"任务失败: {result}")
else:
print(f"任务成功: {result}")

这样单个任务失败不会导致整个gather崩溃，你可以优雅地处理每个任务的结果。

异步和并发的本质区别

很多人会把异步和多线程混淆，觉得它们都能“同时”做多件事。但理解它们的区别，能帮你做出更合理的技术选型。

多线程是操作系统层面的并发。每个线程都有自己的栈空间，线程切换由操作系统调度，开销较大。Python因为有GIL（全局解释器锁），多线程在CPU密集型任务上反而更慢，但在IO密集型任务上依然有用。

异步是单线程内的并发。所有协程共享同一个线程，切换发生在await的时候，开销极小。一个事件循环可以轻松处理上万协程，而开上万个线程基本不可能。

选哪个？IO密集型任务（网络请求、文件读写、数据库查询）用异步，代码更简洁，资源消耗更少。CPU密集型任务（计算、加密、压缩）用多进程，或者把计算部分交给专门的进程池，异步只负责调度。

实际项目中的异步架构

真实项目很少只用一个asyncio.run()就完事。更常见的做法是分层设计：

底层是异步IO操作，用aiohttp、aiomysql、aioredis这些异步库。中间层是业务逻辑，用async def定义协程，处理数据转换、错误重试、超时控制。最上层是调度层，管理任务队列、控制并发、监控执行状态。

一个典型的数据采集任务大概是这样的：

import asyncio
import aiohttp
from asyncio import Queue

async def worker(name, queue, session):
"""工作协程：从队列取URL，抓取页面"""
while True:
url = await queue.get()
try:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
content = await resp.text()
print(f"{name} 抓取 {url} 成功，长度 {len(content)}")

            # 这里可以存储数据
    except Exception as e:
        print(f"{name} 抓取 {url} 失败: {e}")
    finally:
        queue.task_done()

async def main():
urls = [f"http://example.com/page/{i}" for i in range(1000)]
queue = Queue()

# 把所有URL放入队列
for url in urls:
    await queue.put(url)

async with aiohttp.ClientSession() as session:
    # 启动5个工作协程
    workers = [asyncio.create_task(worker(f"worker-{i}", queue, session)) 
               for i in range(5)]

    # 等待所有任务完成
    await queue.join()

    # 取消工作协程
    for w in workers:
        w.cancel()

asyncio.run(main())

队列模式的好处是解耦了任务生产者和消费者，可以灵活调整并发数量，也方便做断点续传、失败重试这些复杂逻辑。

写在最后

异步编程刚接触时确实有点绕，但核心思想很简单：别闲着，等的时候去干点别的。async定义协程，await让出控制权，事件循环负责调度。记住这三件事，你就已经入门了。

那些坑，说到底都是忘了“异步代码里不能有阻塞操作”这条原则。遇到问题的时候，先检查是不是用了同步库，是不是忘记await了，是不是并发数设置得太高。

异步Python这几年越来越成熟。aiohttp、asyncpg、FastAPI这些生态组件已经足够支撑大型项目。如果你还在用多线程处理IO密集型任务，不妨试试异步。同样的机器资源，异步往往能扛住更高的并发，代码逻辑也更清晰。

写异步代码像是在指挥一支交响乐团——每个乐器都在自己的节奏上演奏，你不需要盯着每个人，只需要把握好整体进度。一旦适应了这种编程思维，你会发现，原来高并发也可以写得这么优雅。

​