如何以异步方式记录派生类的方法调用？

你需要让日志逻辑异步执行，避免阻塞业务方法，同时隔离日志过程中的错误，且业务类及客户端代码无需修改、无需关注异步相关细节。以下是两种可行的实现方案：

方案一：线程池实现（适合IO密集型日志任务） #

利用concurrent.futures.ThreadPoolExecutor将日志任务提交到后台线程执行，主线程的业务方法可立即返回，完全不受日志IO操作的延迟影响。

修改后的Logger类代码：

import functools
from concurrent.futures import ThreadPoolExecutor
import traceback

class Logger:
    def __init__(self):
        # 根据日志吞吐量调整线程数
        self._executor = ThreadPoolExecutor(max_workers=2)

    def _log_task(self, func_name, args, kwargs):
        try:
            # 替换为实际的日志逻辑（如数据库写入、远程日志接口调用等）
            print(f"异步日志记录：方法名={func_name}, 参数={args}, 关键字参数={kwargs}")
            # 模拟耗时IO操作，比如数据库写入
            # time.sleep(0.5)
        except Exception:
            # 单独捕获日志异常，避免影响业务逻辑
            traceback.print_exc()

    def _decorator(self, f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            # 提交日志任务到线程池，异步执行
            self._executor.submit(self._log_task, f.__name__, args, kwargs)
            # 直接执行业务方法，不等待日志完成
            return f(*args, **kwargs)

        return wrapper

    def __getattribute__(self, item):
        value = object.__getattribute__(self, item)
        # 排除Logger自身的核心方法，避免递归装饰
        if callable(value) and item not in ('_decorator', '_log_task', '_executor'):
            decorator = object.__getattribute__(self, '_decorator')
            return decorator(value)
        return value

    def __del__(self):
        # 程序退出时关闭线程池，可根据需求设置wait参数
        self._executor.shutdown(wait=False)

class A(Logger):
    def method(self, a, b):
        print(f"业务方法method执行：a={a}, b={b}")

    def another_method(self, c):
        print(f"业务方法another_method执行：c={c}")

    @staticmethod
    def static_method(d):
        print(f"静态方法static_method执行：d={d}")

# 客户端代码完全无需修改
if __name__ == "__main__":
    a = A()
    a.method(1, 2)
    a.another_method(3)
    A.static_method(4)

核心特点： #

• 线程池专门处理日志的IO操作，业务方法无阻塞、无延迟
• 日志任务的异常被独立捕获，不会扩散到业务逻辑
• 业务类A和客户端代码完全不需要修改，感知不到异步逻辑的存在

方案二：asyncio后台事件循环（适合异步日志逻辑） #

如果你的日志逻辑本身是异步实现的（比如使用异步数据库驱动），可以在Logger内部维护一个独立的事件循环线程，将异步日志任务提交到该循环执行，业务代码依然无需关注asyncio相关细节。

示例代码：

import functools
import asyncio
import traceback
from threading import Thread

class Logger:
    def __init__(self):
        # 启动独立的事件循环线程
        self._loop = asyncio.new_event_loop()
        self._loop_thread = Thread(target=self._run_loop, daemon=True)
        self._loop_thread.start()

    def _run_loop(self):
        asyncio.set_event_loop(self._loop)
        self._loop.run_forever()

    async def _async_log_task(self, func_name, args, kwargs):
        try:
            # 异步日志逻辑，比如调用异步数据库接口
            print(f"异步日志记录：方法名={func_name}, 参数={args}, 关键字参数={kwargs}")
            # await async_db_client.log(func_name, args, kwargs)
        except Exception:
            traceback.print_exc()

    def _decorator(self, f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            # 将异步日志任务提交到后台事件循环
            asyncio.run_coroutine_threadsafe(
                self._async_log_task(f.__name__, args, kwargs),
                loop=self._loop
            )
            return f(*args, **kwargs)

        return wrapper

    def __getattribute__(self, item):
        value = object.__getattribute__(self, item)
        if callable(value) and item not in ('_decorator', '_async_log_task', '_run_loop'):
            decorator = object.__getattribute__(self, '_decorator')
            return decorator(value)
        return value

# 业务类A和客户端代码完全不变
class A(Logger):
    def method(self, a, b):
        print(f"业务方法method执行：a={a}, b={b}")

    def another_method(self, c):
        print(f"业务方法another_method执行：c={c}")

    @staticmethod
    def static_method(d):
        print(f"静态方法static_method执行：d={d}")

核心特点： #

• 事件循环在独立线程运行，不干扰主线程的同步业务代码
• 异步日志任务通过asyncio.run_coroutine_threadsafe提交，无需业务代码添加async/await
• 守护线程确保程序退出时自动终止事件循环

内容的提问来源于stack exchange，提问作者sourcream