在Trio框架中，如何实现与对象生命周期绑定的后台任务？

这个需求太常见了——Trio的nursery确实依赖上下文，但咱们完全可以把nursery的管理逻辑封装到类内部，通过异步上下文管理器让类的生命周期和后台任务的生命周期自动绑定，既不用暴露实现细节，又能完美控制任务的启动和停止。

核心思路 #

让你的类实现异步上下文管理器协议（也就是定义__aenter__和__aexit__异步方法）：

• 在__aenter__中创建并启动一个nursery，把它存在实例内部
• 用这个内部nursery启动所有需要长期运行的后台任务（比如心跳）
• 在__aexit__中等待nursery完成，确保所有后台任务优雅终止

这样外部代码只需要通过async with来使用你的类，完全不用关心nursery的存在，实现细节被完美封装。

完整代码示例 #

import trio
import time

class HeartbeatService:
    def __init__(self, interval: float = 1.0):
        self._interval = interval
        self._nursery = None  # 先初始化，在__aenter__中赋值

    async def __aenter__(self):
        # 创建nursery并保存到实例内部
        self._nursery = await trio.Nursery().__aenter__()
        # 启动后台心跳任务
        self._nursery.start_soon(self._heartbeat_task)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 等待nursery中的所有任务完成
        await self._nursery.__aexit__(exc_type, exc_val, exc_tb)
        self._nursery = None

    async def _heartbeat_task(self):
        """长期运行的心跳任务"""
        try:
            while True:
                print(f"[{time.strftime('%H:%M:%S')}] 心跳正常")
                await trio.sleep(self._interval)
        except trio.Cancelled:
            # 捕获取消信号，做必要的清理工作
            print("心跳任务已终止")
            raise  # 必须重新抛出，让nursery知道任务已响应取消

# 外部使用示例
async def main():
    print("启动服务...")
    async with HeartbeatService(interval=2.0) as service:
        # 这里可以做其他业务逻辑，比如处理请求
        await trio.sleep(5)
        print("业务逻辑完成，准备停止服务...")
    print("服务已完全停止")

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

关键细节解释 #

1. 异步上下文管理器的必要性：
   因为nursery必须在异步上下文中创建，而类的__init__是同步方法，所以不能在__init__里初始化nursery，必须放到异步的__aenter__中。

2. 后台任务的取消处理：
   当退出async with块时，Trio会给nursery中的所有任务发送Cancelled异常，你的后台任务需要捕获这个异常（如果需要清理资源），然后重新抛出，这样nursery才能正确确认任务已终止。

3. 多个后台任务的管理：
   如果你的类需要多个长期任务，只需要在__aenter__中多次调用self._nursery.start_soon()提交任务即可，nursery会统一管理所有任务的生命周期。

4. 手动控制启动/停止（可选）：
   如果你不想用async with，也可以给类添加start()和stop()方法：

   async def start(self):
       self._nursery = await trio.Nursery().__aenter__()
       self._nursery.start_soon(self._heartbeat_task)
   
   async def stop(self):
       await self._nursery.__aexit__(None, None, None)
       self._nursery = None
   

   但这种方式需要你自己确保stop()被调用，不如异步上下文管理器安全（它会自动处理异常情况，比如代码抛出错误时也会正确停止任务）。

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