全局异步任务管理器 - 单例模式
| 11 | |
| 12 | |
| 13 | class TaskManager: |
| 14 | """全局异步任务管理器 - 单例模式""" |
| 15 | |
| 16 | _instance = None |
| 17 | _lock = asyncio.Lock() |
| 18 | |
| 19 | def __new__(cls): |
| 20 | if cls._instance is None: |
| 21 | cls._instance = super().__new__(cls) |
| 22 | cls._instance._initialized = False |
| 23 | return cls._instance |
| 24 | |
| 25 | def __init__(self): |
| 26 | if self._initialized: |
| 27 | return |
| 28 | |
| 29 | self._tasks: Set[asyncio.Task] = set() |
| 30 | self._resources: Set[Any] = set() # 需要关闭的资源 |
| 31 | self._shutdown_event = asyncio.Event() |
| 32 | self._initialized = True |
| 33 | log.debug("TaskManager initialized") |
| 34 | |
| 35 | def register_task(self, task: asyncio.Task, description: str = None) -> asyncio.Task: |
| 36 | """注册一个任务供生命周期管理""" |
| 37 | self._tasks.add(task) |
| 38 | task.add_done_callback(lambda t: self._tasks.discard(t)) |
| 39 | |
| 40 | if description: |
| 41 | task.set_name(description) |
| 42 | |
| 43 | log.debug(f"Registered task: {task.get_name() or 'unnamed'}") |
| 44 | return task |
| 45 | |
| 46 | def create_task(self, coro, *, name: str = None) -> asyncio.Task: |
| 47 | """创建并注册一个任务""" |
| 48 | task = asyncio.create_task(coro, name=name) |
| 49 | return self.register_task(task, name) |
| 50 | |
| 51 | def register_resource(self, resource: Any) -> Any: |
| 52 | """注册一个需要清理的资源（如HTTP客户端、文件句柄等）""" |
| 53 | # 使用弱引用避免循环引用 |
| 54 | self._resources.add(weakref.ref(resource)) |
| 55 | log.debug(f"Registered resource: {type(resource).__name__}") |
| 56 | return resource |
| 57 | |
| 58 | async def shutdown(self, timeout: float = 30.0): |
| 59 | """关闭所有任务和资源""" |
| 60 | log.info("TaskManager shutdown initiated") |
| 61 | |
| 62 | # 设置关闭标志 |
| 63 | self._shutdown_event.set() |
| 64 | |
| 65 | # 取消所有未完成的任务 |
| 66 | cancelled_count = 0 |
| 67 | for task in list(self._tasks): |
| 68 | if not task.done(): |
| 69 | task.cancel() |
| 70 | cancelled_count += 1 |