如何使用发电机形成多个管道？

Question

我正在使用python，并试图找到一种方法，将多个生成器优雅地连接在一起。这个问题的一个例子是有一个根生成器，它提供某种类型的数据，并且每个值都被传递给它的“子级”，就像一个级联，它反过来可以修改它们接收到的对象。我可以走这条路：

for x in gen1:
    gen2(x)
    gen3(x)

但又丑又不雅。我在想一种更实用的做事方式。在

Answer 1

Dave Beazley gave this example in a talk he did in 2008。目标是求出apacheweb服务器日志中传输的数据字节数。假设日志格式如下：

81.107.39.38 -  ... "GET /ply/ HTTP/1.1" 200 7587
81.107.39.38 -  ... "GET /favicon.ico HTTP/1.1" 404 133
81.107.39.38 -  ... "GET /admin HTTP/1.1" 403 -

传统（非生成器）解决方案可能如下所示：

^{pr2}$

使用生成器表达式的生成器管道可以可视化为：

access-log => wwwlog => bytecolumn => bytes => sum() => total

可能看起来像：

with open("access-log") as wwwlog:
    bytecolumn = (line.rsplit(None,1)[1] for line in wwwlog)
    bytes = (int(x) for x in bytecolumn if x != '-')
print("Total: {}".format(sum(bytes)))

戴夫·比兹利的幻灯片和更多的例子可供选择on his website。His later presentations elucidate this further。在

如果不知道您要做什么，很难说得更多，这样我们就可以评估您所做的每件事是否需要自定义生成器（生成器表达式/理解可以在许多情况下正常工作，而无需声明生成器函数）。在

Answer 2

管道可能更像这样：

for x in gen3(gen2(gen1())):
    print x

例如：

^{pr2}$

在Python中没有一种方法来fork（或“tee”）管道。如果需要多个管道，则必须复制它们：gen2(gen1())和{}。在

Answer 3

您可以将生成器转换为协程，这样它们就可以send()并接收彼此的值（使用(yield)表达式）。这将使每个人都有机会更改接收到的值，并/或将它们传递给下一个生成器/协同程序（或完全忽略它们）。在

{I>在下面的示例中使用名为cdi}的coroutine来编写一个名为coroutine的函数。这会导致它们执行到第一个yield表达式/语句之前。这是一个在youtube视频中显示的一个稍作修改的版本，这个视频是davebeazley在pycon2009上发表的一个非常有启发性的演讲，题目是A Curious Course on Coroutines and Concurrency。在

您应该能够从生成的输出中看到，数据值正由通过单个send()配置到头协程的每个管道进行处理，然后头协程将其有效地“多路复用”到每个管道中。由于每个子协程也这样做，所以可以建立一个详细的过程“树”。在

import sys

def coroutine(func):
    """ Decorator to "prime" generators used as coroutines. """
    def start(*args,**kwargs):
        cr = func(*args,**kwargs)  # Create coroutine generator function.
        next(cr)                   # Advance to just before its first yield.
        return cr
    return start

def pipe(name, value, divisor, coroutines):
    """ Utility function to send values to list of coroutines. """
    print('  {}: {} is divisible by {}'.format(name, value, divisor))
    for cr in coroutines:
        cr.send(value)

def this_func_name():
    """ Helper function that returns name of function calling it. """
    frame = sys._getframe(1)
    return frame.f_code.co_name


@coroutine
def gen1(*coroutines):
    while True:
        value = (yield)     # Receive values sent here via "send()".
        if value % 2 == 0:  # Only pipe even values.
            pipe(this_func_name(), value, 2, coroutines)

@coroutine
def gen2(*coroutines):
    while True:
        value = (yield)     # Receive values sent here via "send()".
        if value % 4 == 0:  # Only pipe values divisible by 4.
            pipe(this_func_name(), value, 4, coroutines)

@coroutine
def gen3(*coroutines):
    while True:
        value = (yield)     # Receive values sent here via "send()".
        if value % 6 == 0:  # Only pipe values divisible by 6.
            pipe(this_func_name(), value, 6, coroutines)

# Create and link together some coroutine pipelines.
g3 = gen3()
g2 = gen2()
g1 = gen1(g2, g3)

# Send values through both pipelines (g1 -> g2, and g1 -> g3) of coroutines.
for value in range(17):
    print('piping {}'.format(value))
    g1.send(value)

输出：

^{pr2}$