Python事件循环教程：如何使用asyncio事件循环 | Python异步编程指南

http://zb.zfpaper.net   
http://zhibo.zfpaper.net    
http://jrs.zfpaper.net  
http://tv.zfpaper.net   
http://qq.zfpaper.net

import asyncio

# 定义一个简单的协程
async def hello_world():
    print("Hello")
    await asyncio.sleep(1)  # 模拟I/O操作
    print("World")

# 获取事件循环
loop = asyncio.get_event_loop()

try:
    # 运行协程直到完成
    loop.run_until_complete(hello_world())
finally:
    # 关闭事件循环
    loop.close()

import asyncio

async def task(name, seconds):
    print(f"Task {name} started")
    await asyncio.sleep(seconds)
    print(f"Task {name} completed after {seconds} seconds")

async def main():
    # 创建多个任务
    tasks = [
        task("A", 2),
        task("B", 1),
        task("C", 3)
    ]
    
    # 同时运行所有任务
    await asyncio.gather(*tasks)

# Python 3.7+ 使用更简洁的方式运行
asyncio.run(main())

import asyncio
import time

def blocking_io():
    print("阻塞操作开始")
    time.sleep(2)  # 模拟耗时同步操作
    print("阻塞操作完成")

async def main():
    loop = asyncio.get_running_loop()
    
    # 在默认线程池中执行阻塞函数
    await loop.run_in_executor(None, blocking_io)
    print("继续执行异步代码")

asyncio.run(main())

import asyncio

async def periodic_task():
    while True:
        print("执行周期性任务")
        await asyncio.sleep(5)  # 每5秒执行一次

async def main():
    task = asyncio.create_task(periodic_task())
    
    # 运行10秒后取消任务
    await asyncio.sleep(10)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("周期性任务已取消")

asyncio.run(main())

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = [
        'https://example.com',
        'https://python.org',
        'https://github.com'
    ]
    
    # 同时获取所有URL
    pages = await asyncio.gather(
        *[fetch(url) for url in urls]
    )
    
    for url, content in zip(urls, pages):
        print(f"{url}: {len(content)} bytes")

asyncio.run(main())

import asyncio
import asyncpg

async def query_db():
    conn = await asyncpg.connect(
        user='user', 
        password='password',
        database='database',
        host='localhost'
    )
    
    # 执行异步查询
    result = await conn.fetch("SELECT * FROM users")
    for record in result:
        print(record)
    
    await conn.close()

asyncio.run(query_db())