python - 如何使用生成器形成多个管道？

Question

我正在使用 python，我正在尝试找到一种优雅地将多个生成器链接在一起的方法。问题的一个例子是有一个根生成器，它提供某种数据，每个值都像级联一样传递给它的“ child ”，这反过来可能会修改他们接收到的对象。我可以选择这条路线:

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

但是很丑，也不优雅。我正在考虑一种更实用的做事方式。

Answer 1

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

请注意，在下面的示例代码中，我使用了一个名为 coroutine 的装饰器来“初始化”生成器/协程函数。这导致它们执行到恰好在它们的第一个 yield 表达式/语句之前。它是这个 youtube 视频中显示的一个非常有启发性的演讲的一个略微修改的版本，标题为 A Curious Course on Coroutines and Concurrency Dave Beazley 在 PyCon 2009 上给出的。

正如您应该能够从生成的输出中看到的那样，数据值正在由通过单个 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)

输出:

piping 0
  gen1: 0 is divisible by 2
  gen2: 0 is divisible by 4
  gen3: 0 is divisible by 6
piping 1
piping 2
  gen1: 2 is divisible by 2
piping 3
piping 4
  gen1: 4 is divisible by 2
  gen2: 4 is divisible by 4
piping 5
piping 6
  gen1: 6 is divisible by 2
  gen3: 6 is divisible by 6
piping 7
piping 8
  gen1: 8 is divisible by 2
  gen2: 8 is divisible by 4
piping 9
piping 10
  gen1: 10 is divisible by 2
piping 11
piping 12
  gen1: 12 is divisible by 2
  gen2: 12 is divisible by 4
  gen3: 12 is divisible by 6
piping 13
piping 14
  gen1: 14 is divisible by 2
piping 15
piping 16
  gen1: 16 is divisible by 2
  gen2: 16 is divisible by 4