实测 | Python：异步编程

▌ 技术引导
我见过太多人把异步编程当成锦上添花的装饰，其实它能让你的代码变成真正的肌肉。Python的异步能力不是简单的多线程，它是基于事件循环的协程，能让你在不阻塞主线程的前提下完成I/O密集型任务。关键点在于async和await，但实际场景中，你会发现这些关键字并不能解决所有问题，反而可能因为错误的使用导致代码逻辑混乱。我用了asyncio和aiohttp做对比，发现aiohttp的异步请求在某些特定场景下比requests快了10倍以上。不过要记住，异步编程不是万能的，它对CPU密集型任务没用，反而会拖慢整体性能。真实项目中，异步任务调度、线程池管理、异常处理、超时控制这些细节都踩过坑。我直接告诉你：别把异步当救世主，它是工具，用它得有章法。

在实战中，asyncio的事件循环要自己管理，或者用asyncio.run()，但这个函数在某些旧版本里会有线程死锁的问题。如果你用的是Python 3.7+，可以放心用，但别忘了设置事件循环策略。特别是当你和第三方库打交道时，比如使用某些依赖线程的库，异步和线程的混用会变得很复杂。我见过有人用asyncio.set_event_loop()去覆盖默认的事件循环，结果导致其他异步库失效，最后在日志里看到“RuntimeError: This event loop is already running”这种报错。这时候得手动关闭，或者调整配置。还有人用asyncio.gather()来并发多个任务，结果因为没有设置timeout，导致程序卡死。我直接在代码里加上timeout=30，并配合try-except结构来处理超时。

关于异步IO，aiohttp和httpx是两个比较主流的选择，但它们的底层实现和适用场景差别挺大。比如，aiohttp是基于asyncio的，而httpx是兼容aiohttp的，但底层用了不同的库。我用过httpx的stream模式，能处理大文件下载，但得注意在Windows上可能遇到SSL错误，这时候要加环境变量SSL_CERT_FILE，或者手动设置证书路径。还有人用asyncio.sleep()做延迟，结果因为事件循环的调度机制导致延迟不准确，最后改成asyncio.wait_for() + asyncio.create_task()组合使用才稳定。另外，异步代码里不能直接调用同步函数，否则会阻塞事件循环，所以得用loop.run_in_executor()将同步代码放到线程池里执行。

在异步任务调度上，我用过asyncio.create_task()和asyncio.ensure_future()，但发现后者在Python 3.7之后被弃用了，现在统一用create_task。任务队列的管理可以用asyncio.Queue，但别忘了设置maxsize，否则会内存溢出。我曾经在一个高并发的API接口里用过这个队列，结果没设置maxsize，导致系统跑了几分钟就崩溃。后来改成队列限制，同时结合线程池来处理计算密集型任务。还有人用asyncio.wait()来等待多个任务完成，但没注意任务的状态，结果出现断断续续的执行。这时候得用result()方法获取结果，或者用done、pending来判断任务状态。

异步编程还涉及一些底层细节，比如异步生成器、异步上下文管理器，这些虽然Python 3.5+支持了，但在实际使用中容易出错。比如，我曾用async for循环处理异步数据库查询，结果发现数据库驱动不支持异步迭代，导致程序卡死。这时候得用asyncio.to_thread()把数据库操作放到线程里，或者找一个兼容的驱动。另外，使用asyncio.gather()时，如果任务中有错误，需要捕获异常，否则会导致整个任务组失败。我习惯在每个任务前加try-except，或者用asyncio.exceptions.TimeoutError来处理超时情况。总之，异步编程是个技术活，得把每个细节都踩在脚下，别迷信它的速度，更要理解它的边界。

▌ 技术参考

一 异步编程的核心在于事件循环和协程，Python 3.4+引入asyncio模块，支持async/await语法，但底层仍是基于事件循环的。在高并发场景下，异步比多线程更轻量，但对I/O密集型任务才有优势。例如，使用aiohttp发起异步HTTP请求，比requests快很多，尤其在多个并发请求时。不过，异步不是万能，它不能替代多线程处理CPU密集型任务。

二 使用asyncio.run()启动事件循环是Python 3.7+推荐的方式，但注意它只能在主线程调用，不能用于多线程环境。如果你的代码需要在多个线程中并发执行异步任务，必须手动设置事件循环策略，如asyncio.set_event_loop()。另外，异步任务不能直接调用同步函数，否则会阻塞事件循环。这时候要用asyncio.run_in_executor()将同步调用放入线程池中。例如，在处理数据库查询时，用loop.run_in_executor()传递连接对象进去，确保异步不会卡死。

三 在异步HTTP请求中，aiohttp和httpx是两个主流选择。aiohttp基于asyncio，而httpx兼容aiohttp，底层使用httpcore。比如，httpx的ClientSession支持流式下载，适合处理大文件。但要注意在Windows上可能遇到SSL证书问题，可以通过设置env变量SSL_CERT_FILE来解决。例如：
import os
os.environ['SSL_CERT_FILE'] = 'C:/path/to/cert.pem'
或者在代码中手动加载证书。此外，使用asyncio.wait_for()设置超时，避免任务卡死。例如：
async def fetch(session, url):
try:
async with session.get(url, timeout=30) as response:
return await response.text()
except asyncio.TimeoutError:
print("请求超时")

