Python中如何封装自定义Future以适配asyncio异步框架？

这个问题我之前在对接Kafka生产者和s3transfer的时候实打实踩过坑！那些第三方库搞的自定义Future完全不兼容asyncio的标准接口，直接await或者用asyncio.wrap_future()根本行不通——毕竟它们连concurrent.futures.Future都不是，就是个普通object子类。

下面两种是我验证过的靠谱适配方式，按优先级推荐：

1. 用asyncio.Future做回调桥接（最优解） #

核心思路是创建一个asyncio原生的Future作为中间层，给自定义Future绑定完成回调，当自定义Future执行完毕时，把结果或异常同步到asyncio.Future上，这样就能正常await了。

举个Kafka生产者的实际例子：

import asyncio
from kafka import KafkaProducer

async def adapt_custom_future(custom_fut):
    # 初始化一个asyncio Future作为桥接器
    async_fut = asyncio.Future()

    def sync_result(_):
        # 确保async_fut还未完成，避免重复设置
        if async_fut.done():
            return
        try:
            # 从自定义Future获取结果
            result = custom_fut.result()
            async_fut.set_result(result)
        except Exception as err:
            async_fut.set_exception(err)

    # 给自定义Future绑定完成回调
    custom_fut.add_done_callback(sync_result)

    # 额外处理取消逻辑：如果async_fut被取消，同步取消自定义Future
    def sync_cancel(_):
        if not custom_fut.done():
            custom_fut.cancel()
    async_fut.add_done_callback(sync_cancel)

    # 现在可以正常await这个asyncio Future了
    return await async_fut

# 使用示例
async def main():
    producer = KafkaProducer(bootstrap_servers="localhost:9092")
    # Kafka的send方法返回的就是自定义Future
    kafka_fut = producer.send("test_topic", b"hello async kafka")
    # 适配后直接await
    send_result = await adapt_custom_future(kafka_fut)
    print(f"消息发送成功：{send_result}")

asyncio.run(main())

这种方式效率最高，因为是基于回调触发，不用轮询占用CPU，而且能完整同步结果、异常和取消状态。

2. 轮询自定义Future的状态（备选方案） #

如果某些自定义Future不支持添加回调（虽然大部分第三方库都会提供add_done_callback），可以用轮询的方式检查它的done()状态，直到完成再获取结果：

async def poll_custom_future(custom_fut):
    # 循环检查自定义Future是否完成
    while not custom_fut.done():
        # 每次暂停10ms，避免CPU空转
        await asyncio.sleep(0.01)
    # 结果或异常直接抛出，和正常await逻辑一致
    try:
        return custom_fut.result()
    except Exception as err:
        raise err

这种方式简单但不够高效，适合回调接口缺失的极端场景，一般优先用第一种。

为什么不能直接用asyncio.wrap_future？ #

asyncio.wrap_future()的设计目标是适配concurrent.futures.Future及其子类，它依赖该类的特定接口（比如_asyncio_future_blocking属性、内部的回调机制等）。而第三方库的自定义Future只是普通object，没有这些接口，自然无法被识别。

内容的提问来源于stack exchange，提问作者kirill.shirokov