如何基于Asyncio实现多Future结果的实时返回？

我有一个异步生成器函数run_command，通过httpx异步流式请求外部接口，分批返回结果（结果可能延迟较长时间到达）。单数据集调用时可正常运行，结果能通过Django Channels推送到GUI。现在需要并发调用该函数处理多数据集，尝试使用asyncio.as_completed时触发TypeError: An asyncio.Future, a coroutine or an awaitable is required等错误，请问如何实现多Future结果到达即返回？

asyncio.as_completed只接受可等待对象（Future、协程、Task），但异步生成器本身不是可等待对象——它需要被迭代来产出值。要实现多异步生成器的结果实时返回，得把每个生成器的迭代逻辑封装成Task，同时监听每个Task的产出，或者用任务包装+实时处理的方式。

方法一：用任务包装每个生成器的迭代逻辑 #

把每个run_command的迭代和结果推送逻辑包装成独立的协程，然后用asyncio.create_task创建任务，让它们并发运行，每个任务内部实时处理自己的流式结果：

import asyncio
import httpx
from channels.layers import get_channel_layer

async def process_dataset(dataset_id, channel_name):
    channel_layer = get_channel_layer()
    # 迭代异步生成器的每一批结果
    async for chunk in run_command(dataset_id):
        # 实时推送到Django Channels
        await channel_layer.send(channel_name, {
            "type": "send.result",
            "data": {"dataset_id": dataset_id, "chunk": chunk}
        })

async def run_concurrent(datasets, channel_name):
    # 创建所有并发任务
    tasks = [asyncio.create_task(process_dataset(ds_id, channel_name)) for ds_id in datasets]
    # 等待所有任务完成，return_exceptions=True避免单个任务报错导致全部终止
    await asyncio.gather(*tasks, return_exceptions=True)

方法二：用队列集中处理所有生成器的实时结果 #

如果需要统一管控所有生成器的输出，可以用asyncio.Queue作为中间层，每个生成器把结果丢到队列里，单独开一个消费者任务负责推送到GUI：

async def producer(dataset_id, queue):
    async for chunk in run_command(dataset_id):
        await queue.put(("result", dataset_id, chunk))
    # 标记当前生成器处理完成
    await queue.put(("done", dataset_id))

async def consumer(queue, channel_name):
    channel_layer = get_channel_layer()
    active_producers = len(datasets)
    while active_producers > 0:
        msg_type, dataset_id, data = await queue.get()
        if msg_type == "result":
            await channel_layer.send(channel_name, {
                "type": "send.result",
                "data": {"dataset_id": dataset_id, "chunk": data}
            })
        elif msg_type == "done":
            active_producers -= 1
        queue.task_done()

async def run_concurrent(datasets, channel_name):
    queue = asyncio.Queue(maxsize=10)  # 根据业务场景调整队列大小
    # 创建所有生产者任务
    producer_tasks = [asyncio.create_task(producer(ds_id, queue)) for ds_id in datasets]
    # 创建消费者任务
    consumer_task = asyncio.create_task(consumer(queue, channel_name))
    # 等待所有生产者完成
    await asyncio.gather(*producer_tasks)
    # 等待队列中所有消息处理完毕
    await queue.join()
    # 取消消费者任务
    consumer_task.cancel()
    try:
        await consumer_task
    except asyncio.CancelledError:
        pass

错误原因说明 #

asyncio.as_completed期望的是能一次性返回最终结果的可等待对象，但异步生成器是逐步产出值的，并非一次性完成的类型。你之前直接把run_command(dataset_id)传给as_completed，它不属于可等待对象范畴，因此触发类型错误。

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