asyncio 具有 map&reduce 风格且不会淹没事件循环答案

【问题标题】：asyncio with map&reduce flavor and without flooding the event loopasyncio 具有 map&reduce 风格且不会淹没事件循环
【发布时间】：2014-08-03 14:50:18
【问题描述】：

我正在尝试在实际应用程序中使用 asyncio，但结果并非如此很简单，急需 asyncio 大师的帮助。

在没有泛滥事件循环的情况下产生其他任务的任务（成功！）

考虑一个任务，比如从一些“播种”网页开始抓取网页。每个网页导致以指数方式生成新的下载任务（！）进展。但是，我们既不想淹没事件循环，也不想超载我们的网络。我们想控制任务流程。这就是我通过修改此处提出的漂亮的 Maxime 解决方案取得良好效果： https://mail.python.org/pipermail/python-list/2014-July/687823.html

map & reduce（失败）

好吧，但我还需要一个非常自然的东西，一种 map() 和 reduce() 或者 functools.reduce() 如果我们已经在 python3 上。也就是说，我需要为完成的所有下载任务调用“汇总”函数来自页面的链接。这是我失败的地方:(

我会提出一个过于简单但仍然很好的测试来模拟用例：让我们以无效的形式使用斐波那契函数实现。也就是说，让 coro_sum() 应用在 reduce() 中，而 coro_fib 是我们应用的地图（）。像这样的：

@asyncio.coroutine
def coro_sum(x):
    return sum(x)

@asyncio.coroutine
def coro_fib(x):
    if x < 2:
        return 1
    res_coro =
executor_pool.spawn_task_when_arg_list_of_coros_ready(coro=coro_sum,

 arg_coro_list=[coro_fib(x - 1), coro_fib(x - 2)])
    return res_coro

这样我们就可以运行以下测试。

对一名工人进行测试#1：

executor_pool = ExecutorPool(workers=1)
executor_pool.as_completed( coro_fib(x) for x in range(20) )

对两个工人的测试#2：

executor_pool = ExecutorPool(workers=2)
executor_pool.as_completed( coro_fib(x) for x in range(20) )

每个 coro_fib() 和 coro_sum() 都非常重要调用是通过某个工作人员上的任务完成的，而不仅仅是隐式产生的并且不受管理！

很高兴找到对这个非常自然的目标感兴趣的 asyncio 大师。非常感谢您的帮助和想法。

最好的问候

瓦莱里

【问题讨论】：

@Joren，奇怪......看起来链接不是永久存在的。反正更新了。

标签： python-3.x scheduled-tasks python-asyncio

【解决方案1】：

有multiple ways to compute fibonacci series asynchroniously。首先，检查爆炸变体在您的情况下是否失败：

@asyncio.coroutine
def coro_sum(summands):
    return sum(summands)

@asyncio.coroutine
def coro_fib(n):
    if   n == 0: s = 0
    elif n == 1: s = 1
    else:
        summands, _ = yield from asyncio.wait([coro_fib(n-2), coro_fib(n-1)])
        s = yield from coro_sum(f.result() for f in summands)
    return s

您可以将summands 替换为：

a = yield from coro_fib(n-2) # don't return until its ready
b = yield from coro_fib(n-1)
s = yield from coro_sum([a, b])

一般来说，为了防止指数增长，您可以使用asyncio.Queue (synchronization via communication)、asyncio.Semaphore (synchonization using mutex) 原语。

【讨论】：