Python 非同步程式除錯進階技術

global shared_counter

async def race_task(task_id, delay):
    local = shared_counter
    logger.debug("任務 %s：讀取 shared_counter = %d", task_id, local)
    await asyncio.sleep(delay)
    shared_counter = local + 1
    logger.debug("任務 %s：更新 shared_counter 到 %d", task_id, shared_counter)

async def run_race():
    tasks = [asyncio.create_task(race_task(i, 0.01)) for i in range(10)]
    await asyncio.gather(*tasks)
    logger.debug("shared_counter 的最終值：%d", shared_counter)

asyncio.run(run_race())

import asyncio
import aiomonitor

async def sample_task(task_id):
    await asyncio.sleep(0.1)
    return f"任務 {task_id} 完成"

async def main():
    tasks = [asyncio.create_task(sample_task(i)) for i in range(5)]
    result = await asyncio.gather(*tasks)
    print(result)

with aiomonitor.start_monitor(loop=asyncio.get_event_loop()):
    asyncio.run(main())

import asyncio

async def nested_coroutine():
    # 觸發錯誤以產生非同步堆積疊追蹤
    raise ValueError("用於堆積疊追蹤演示的故意錯誤")

async def parent_coroutine():
    try:
        await nested_coroutine()
    except Exception as e:
        # 從異常上下文列印非同步堆積疊追蹤
        import traceback
        traceback.print_exc()

asyncio.run(parent_coroutine())

import asyncio
import functools
import logging

logger = logging.getLogger("trace_debug")
logger.setLevel(logging.DEBUG)

def async_trace(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        logger.debug("進入 %s", func.__name__)
        try:
            result = await func(*args, **kwargs)
            logger.debug("離開 %s", func.__name__)
            return result
        except Exception as e:
            logger.debug("在 %s 中的異常：%s", func.__name__, e)
            raise
    return wrapper

@async_trace
async def complex_async_operation(x):
    await asyncio.sleep(0.05)
    if x < 0:
        raise ValueError("無效輸入：x 必須為非負數")
    return x * 2

async def run_operation():
    try:
        result = await complex_async_operation(5)
        print("結果：", result)
    except Exception as e:
        print("錯誤：", e)

asyncio.run(run_operation())

import asyncio
import logging
import uuid
import contextvars

# 建立一個上下文變數來儲存唯一的任務識別碼
task_id_var = contextvars.ContextVar("task_id", default="no-task")

class ContextFilter(logging.Filter):
    def filter(self, record):
        record.task_id = task_id_var.get()
        return True

logger = logging.getLogger("async_logger")
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
formatter = logging.Formatter(
    "%(asctime)s - %(levelname)s - [Task: %(task_id)s] - %(message)s"
)
handler.setFormatter(formatter)
handler.addFilter(ContextFilter())
logger.addHandler(handler)

import functools

def async_trace(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        task_id = str(uuid.uuid4())
        token = task_id_var.set(task_id)
        logger.debug("Entering %s", func.__name__)
        try:
            result = await func(*args, **kwargs)
            logger.debug("Exiting %s", func.__name__)
            return result
        except Exception as ex:
            logger.error("Exception in %s: %s", func.__name__, ex, exc_info=True)
            raise
        finally:
            task_id_var.reset(token)
    return wrapper

@async_trace
async def process_data(value):
    await asyncio.sleep(0.05)
    return value * 2

async def main():
    results = await asyncio.gather(
        process_data(1),
        process_data(2),
        process_data(3)
    )
    print("Results:", results)

asyncio.run(main())

import json

class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'task_id': record.task_id,
            'message': record.getMessage()
        }
        return json.dumps(log_data)

# 使用 JsonFormatter
handler.setFormatter(JsonFormatter())

import json_log_formatter
import logging

# 設定結構化記錄格式
formatter = json_log_formatter.JSONFormatter()
json_handler = logging.StreamHandler()
json_handler.setFormatter(formatter)
json_handler.addFilter(ContextFilter())

# 將處理器新增到記錄器
logger.addHandler(json_handler)
logger.debug("啟用非同步任務追蹤的結構化記錄。")

import asyncio

# 取得事件迴圈並啟用除錯模式
loop = asyncio.get_event_loop()
loop.set_debug(True)

# 執行主程式
asyncio.run(main())

@async_trace
async def update_shared_resource(shared, key, increment):
    # 在更新前記錄目前狀態
    logger.debug("更新前的目前狀態：%s", shared)
    
    # 人為引入 yield 以模擬潛在的交錯執行
    await asyncio.sleep(0)
    
    # 更新共用資源
    shared[key] = shared.get(key, 0) + increment
    
    # 在更新後記錄新狀態
    logger.debug("更新後的新狀態：%s", shared)
    return shared[key]

async def run_updates():
    shared_resource = {}
    tasks = [
        update_shared_resource(shared_resource, "counter", 1)
        for _ in range(10)
    ]
    await asyncio.gather(*tasks)
    logger.info("最終共用資源狀態：%s", shared_resource)