堅不可摧的 Asyncio 伺服器：深錯誤處理與優雅關閉

import asyncio
import signal

async def cancel_tasks():
    print('收到 SIGINT 訊號!')
    tasks = asyncio.all_tasks()
    print(f'正在取消 {len(tasks)} 個任務.')
    [task.cancel() for task in tasks]

async def main():
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, cancel_tasks)
    await asyncio.sleep(10)

asyncio.run(main())

import asyncio
import signal
from typing import List

class GracefulExit(SystemExit):
    pass

def shutdown():
    raise GracefulExit()

async def echo(connection, loop):
    # ... existing echo code ...
    pass  # Placeholder for existing code

async def connection_listener(server_socket, loop):
    # ... existing connection_listener code ...
    pass  # Placeholder for existing code

echo_tasks: List[asyncio.Task] = []

async def close_echo_tasks(echo_tasks):
    waiters = [asyncio.wait_for(task, 2) for task in echo_tasks]
    for task in waiters:
        try:
            await task
        except asyncio.TimeoutError:
            print("任務逾時!")
            pass

async def main():
    loop = asyncio.get_event_loop()
    # ... existing main code ...
    for signame in {'SIGINT', 'SIGTERM'}:
        loop.add_signal_handler(getattr(signal, signame), shutdown)
    await connection_listener(None, loop)  # Placeholder - remove None in actual code


loop = asyncio.new_event_loop()
try:
    loop.run_until_complete(main())
except GracefulExit:
    print("開始關閉程式...")
    loop.run_until_complete(close_echo_tasks(echo_tasks))
finally:
    loop.close()

import asyncio
import aiohttp

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

async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, "http://python.org")
        print(html)

asyncio.run(main())

import asyncio
import aiohttp

async def fetch_page(session: aiohttp.ClientSession, url: str) -> str:
    """非同步擷取網頁內容。"""
    async with session.get(url) as response:
        return await response.text()

async def process_data(data: str):
    """處理網頁資料。"""
    #  在此加入資料處理邏輯
    print(f"處理資料：{data[:50]}...")


async def main():
    """主程式進入點。"""
    async with aiohttp.ClientSession() as session:
        urls = [f"https://example.com/page/{i}" for i in range(1, 11)]  # 產生10個測試網址
        tasks = [fetch_page(session, url) for url in urls]
        pages = await asyncio.gather(*tasks)

        for page in pages:
            await process_data(page)


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import aiohttp
from aiohttp import ClientSession

async def fetch_status(session: ClientSession, url: str) -> int:
    async with session.get(url) as response:
        return response.status

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ['https://example.com' for _ in range(1000)]
        tasks = [fetch_status(session, url) for url in urls]
        status_codes = await asyncio.gather(*tasks)
        print(status_codes)

asyncio.run(main())

import asyncio
import aiohttp
from aiohttp import ClientSession

async def fetch_status(session: ClientSession, url: str) -> int:
    async with session.get(url) as response:
        return response.status

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ['https://example.com' for _ in range(1000)]
        status_codes = []
        for url in urls:
            status_codes.append(await fetch_status(session, url))
        print(status_codes)

asyncio.run(main())

import asyncio

async def delay(seconds: int) -> int:
    await asyncio.sleep(seconds)
    return seconds

async def main():
    results = await asyncio.gather(delay(3), delay(1))
    print(results)

asyncio.run(main())

import asyncio
import aiohttp
from aiohttp import ClientSession

async def fetch_status(session: ClientSession, url: str):
    try:
        async with session.get(url) as response:
            return response.status
    except aiohttp.ClientError as e:
        return e

async def main():
    async with aiohttp.ClientSession() as session:
        urls = ['https://example.com', 'python://example.com']  # 包含一個無效 URL
        tasks = [fetch_status(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for result in results:
            if isinstance(result, Exception):
                print(f"請求發生錯誤: {result}")
            else:
                print(f"狀態碼: {result}")

asyncio.run(main())