# 偽代碼
任務列表 = [ 任務1,任務2,任務3,...]
while True:
    可執(zhí)行的任務列表,已完成的任務列表 = 去任務列表中檢查所有的任務,將'可執(zhí)行'和'已完成'的任務返回
	for 就緒任務 in 已準備就緒的任務列表:
		執(zhí)行已就緒的任務
	for 已完成的任務 in 已完成的任務列表:
		在任務列表中移除 已完成的任務
	如果 任務列表 中的任務都已完成,則終止循環(huán)

# 協(xié)程的定義與狀態(tài)
import asyncio
async def my_coroutine():
    print("協(xié)程開始")
    await asyncio.sleep(1)  # 暫停點：釋放CPU，允許切換
    print("協(xié)程結(jié)束")
    return "結(jié)果"
# 協(xié)程對象（未執(zhí)行）
coro = my_coroutine()
print(type(coro))  # <class 'coroutine'>
# 必須通過事件循環(huán)執(zhí)行
async def main():
    result = await coro  # 調(diào)度執(zhí)行，等待結(jié)果
    print(result)  # 輸出：結(jié)果
asyncio.run(main())

async def task_func(name, delay):
    print(f"任務 name={name}, delay={delay} === 111")
    await asyncio.sleep(delay)
    print(f"任務 name={name}, delay={delay} === 222")
    return f"任務 {name} 完成"
async def main():
    # 創(chuàng)建任務（立即加入事件循環(huán)，開始調(diào)度）
    task1 = asyncio.create_task(task_func("A", 1))
    task2 = asyncio.create_task(task_func("B", 2))
    print("任務狀態(tài):", task1.done())  # False（未完成）
    # 等待任務完成并獲取結(jié)果
    result1 = await task1
    result2 = await task2
    print("結(jié)果:", result1, result2)  # 任務 A 完成 任務 B 完成
    print("任務狀態(tài):", task1.done())  # True（已完成）
asyncio.run(main())

async def task_func(name, delay):
    print(f"任務 name={name}, delay={delay} === 111")
    await asyncio.sleep(delay)
    print(f"任務 name={name}, delay={delay} === 222")
    return f"任務 {name} 完成"
async def main():
    # 創(chuàng)建任務（立即加入事件循環(huán)，開始調(diào)度）
    task_list = [
        asyncio.create_task(task_func("A", 1), name="task_A"),
        asyncio.create_task(task_func("B", 2), name="task_B"),
    ]
    done, pending = await asyncio.wait(task_list, timeout=None)
    # 等待任務完成并獲取結(jié)果
    print(done)
asyncio.run(main())

async def main():
    # 創(chuàng)建一個空的Future對象
    future = asyncio.Future()
    # 定義一個設(shè)置Future結(jié)果的協(xié)程
    async def set_future_result():
        await asyncio.sleep(1)
        future.set_result("Future 結(jié)果")  # 設(shè)置結(jié)果，標記為完成
    # 并發(fā)執(zhí)行：設(shè)置結(jié)果的協(xié)程 + 等待結(jié)果的操作
    asyncio.create_task(set_future_result())
    result = await future  # 等待Future完成
    print(result)  # 輸出：Future 結(jié)果
asyncio.run(main())

async def nested():
    return 42
async def main():
    # 直接調(diào)用協(xié)程不會執(zhí)行，必須用await
    result = await nested()  # 等待nested完成，獲取結(jié)果
    print(result)  # 42
asyncio.run(main())

# Python 3.7+ 推薦方式
async def main():
    await asyncio.sleep(1)
    print("完成")
asyncio.run(main())  # 自動處理事件循環(huán)的生命周期
# 低版本手動管理方式（3.6及以下）
loop = asyncio.get_event_loop()  # 獲取事件循環(huán)
try:
    loop.run_until_complete(main())  # 運行直到協(xié)程完成
finally:
    loop.close()  # 關(guān)閉循環(huán)

async def task(i):
    await asyncio.sleep(i)
    return i
async def main():
    # 并發(fā)執(zhí)行3個任務
    results = await asyncio.gather(
        task(1),
        task(2),
        task(0.5)
    )
    print(results)  # [1, 2, 0.5]（按輸入順序，而非完成順序）
asyncio.run(main())

async def faulty_task():
    raise ValueError("出錯了")
async def main():
    results = await asyncio.gather(
        faulty_task(),
        task(1),
        return_exceptions=True  # 異常會被包裝到結(jié)果中
    )
    print(results)  # [ValueError('出錯了'), 1]

