Python asyncio核心原理与高阶应用全解析

import asyncio
async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")
# 运行异步函数
asyncio.run(main())

import asyncio
async def task1():
    await asyncio.sleep(1)
    print("Task 1 completed")
    return "Task 1 result"
async def task2():
    await asyncio.sleep(2)
    print("Task 2 completed")
    return "Task 2 result"
async def main():
    # 创建任务
    task1_obj = asyncio.create_task(task1())
    task2_obj = asyncio.create_task(task2())
    # 等待任务完成
    result1 = await task1_obj
    result2 = await task2_obj
    print(f"Results: {result1}, {result2}")
asyncio.run(main())

import asyncio
async def fetch_data(id):
    await asyncio.sleep(1)
    return f"Data {id}"
async def main():
    # 并发执行多个任务
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    print(f"Results: {results}")
asyncio.run(main())

import asyncio
async def slow_operation():
    await asyncio.sleep(2)
    return "Operation completed"
async def main():
    try:
        # 设置超时
        result = await asyncio.wait_for(slow_operation(), timeout=1)
        print(f"Result: {result}")
    except asyncio.TimeoutError:
        print("Operation timed out")
asyncio.run(main())

import asyncio
async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")
# 获取事件循环
loop = asyncio.get_event_loop()
# 运行协程
loop.run_until_complete(main())
# 关闭事件循环
loop.close()

import asyncio
import aiohttp
async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()
async def main():
    urls = [
        "https://api.github.com",
        "https://api.example.com",
        "https://api.google.com"
    ]
    # 并发请求
    tasks = [fetch_url(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for url, result in zip(urls, results):
        print(f"URL: {url}, Length: {len(result)}")
asyncio.run(main())

import asyncio
async def read_file(filename):
    async with open(filename, 'r') as f:
        content = await f.read()
    return content
async def write_file(filename, content):
    async with open(filename, 'w') as f:
        await f.write(content)
async def main():
    # 读取文件
    content = await read_file('input.txt')
    print(f"Read content: {content}")
    # 写入文件
    await write_file('output.txt', content.upper())
    print("File written")
asyncio.run(main())

import asyncio
import asyncpg
async def main():
    # 连接数据库
    conn = await asyncpg.connect(
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='test'
    )
    # 执行查询
    rows = await conn.fetch('SELECT * FROM users')
    for row in rows:
        print(row)
    # 关闭连接
    await conn.close()
asyncio.run(main())

import asyncio
from aiohttp import web
async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    return web.Response(text=f"Hello, {name}!")
async def main():
    app = web.Application()
    app.add_routes([
        web.get('/', handle),
        web.get('/{name}', handle)
    ])
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print("Server started on http://localhost:8080")
    await asyncio.Event().wait()
asyncio.run(main())