python - 如何包装自定义 future 以在 Python 中与 asyncio 一起使用？

Question

有很多库使用他们的自定义版本 Future . kafka和 s3transfer只是两个例子:他们所有自定义的类 future 类都有 object 作为父类(super class)。

毫不奇怪，您不能对此类对象直接调用 asyncio.wrap_future()，也不能对它们使用 await。

包装此类 futures 以与 asyncio 一起使用的正确方法是什么？

Answer 1

如果 future 类支持标准的 future 特性，例如 done 回调和 result 方法，只需使用如下内容:

def wrap_future(f):
    loop = asyncio.get_event_loop()
    aio_future = loop.create_future()
    def on_done(*_):
        try:
            result = f.result()
        except Exception as e:
            loop.call_soon_threadsafe(aio_future.set_exception, e)
        else:
            loop.call_soon_threadsafe(aio_future.set_result, result)
    f.add_done_callback(on_done)
    return aio_future

将该代码视为一个模板，您可以对其进行自定义以匹配您正在处理的 future 的具体情况。

预期用途是从运行异步事件循环的线程调用它:

value = await wrap_future(some_foreign_future)

如果您从不同的线程调用它，请务必显式传递 loop，因为 asyncio.get_event_loop 从未注册到 asyncio 的线程调用时将失败。