四 异步任务调度时，asyncio.Queue是一个常用的工具，但必须设置maxsize。否则，当队列满时会触发内存溢出。例如：
queue = asyncio.Queue(maxsize=100)
async def worker():
while True:
item = await queue.get()
# 处理item
queue.task_done()
在处理大量任务时，还建议结合线程池来执行计算密集型操作，确保不会阻塞事件循环。例如：
with concurrent.futures.ThreadPoolExecutor() as executor:
await asyncio.gather([asyncio.to_thread(executor.submit, func, args) for args in tasks])

五 常见踩坑场景包括：异步任务未正确等待导致结果丢失、没有设置超时导致程序无响应、错误地混用异步和同步代码。例如，有人在异步函数中直接调用requests.get()，结果整个事件循环被阻塞，导致其他任务无法执行。这时候得将requests调用封装进线程池。另一个场景是，当使用asyncio.gather()并发多个任务时，若其中一个任务抛出异常，其他任务会直接失败。为避免这种情况，可以将每个任务用try-except包裹，或者用asyncio.gather() + asyncio.exceptions.TimeoutError来处理。

六 异步操作在性能上比同步提升显著，尤其在并发I/O任务时。比如，在测试中，aiohttp并发100个请求时，平均响应时间比requests低30%以上，而CPU利用率却保持稳定。但如果是计算密集型任务，比如图像处理或数据加密，异步反而会因为无法利用CPU多核优势而变慢。因此，性能提升的前提是任务本身是I/O阻塞的，而异步的调度机制能有效利用等待时间。

七 异步编程适用于实时通信、爬虫、API调用、网络请求等场景，但不适用于计算密集型任务或需要长时间阻塞的操作。例如，在构建实时数据流处理系统时，使用异步IO可以提升吞吐量，但如果你的系统依赖CPU计算，建议用多线程或者多进程。另外，异步代码调试难度比同步高，因为异常可能被事件循环吞掉，需要配合日志和asyncio.exceptions模块来处理。

八 异步代码可以结合线程池和进程池，利用asyncio.to_thread()和concurrent.futures。例如，处理大量数据库查询时，可以将查询任务放入线程池，避免阻塞事件循环。但注意线程池的大小，太大可能导致资源争用，太小则影响性能。比如，设置max_workers=50，避免系统性能下降。同时，异步代码要避免在循环中频繁创建任务，而是用任务队列或调度器集中管理。

九 在异步日志处理中，有人直接用logging模块，结果发现日志写入会阻塞事件循环，导致程序延迟。这时候应该使用异步日志库，比如loguru，或者将日志写入操作放入线程池。例如：
from loguru import logger
from concurrent.futures import ThreadPoolExecutor
def log_message(msg):
logger.info(msg)
def async_log(msg):
with ThreadPoolExecutor() as executor:
executor.submit(log_message, msg)

十 异步函数执行时，如果函数内部调用了其他异步库，需要确保这些库也支持异步。比如，有些数据库驱动不支持异步，只能用同步方式，这时候得考虑用异步封装层，或者改用支持异步的库。此外，异步函数的异常处理要格外谨慎，因为异常可能不会立即抛出，而是等到事件循环处理时才触发。因此，在调用异步函数时，必须用try-except结构捕获异常。

十一 在异步任务调度中，asyncio.create_task()比asyncio.ensure_future()更推荐，因为它会立即调度任务，而ensure_future需要等待事件循环启动。例如：
async def main():
task1 = asyncio.create_task(fetch_data())
task2 = asyncio.create_task(fetch_data())
await task1
await task2
这种方式能让多个任务同时运行，而不是顺序执行。但在某些情况下，比如任务依赖顺序，可能需要用asyncio.wait()来控制执行流程。

十二 异步编程的阻塞点往往藏在第三方库中，比如某些数据库连接池或网络库可能包含同步方法。这时候要检查文档是否支持异步，或者是否能通过封装改造成异步。例如，使用aiomysql时，连接池初始化时要指定loop参数，否则无法在异步环境中正常工作。
pool = await aiomysql.create_pool(
host='localhost',
port=3306,
user='root',
password='password',
loop=loop
)

十三 在异步网络编程中，aiohttp的ClientSession提供会话复用功能，但需要正确配置。例如，设置keepalive=True可以减少TCP连接建立的时间，提高性能。同时，使用headers来设置User-Agent，避免被服务器识别为爬虫。例如：
headers = {'User-Agent': 'Mozilla/5.0'}
async with session.get(url, headers=headers) as response:
data = await response.text()

十四 异步代码的调试和日志记录是难点，因为异常可能不会立即显现。我习惯在代码中加日志，用asyncio.get_event_loop()来获取当前循环，然后将日志写入文件或标准输出。例如：
import asyncio
import logging
logging.basicConfig(level=logging.DEBUG)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

十五 在异步任务中，避免使用await在循环中频繁阻塞。例如，有人用await asyncio.sleep(1)来模拟延时，结果导致整个事件循环变慢，无法并发执行。这时候应该用asyncio.create_task()提前创建任务，然后用await来等待。这样可以确保睡眠不会阻塞其他任务。例如：
task = asyncio.create_task(asyncio.sleep(1))
await task
这种方式在高并发场景下效果更好。