如何获取asyncio.start_server创建的运行client_connected_cb的Task实例

首先，asyncio官方没有提供公开API可以直接获取asyncio.start_server自动创建的client_connected_cb对应的Task实例，目前可用的实现方案分为两类：

方案1：手动维护任务集合（推荐生产环境使用） #

就是你提到的临时方案，这也是目前社区公认的最稳定的实现方式，不会依赖asyncio的内部私有实现，兼容所有支持asyncio的Python版本。
具体实现逻辑：

1. 预先创建一个集合存储活跃的客户端连接任务
2. 在client_connected_cb执行的最开始通过asyncio.current_task()获取当前任务实例，加入集合
3. 用finally块保证任务不管正常结束还是抛出异常，都会从集合中移除，避免内存泄漏
4. teardown阶段先停止服务端接收新连接，再等待集合中所有任务执行完成

示例代码：

import asyncio

# 存储活跃的客户端任务
active_client_tasks = set()

async def client_connected_cb(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    current_task = asyncio.current_task()
    active_client_tasks.add(current_task)
    try:
        # 原有业务逻辑
        data = await reader.read(1024)
        # 省略处理逻辑...
    finally:
        # 任务结束后移除
        active_client_tasks.discard(current_task)
        # 确保writer完全关闭
        writer.close()
        await writer.wait_closed()

async def teardown(server: asyncio.Server):
    # 1. 停止接收新的客户端连接
    server.close()
    await server.wait_closed()
    # 2. 等待所有活跃客户端任务执行完毕，忽略可能的异常
    if active_client_tasks:
        await asyncio.gather(*active_client_tasks, return_exceptions=True)

方案2：读取服务端私有属性（仅测试用，不推荐生产使用） #

asyncio.start_server返回的Server实例内部维护了私有属性_active_tasks存储所有活跃的客户端连接任务，你可以直接读取该属性获取任务列表。
注意：这是asyncio的内部私有实现，不同Python版本的属性名、实现逻辑可能随时变更，生产环境使用会有兼容性风险
示例teardown代码：

async def teardown(server: asyncio.Server):
    server.close()
    await server.wait_closed()
    # 读取私有属性获取活跃任务
    if hasattr(server, "_active_tasks") and server._active_tasks:
        await asyncio.gather(*server._active_tasks, return_exceptions=True)

内容的提问来源于stack exchange，提问作者Mr. Fegur