异步循环中同步函数回调内执行异步代码的最优方案问询

场景背景 #

在异步代码中调用第三方同步库（无法修改其逻辑），该库会触发我提供的同步回调函数，我需要在这个同步回调内执行异步操作，且持有当前运行所有代码的事件循环（loop）引用。

已尝试的无效方案 #

• 方案1：通过事件循环等待任务完成
  尝试调用loop.run_until_complete(task)，但触发RuntimeError: This event loop is already running报错。且由于同步回调的签名无法修改，无法直接使用await等待任务完成。
• 方案2：主动循环等待任务完成
  用while not task.done(): time.sleep(1)主动轮询，会完全阻塞事件队列，导致整个异步逻辑无法正常运行。

当前可行方案（方案3） #

在同步回调中创建独立异步任务，并在异步代码中跟踪这些任务直至完成，避免任务被垃圾回收。具体实现如下：

import asyncio

# 无法控制的第三方同步库
i: int = 1_000
def sync_lib_call(user_cb):
    global i
    i += 1  # 模拟数据生成
    user_cb(i)

# 我的代码
queue = asyncio.Queue()
loop = None
enqueue_tasks = []

async def put_in_queue(value: int):
    print(f"enqueue: {value}")
    await queue.put(value)

def my_callback(value: int):
    task = loop.create_task(put_in_queue(value))
    # 强引用任务，防止被垃圾回收
    enqueue_tasks.append(task)

async def work():
    global loop, enqueue_tasks
    loop = asyncio.get_running_loop()
    sync_lib_call(my_callback) # 调用同步库触发回调
    # 清理已完成的任务引用
    enqueue_tasks = [t for t in enqueue_tasks if not t.done()]
    await asyncio.sleep(1) # 执行其他异步操作

async def main():
    while True:
        print(f"qsize: {queue.qsize()}")
        print(enqueue_tasks)
        await work()

try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass

替代方案对比 #

1. asyncio.to_thread() #

该方案不适用当前场景：它仅适合同步函数/回调不生成新的长期运行异步任务的情况，而我的场景中回调内的异步代码可能创建需要在同步代码返回后仍保持活跃的任务。

2. asyncio.run_coroutine_threadsafe() #

该方案可行，是方案3的替代选项，但队列满时行为与方案3不同：

• 此方案会阻塞同步代码，直到队列有空闲位置
• 方案3会将异步任务加入跟踪列表，等待后续自动推入主队列

最终选择哪种方案，取决于队列满时的预期行为：是等待队列空闲、丢弃数据，还是添加异步推送任务最终完成入队。

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