import time

def sync_task():
    time.sleep(1)  # 阻塞線程
    print("Sync task done")

start = time.time()
sync_task()
sync_task()
print(f"總耗時: {time.time() - start:.2f}s")  # 輸出約 2.0s

import asyncio

async def async_task():
    await asyncio.sleep(1)  # 非阻塞掛起
    print("Async task done")

async def main():
    start = time.time()
    await asyncio.gather(async_task(), async_task())
    print(f"總耗時: {time.time() - start:.2f}s")  # 輸出約 1.0s

asyncio.run(main())

Created → Pending → Running → Suspended (at await) → Resumed → Completed/Failed

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://www.example.com",
        "https://www.python.org",
        "https://www.github.com"
    ]
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for url, content in zip(urls, results):
        print(f"{url} 內(nèi)容長度: {len(content)}")

asyncio.run(main())

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import asyncio

app = FastAPI()

async def background_task():
    while True:
        await asyncio.sleep(5)
        print("后臺任務運行中...")

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(background_task())

@app.get("/", response_class=HTMLResponse)
async def read_root():
    await asyncio.sleep(1)  # 模擬異步數(shù)據(jù)庫查詢
    return "<h1>Hello Async World!</h1>"

import websockets
import asyncio

connected = set()

async def chat_server(websocket):
    connected.add(websocket)
    try:
        async for message in websocket:
            await asyncio.gather(
                *[client.send(f"用戶說: {message}") for client in connected]
            )
    finally:
        connected.remove(websocket)

async def main():
    async with websockets.serve(chat_server, "localhost", 8765):
        await asyncio.Future()  # 永久運行

asyncio.run(main())

lock = asyncio.Lock()
async def safe_update():
    async with lock:
        # 修改共享資源

async def risky_task():
    try:
        await async_operation()
    except Exception as e:
        print(f"錯誤捕獲: {e}")

async def main():
    task = asyncio.create_task(risky_task())
    await task

sem = asyncio.Semaphore(10)
async def limited_task():
    async with sem:
        await heavy_operation()