asyncio任务父函数为非async时如何实现不阻塞的sleep延迟

可行解决方案 #

方案1：最小改动适配异步逻辑（推荐） #

你当前代码里已经使用asyncio.create_task()调用keep_alive，该方法本身就要求传入异步协程对象，只需对keep_alive做极小修改即可，完全不影响其余现有代码结构：

• 给keep_alive加上async修饰符
• 把sleep逻辑替换为await asyncio.sleep(5)

修改后代码片段：

class App:
    def __init__(self):
        # 原有代码完全不变
        ...
        xxx.add_handler(self.event_handler, event_xyz)
        asyncio.create_task(self.keep_alive())
        xxx.run_until_disconnected()

    # 仅修改这一个函数的定义和sleep行
    async def keep_alive(self):
        # 原有保活逻辑完全不变
        ...
        await asyncio.sleep(5) # 不会阻塞事件循环
    
    # 其余所有函数完全不变
    ...

方案2：独立线程运行阻塞逻辑（完全不用修改keep_alive原有代码） #

如果确实不能修改keep_alive的同步属性，就把阻塞逻辑放到独立线程运行，time.sleep只会阻塞该线程，不会影响asyncio事件循环的其余任务：
Python 3.9+ 可以直接用asyncio.to_thread，代码改动只涉及__init__里的任务创建行：

# __init__ 内修改创建任务的行即可，其余代码完全不变
asyncio.create_task(asyncio.to_thread(self.keep_alive))

如果要兼容Python3.9以下版本，用loop.run_in_executor实现：

def __init__(self):
    # 原有代码不变
    ...
    loop = asyncio.get_event_loop()
    # 把同步keep_alive放到默认线程池运行
    asyncio.create_task(loop.run_in_executor(None, self.keep_alive))
    xxx.run_until_disconnected()

两种方案都不需要调整现有业务逻辑，完全满足不改动整体代码结构的要求。

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