async def main():
    tasks = [task(1), task(2), task(0.5)]
    # 等待所有任務完成（默認）
    done, pending = await asyncio.wait(tasks)
    print("已完成任務數(shù):", len(done))  # 3
    print("未完成任務數(shù):", len(pending))  # 0
    # 等待第一個任務完成
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print("第一個完成的任務結(jié)果:", [t.result() for t in done])  # [0.5]

import asyncio
import requests  # 同步阻塞庫
# 同步函數(shù)（阻塞）
def sync_get(url):
    return requests.get(url).status_code
async def async_get(url):
    # 獲取事件循環(huán)
    loop = asyncio.get_event_loop()
    # 提交到線程池執(zhí)行（None 表示使用默認線程池）
    future = loop.run_in_executor(
        None,  # 線程池執(zhí)行器（可選自定義）
        sync_get,  # 同步函數(shù)
        url  # 函數(shù)參數(shù)
    )
    return await future  # 等待線程池結(jié)果
async def main():
    urls = ["https://www.baidu.com", "https://www.github.com"]
    # 并發(fā)執(zhí)行同步函數(shù)的異步包裝
    results = await asyncio.gather(*[async_get(url) for url in urls])
    print("結(jié)果:", results)  # [200, 200]
asyncio.run(main())

from concurrent.futures import ThreadPoolExecutor
async def main():
    # 自定義線程池（最大10個線程）
    executor = ThreadPoolExecutor(max_workers=10)
    loop = asyncio.get_event_loop()
    # 使用自定義線程池
    future = loop.run_in_executor(executor, sync_get, "https://www.baidu.com")
    print(await future)  # 200

import asyncio
import aiohttp
from aiohttp import ClientTimeout
async def fetch(session, url, retry=3):
    """帶重試機制的異步請求"""
    timeout = ClientTimeout(total=10)  # 超時控制（10秒）
    try:
        async with session.get(url, timeout=timeout) as response:
            return {
                "url": url,
                "status": response.status,
                "length": len(await response.text())
            }
    except Exception as e:
        if retry > 0:
            print(f"請求 {url} 失敗，重試 {retry-1} 次: {e}")
            await asyncio.sleep(1)  # 重試前等待1秒
            return await fetch(session, url, retry-1)
        return {"url": url, "error": str(e)}
async def main():
    urls = [
        "https://www.baidu.com",
        "https://www.github.com",
        "https://www.python.org",
        "https://invalid.url"
    ]
    # 創(chuàng)建會話（復用連接，提高效率）
    async with aiohttp.ClientSession() as session:
        # 生成任務列表
        tasks = [fetch(session, url) for url in urls]
        # 并發(fā)執(zhí)行
        results = await asyncio.gather(*tasks)
    # 輸出結(jié)果
    for res in results:
        if "error" in res:
            print(f"{res['url']}: {res['error']}")
        else:
            print(f"{res['url']} | 狀態(tài): {res['status']} | 長度: {res['length']}")
asyncio.run(main())

import asyncio
import aiofiles
async def write_file(filename, content):
    """異步寫入文件"""
    async with aiofiles.open(filename, 'w', encoding='utf-8') as f:
        await f.write(content)  # 異步寫入
    print(f"已寫入: {filename}")
async def read_file(filename):
    """異步讀取文件"""
    async with aiofiles.open(filename, 'r', encoding='utf-8') as f:
        content = await f.read()  # 異步讀取
    return filename, content
async def main():
    # 并發(fā)寫入3個文件
    await asyncio.gather(
        write_file("file1.txt", "異步文件1"),
        write_file("file2.txt", "異步文件2"),
        write_file("file3.txt", "異步文件3")
    )
    # 并發(fā)讀取文件
    files = ["file1.txt", "file2.txt", "file3.txt"]
    results = await asyncio.gather(*[read_file(f) for f in files])
    # 打印內(nèi)容
    for name, content in results:
        print(f"{name} 內(nèi)容: {content}")
asyncio.run(main())

import asyncio
import aiomysql
async def query_db():
    # 建立異步連接
    connection = await aiomysql.connect(
        host='localhost',
        port=3306,
        user='root',
        password='password',
        db='test',
        autocommit=True
    )
    try:
        # 創(chuàng)建游標
        async with connection.cursor(aiomysql.DictCursor) as cursor:
            # 異步執(zhí)行查詢
            await cursor.execute("SELECT * FROM users LIMIT 3")
            # 異步獲取結(jié)果
            results = await cursor.fetchall()
            print("查詢結(jié)果:", results)
    finally:
        # 關(guān)閉連接
        connection.close()
asyncio.run(query_db())