如何从同一迭代器并发执行两个聚合函数且无需缓冲输出？

当然可以做到！而且完全不需要缓冲迭代器的任何输出——Python的协程（不管是同步生成器还是异步协程）天生就支持暂停和恢复执行，正好能满足你的需求：让同一个迭代器的每个值实时分发给多个聚合函数，同时每个函数只维护自己的聚合状态，内存占用始终保持恒定。

核心思路 #

我们不需要把迭代器的所有值存起来，而是做一个值分发器：逐个取出迭代器的值，立刻传递给每个聚合函数的执行过程；每个聚合函数用协程实现，接收值后更新自己的状态，直到所有值处理完毕，再返回最终结果。

同步场景方案（普通迭代器） #

如果你的迭代器是同步的（比如iter(range(1, 1000))），用Python的生成器协程就能完美解决，不需要任何异步框架：

步骤1：定义聚合协程 #

把sum()和max()的逻辑包装成协程，它们通过yield接收值，最后返回聚合结果：

def coro_sum():
    total = 0
    while True:
        val = yield
        if val is None:  # 用None作为结束信号
            return total
        total += val

def coro_max():
    current_max = None
    while True:
        val = yield
        if val is None:
            return current_max
        if current_max is None or val > current_max:
            current_max = val

步骤2：实现值分发器 #

这个函数负责启动协程、分发每个值，最后收集结果：

def get_aggregate_results(iterator, *coros):
    # 先启动所有协程（让它们走到第一个yield的位置）
    for coro in coros:
        next(coro)
    
    # 逐个分发迭代器的值给所有协程
    for val in iterator:
        for coro in coros:
            coro.send(val)
    
    # 发送结束信号，收集每个协程的结果
    results = []
    for coro in coros:
        try:
            coro.send(None)
        except StopIteration as e:
            results.append(e.value)
    
    return results

测试代码 #

# 创建目标迭代器
my_iter = iter(range(1, 1000))
# 获取sum和max的结果
sum_result, max_result = get_aggregate_results(my_iter, coro_sum(), coro_max())

print(f"Sum: {sum_result}, Max: {max_result}")
# 输出：Sum: 499500, Max: 999

这个方案里，没有任何值被缓冲——每个值从迭代器取出后，立刻传给两个协程，然后就被丢弃了。协程只维护自己的状态变量（比如total、current_max），内存占用完全不随迭代器大小变化。

异步场景方案（异步迭代器） #

如果你的数据源是异步的（比如从网络流、异步数据库读取数据），可以用asyncio实现类似的逻辑，同样不需要缓冲：

步骤1：定义异步聚合协程 #

用async def定义协程，通过队列接收值：

import asyncio

async def async_coro_sum(queue):
    total = 0
    while True:
        val = await queue.get()
        if val is None:  # 结束信号
            queue.task_done()
            return total
        total += val
        queue.task_done()

async def async_coro_max(queue):
    current_max = None
    while True:
        val = await queue.get()
        if val is None:
            queue.task_done()
            return current_max
        if current_max is None or val > current_max:
            current_max = val
        queue.task_done()

步骤2：实现异步分发器 #

async def async_process(async_iterator):
    # 用大小为1的队列，避免缓冲多余数据
    queue = asyncio.Queue(maxsize=1)
    
    # 创建异步任务
    sum_task = asyncio.create_task(async_coro_sum(queue))
    max_task = asyncio.create_task(async_coro_max(queue))
    
    # 异步分发每个值
    async for val in async_iterator:
        await queue.put(val)
    
    # 发送结束信号（每个任务一个）
    await queue.put(None)
    await queue.put(None)
    
    # 等待任务完成并获取结果
    sum_result = await sum_task
    max_result = await max_task
    
    return sum_result, max_result

测试代码 #

# 模拟一个异步迭代器
async def async_range(start, end):
    for i in range(start, end):
        await asyncio.sleep(0)  # 模拟异步IO操作
        yield i

async def main():
    sum_res, max_res = await async_process(async_range(1, 1000))
    print(f"Async Sum: {sum_res}, Async Max: {max_res}")

asyncio.run(main())

这个异步方案同样不会缓冲大量数据，队列最多只存一个值，每个值被处理后就会被丢弃，适合处理大规模的异步数据流。

关键总结 #

不管是同步还是异步方案，核心都是利用协程的暂停/恢复特性：

• 聚合函数不再是一次性消费整个迭代器，而是逐个接收值并更新状态
• 分发器负责把每个值实时传递给所有聚合函数，完全不需要存储迭代器的输出

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