无法控制asyncio.run()调用时，如何创建周期性异步任务

问题根源 #

你创建的周期性任务在自己手动实例化的事件循环中，但第三方库调用asyncio.run()时会默认新建一个独立的事件循环（第三方代码里没有将你传入的loop传给asyncio.run），导致两个循环完全独立运行——你的任务所在的循环根本没被启动，自然不会执行。

可行解决方案 #

通过**猴子补丁（Monkey Patch）**修改asyncio.run的行为，让它在运行第三方协程的同时，自动启动你的周期性任务，确保两者共享同一个事件循环。

方案一：通用补丁（修改asyncio.run） #

这种方式无需关心第三方库的内部实现，只要它调用了asyncio.run就能生效：

import asyncio

# 第三方库代码（无法修改）
def third_party_run(loop):
    async def runner():
        while True:
            print("third party code executing")
            await asyncio.sleep(5)

    try:
        asyncio.run(runner())
    except KeyboardInterrupt:
        return

# 你的周期性任务
async def periodic():
    while True:
        print("doing my stuff...")
        await asyncio.sleep(1)

# 猴子补丁：重写asyncio.run，合并第三方协程与你的任务
original_asyncio_run = asyncio.run

def patched_run(coro):
    async def combined_task():
        # 启动周期性任务（交给事件循环调度，无需await）
        asyncio.create_task(periodic())
        # 运行第三方的核心协程
        await coro
    return original_asyncio_run(combined_task())

# 替换原函数
asyncio.run = patched_run

# 调用第三方函数（传入的loop参数暂时无用，因为第三方未使用）
third_party_run(None)

运行后会同时输出第三方的日志和你的周期性任务日志。

方案二：针对第三方函数的补丁（利用传入的loop） #

如果你的Python版本是3.10+，asyncio.run支持loop参数，可以补丁第三方的third_party_run函数，让它真正使用你传入的事件循环：

import asyncio

# 第三方库代码（无法修改）
def third_party_run(loop):
    async def runner():
        while True:
            print("third party code executing")
            await asyncio.sleep(5)

    try:
        asyncio.run(runner())
    except KeyboardInterrupt:
        return

# 你的周期性任务
async def periodic():
    while True:
        print("doing my stuff...")
        await asyncio.sleep(1)

# 猴子补丁：修改第三方run函数，让它使用我们传入的loop
original_third_party_run = third_party_run

def patched_third_party_run(loop):
    async def runner():
        while True:
            print("third party code executing")
            await asyncio.sleep(5)
    
    try:
        # 显式将我们的loop传给asyncio.run
        asyncio.run(runner(), loop=loop)
    except KeyboardInterrupt:
        return

third_party_run = patched_third_party_run

# 创建事件循环并添加任务
loop = asyncio.new_event_loop()
loop.create_task(periodic())

# 调用第三方函数，传入我们的loop
third_party_run(loop)

这种方式更贴合你“传入事件循环”的需求，但依赖Python 3.10+的特性。

注意事项 #

• 猴子补丁必须在调用第三方函数之前完成，否则不会生效；
• 程序终止时可能会收到“任务未完成”的警告，这是因为周期性任务是无限循环，可在捕获KeyboardInterrupt时手动取消任务来避免；
• 猴子补丁属于侵入式修改，若第三方库后续更新了asyncio.run的调用逻辑，可能需要调整补丁代码。

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