立即从同步代码执行异步回调

Question

问题

我有一个库，目前没有异步支持，需要从异步代码调用。handler异步代码通过处理程序（下面代码中的函数）调用库。当处理程序执行时，库会定期调用回调 ( callback_wrapper) 来报告进度。

同步处理程序的执行顺序是为了ThreadPoolExecutor使主事件循环能够在处理程序运行时处理更多事件。

发生的情况是同步回调立即执行，但异步回调仅在主处理程序执行后执行。期望的结果是立即执行异步回调。

我猜事件循环在run_in_executor调用时被阻止，但我不知道如何解决这个问题。

代码

import asyncio
import time
from concurrent.futures.thread import ThreadPoolExecutor

loop = asyncio.get_event_loop()


def handler():
    print('handler started')
    callback_wrapper()
    time.sleep(1)
    print('handler stopped')


async def callback():
    print('callback')


def callback_wrapper():
    print('callback wrapper started')
    asyncio.ensure_future(callback(), loop=loop)
    print('callback wrapper stopped')


async def main():
    handler()


with ThreadPoolExecutor() as pool:
    async def thread_handler():
        await loop.run_in_executor(pool, handler)


    loop.run_until_complete(main())

输出

handler started
callback wrapper started
callback wrapper stopped
handler stopped
callback

所需输出

handler started
callback wrapper started
callback
callback wrapper stopped
handler stopped

Answer 1

感谢@user4815162342的输入，我提出了以下解决方案：

import asyncio
import time
from concurrent.futures.thread import ThreadPoolExecutor

loop = asyncio.get_event_loop()


def handler():
    print('handler started')
    callback_wrapper()
    time.sleep(1)
    print('handler stopped')


async def callback():
    print('callback')


def callback_wrapper():
    print('callback wrapper started')
    asyncio.run_coroutine_threadsafe(callback(), loop).result()
    print('callback wrapper stopped')


async def main():
    await thread_handler()


with ThreadPoolExecutor() as pool:
    async def thread_handler():
        await loop.run_in_executor(pool, handler)


    loop.run_until_complete(main())

这会产生所需的结果：

handler started
callback wrapper started
callback
callback wrapper stopped
handler stopped