Python 异步编程利用非阻塞机制提升 I/O 密集型任务执行效率。内容涵盖协程定义、事件循环原理及任务调度策略。深入讲解 asyncio 库常用函数（run、create_task、gather、wait_for）及 Future 对象使用。结合 aiohttp 库演示异步 HTTP 客户端与服务器的构建，包含 GET/POST/JSON 请求处理及静态文件服务。通过并发请求实战案例展示性能优化方法，提供完整代码示例与实施步骤，适用于高并发网络应用开发场景。
异步编程是一种并发编程方式，通过非阻塞操作提高程序执行效率。在异步编程中，程序可以在等待 I/O 操作完成时继续执行其他任务，而不需要阻塞等待。
协程（Coroutine）是一种轻量级的线程，可以在程序中进行暂停和恢复。在 Python 中，协程可以通过 async def 关键字定义。
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
asyncio.run(hello())
import asyncio
async def count():
print('Counting...')
await asyncio.sleep(1)
print('Counted!')
async def main():
await asyncio.gather(count(), count(), count())
asyncio.run(main())
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
async def main():
task1 = asyncio.create_task(hello())
task2 = asyncio.create_task(hello())
await task1
await task2
asyncio.run(main())
import asyncio
async def hello():
try:
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
except asyncio.CancelledError:
print('Task cancelled!')
async def main():
task = asyncio.create_task(hello())
await asyncio.sleep(0.5)
task.cancel()
try:
await task
except asyncio.CancelledError:
print('Main: Task cancelled!')
asyncio.run(main())
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(2)
print('Hello again!')
async def main():
try:
await asyncio.wait_for(hello(), timeout=1)
except asyncio.TimeoutError:
print('Task timed out!')
asyncio.run(main())
事件循环是异步编程的核心组件，负责调度任务的执行。事件循环会不断地从任务队列中取出任务并执行，直到任务队列为空。
import asyncio
loop = asyncio.get_event_loop()
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
loop.run_until_complete(hello())
import asyncio
loop = asyncio.get_event_loop()
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
loop.run_until_complete(asyncio.gather(hello(), hello(), hello()))
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
asyncio.run(hello())
import asyncio
async def hello():
print('Hello, World!')
await asyncio.sleep(1)
print('Hello again!')
async def main():
future = asyncio.Future()
task = asyncio.create_task(hello())
task.add_done_callback(lambda t: future.set_result(t.result()))
await future
asyncio.run(main())
pip install aiohttp
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
html = await fetch(session, 'https://www.example.com')
print(html)
asyncio.run(main())
import aiohttp
import asyncio
import json
async def post_data(session, url, data):
async with session.post(url, data=data) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
data = {'name': '张三', 'age': 25}
response = await post_data(session, 'https://httpbin.org/post', data)
print(response)
asyncio.run(main())
import aiohttp
import asyncio
import json
async def post_json(session, url, data):
async with session.post(url, json=data) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
data = {'name': '张三', 'age': 25}
response = await post_json(session, 'https://httpbin.org/post', data)
print(response)
asyncio.run(main())
开发一个异步 HTTP 客户端，支持以下功能：
import aiohttp
import asyncio
import time
async def fetch(session, url):
start_time = time.time()
async with session.get(url) as response:
text = await response.text()
elapsed_time = time.time() - start_time
return url, len(text), elapsed_time
async def main():
urls = ['https://www.example.com', 'https://www.google.com', 'https://www.github.com', 'https://www.python.org', 'https://www.djangoproject.com']
async with aiohttp.ClientSession() as session:
tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
results = await asyncio.gather(*tasks)
for url, length, elapsed_time in results:
print(f'URL: {url}, 响应长度：{length}, 耗时：{elapsed_time:.2f}秒')
if __name__ == '__main__':
start_time = time.time()
asyncio.run(main())
elapsed_time = time.time() - start_time
print(f'总耗时：{elapsed_time:.2f}秒')
开发一个异步 Web 服务器，支持以下功能：
from aiohttp import web
import asyncio
async def handle_root(request):
return web.Response(text='Hello, World!')
async def handle_api(request):
data = {'name': '张三', 'age': 25}
return web.json_response(data)
async def handle_static(request):
return web.FileResponse('static/index.html')
async def create_app():
app = web.Application()
app.add_routes([
web.get('/', handle_root),
web.get('/api', handle_api),
web.get('/static/{name}', handle_static)
])
return app
if __name__ == '__main__':
asyncio.run(web.run_app(create_app(), host='localhost', port=8080))
本文涵盖了 Python 异步编程的核心概念，包括协程、任务调度及事件循环机制。通过 asyncio 和 aiohttp 库的实战演示，展示了如何构建高效的异步 HTTP 客户端与 Web 服务器。开发者可通过上述代码示例理解高并发场景下的网络应用开发模式。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online