import asyncio
import time

async def say_hello(name, delay):
    print(f"開(kāi)始執(zhí)行 {name}")
    await asyncio.sleep(delay)
    print(f"Hello, {name}!")
    return f"完成 {name}"

async def main():
    # 創(chuàng)建任務(wù)
    task1 = asyncio.create_task(say_hello("Alice", 2))
    task2 = asyncio.create_task(say_hello("Bob", 1))
    
    # 等待任務(wù)完成
    result1 = await task1
    result2 = await task2
    
    print(result1)
    print(result2)

# 運(yùn)行主函數(shù)
asyncio.run(main())

import asyncio

async def fetch_data(url, delay):
    print(f"開(kāi)始獲取 {url}")
    await asyncio.sleep(delay)
    return f"數(shù)據(jù)來(lái)自 {url}"

async def main():
    # 創(chuàng)建多個(gè)任務(wù)
    tasks = [
        asyncio.create_task(fetch_data("https://api1.com", 1)),
        asyncio.create_task(fetch_data("https://api2.com", 2)),
        asyncio.create_task(fetch_data("https://api3.com", 1.5))
    ]
    
    # 等待所有任務(wù)完成
    results = await asyncio.gather(*tasks)
    
    for result in results:
        print(result)

asyncio.run(main())

import asyncio

async def long_running_task():
    try:
        for i in range(10):
            print(f"任務(wù)執(zhí)行中... {i}")
            await asyncio.sleep(1)
        return "任務(wù)完成"
    except asyncio.CancelledError:
        print("任務(wù)被取消")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    
    # 3秒后取消任務(wù)
    await asyncio.sleep(3)
    task.cancel()
    
    try:
        await task
    except asyncio.CancelledError:
        print("捕獲到取消異常")

asyncio.run(main())

import asyncio

async def sample_task():
    await asyncio.sleep(2)
    return "完成"

async def main():
    task = asyncio.create_task(sample_task())
    
    print(f"任務(wù)創(chuàng)建后 - 完成: {task.done()}")
    
    await asyncio.sleep(1)
    print(f"等待1秒后 - 完成: {task.done()}")
    
    await task
    print(f"任務(wù)完成后 - 完成: {task.done()}")
    print(f"任務(wù)結(jié)果: {task.result()}")

asyncio.run(main())

import asyncio

async def task_with_exception():
    await asyncio.sleep(1)
    raise ValueError("這是一個(gè)錯(cuò)誤")

async def normal_task():
    await asyncio.sleep(2)
    return "正常完成"

async def main():
    task1 = asyncio.create_task(task_with_exception())
    task2 = asyncio.create_task(normal_task())
    
    try:
        # 使用 gather 并設(shè)置 return_exceptions=True 來(lái)處理異常
        results = await asyncio.gather(task1, task2, return_exceptions=True)
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"任務(wù) {i} 發(fā)生異常: {result}")
            else:
                print(f"任務(wù) {i} 結(jié)果: {result}")
    except Exception as e:
        print(f"捕獲異常: {e}")

asyncio.run(main())

import asyncio
import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url) as response:
            return await response.text()
    except Exception as e:
        return f"錯(cuò)誤: {e}"

async def fetch_multiple_urls():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/2',
        'https://httpbin.org/delay/1'
    ]
    
    async with aiohttp.ClientSession() as session:
        # 創(chuàng)建任務(wù)列表
        tasks = [asyncio.create_task(fetch_url(session, url)) for url in urls]
        
        # 并發(fā)執(zhí)行所有任務(wù)
        results = await asyncio.gather(*tasks)
        
        for i, result in enumerate(results):
            print(f"URL {i+1}: {len(result) if isinstance(result, str) else result} 字符")

# 注意：需要安裝 aiohttp: pip install aiohttp
# asyncio.run(fetch_multiple_urls())

import asyncio
import aiohttp


async def process_http_request():
    # 使用aiohttp進(jìn)行異步HTTP請(qǐng)求
    timeout = aiohttp.ClientTimeout(total=1000)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        headers = {}
        data = {}
        url = ""
        is_error = False
        try:
            print(f"HTTP異步請(qǐng)求開(kāi)始，data: {data} ")
            async with session.post(
                    url=url,
                    headers=headers,
                    json=data,
                    timeout=timeout  # 為單個(gè)請(qǐng)求設(shè)置超時(shí)
            ) as response:
                if response.status == 200:
                    # 獲取響應(yīng)數(shù)據(jù)
                    response_data = await response.json()
                    if response_data['data']['status'] != 200:
                        print(f"HTTP異步請(qǐng)求處理失敗: {response_data} ")
                        is_error = True
                else:
                    print(f"HTTP異步請(qǐng)求處理失敗，狀態(tài)碼: {response.status}")
                    is_error = True
        except asyncio.TimeoutError:
            print(f"HTTP異步請(qǐng)求超時(shí)!")
            is_error = True

    print(f"HTTP異步請(qǐng)求完成")
    return is_error

import asyncio
# import asyncpg  # 示例使用 asyncpg

async def query_database(query_id):
    # 模擬數(shù)據(jù)庫(kù)查詢
    await asyncio.sleep(1)  # 模擬查詢時(shí)間
    return f"查詢結(jié)果 {query_id}"

async def concurrent_queries():
    # 創(chuàng)建多個(gè)查詢?nèi)蝿?wù)
    tasks = [
        asyncio.create_task(query_database(i)) 
        for i in range(5)
    ]
    
    # 并發(fā)執(zhí)行查詢
    results = await asyncio.gather(*tasks)
    
    for result in results:
        print(result)

asyncio.run(concurrent_queries())

# 舊方式（仍然可用）
task = asyncio.ensure_future(coro)

# 新方式（推薦）
task = asyncio.create_task(coro)

import asyncio

async def good_practice():
    # 好的做法：明確創(chuàng)建任務(wù)
    task1 = asyncio.create_task(some_coroutine())
    task2 = asyncio.create_task(another_coroutine())
    
    # 等待結(jié)果
    result1 = await task1
    result2 = await task2
    
    return result1, result2

async def avoid_this():
    # 避免這樣做：直接await協(xié)程不會(huì)并發(fā)執(zhí)行
    result1 = await some_coroutine()
    result2 = await another_coroutine()
    return result1, result2

import asyncio

async def slow_operation():
    await asyncio.sleep(5)
    return "完成"

async def main():
    try:
        task = asyncio.create_task(slow_operation())
        # 設(shè)置3秒超時(shí)
        result = await asyncio.wait_for(task, timeout=3.0)
        print(result)
    except asyncio.TimeoutError:
        print("操作超時(shí)")
        # 取消任務(wù)
        task.cancel()

asyncio.run(main())

import asyncio

async def background_task(name):
    print(f"任務(wù) {name} 開(kāi)始")
    await asyncio.sleep(2)
    print(f"任務(wù) {name} 完成")
    return f"結(jié)果 {name}"

async def main():
    # 創(chuàng)建任務(wù)但不await - 任務(wù)會(huì)自動(dòng)運(yùn)行
    task1 = asyncio.create_task(background_task("1"))
    task2 = asyncio.create_task(background_task("2"))
    
    print("任務(wù)已創(chuàng)建")
    
    # 做一些其他工作
    await asyncio.sleep(1)
    print("主協(xié)程繼續(xù)執(zhí)行")
    
    # 等待任務(wù)完成獲取結(jié)果
    result1 = await task1
    result2 = await task2
    
    print(f"結(jié)果: {result1}, {result2}")

# 運(yùn)行主函數(shù)
asyncio.run(main())