如何从同步函数中调用异步函数且无需等待其执行完成？

你的问题分析 #

• 第一个测试用asyncio.run()会等待协程完成，因为asyncio.run()的设计就是阻塞当前线程，直到传入的协程执行结束。
• 第二个测试用asyncio.ensure_future()但没启动事件循环，协程任务只是被创建，根本没机会被调度执行。

可行解决方案 #

方案1：后台线程运行事件循环

把异步事件循环放到后台线程中运行，主线程提交协程任务后即可继续执行，无需等待任务完成。

import asyncio
import datetime
from threading import Thread

async def test_timer_function():
    await asyncio.sleep(10)
    print(f"Timer finished at {datetime.now()}")
    return

def run_async_loop(loop):
    asyncio.set_event_loop(loop)
    loop.run_forever()

def main():
    print(f"Starting timer at {datetime.now()}")
    # 创建事件循环并在后台线程启动
    loop = asyncio.new_event_loop()
    # 设置daemon=True，让后台线程随主线程退出而终止
    thread = Thread(target=run_async_loop, args=(loop,), daemon=True)
    thread.start()
    # 提交协程到后台事件循环，主线程不阻塞
    asyncio.run_coroutine_threadsafe(test_timer_function(), loop)
    print(f"Ending timer at {datetime.now()}")
    # 这里主线程可以执行其他逻辑，示例中sleep是为了等待后台任务完成看结果
    import time
    time.sleep(11)

if __name__ == "__main__":
    main()

方案2：使用asyncio.create_task配合长期运行的循环（适用于主线程无需立即退出场景）

如果主线程本身需要长期运行（比如服务端程序），可以直接启动事件循环，将协程转为后台任务后，主线程继续处理其他逻辑：

import asyncio
import datetime

async def test_timer_function():
    await asyncio.sleep(10)
    print(f"Timer finished at {datetime.now()}")

async def main_async():
    print(f"Starting timer at {datetime.now()}")
    # 创建后台任务，不阻塞当前协程
    asyncio.create_task(test_timer_function())
    print(f"Ending timer at {datetime.now()}")
    # 主线程协程继续执行其他逻辑，这里用sleep模拟
    await asyncio.sleep(11)

def main():
    asyncio.run(main_async())

if __name__ == "__main__":
    main()

关键注意点 #

• 跨线程提交协程必须用asyncio.run_coroutine_threadsafe()，这是线程安全的方式，不要直接在非事件循环线程操作协程。
• 后台线程的事件循环设置daemon=True，避免主线程退出后程序残留后台线程无法终止。

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