Python 异步编程与协程
Python
Python 异步编程与协程实战指南
Python 异步编程利用非阻塞 I/O 提升执行效率。文章讲解协程定义、任务调度、事件循环机制及 asyncio 库核心函数。结合 aiohttp 库演示 GET/POST/JSON 请求发送及异步 Web 服务器搭建。实战案例涵盖并发 HTTP 客户端开发与响应处理，包含超时控制与任务取消逻辑，适用于高并发网络应用开发场景。
Python 异步编程利用非阻塞 I/O 提升执行效率。文章讲解协程定义、任务调度、事件循环机制及 asyncio 库核心函数。结合 aiohttp 库演示 GET/POST/JSON 请求发送及异步 Web 服务器搭建。实战案例涵盖并发 HTTP 客户端开发与响应处理，包含超时控制与任务取消逻辑，适用于高并发网络应用开发场景。
异步编程是一种并发编程方式，通过非阻塞操作提高程序执行效率。在等待 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
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_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())
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']
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}秒')
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 create_app():
app = web.Application()
app.add_routes([
web.get('/', handle_root),
web.get('/api', handle_api)
])
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