后端编程Python3-高级程序设计(协程和流水线)

本节是第五讲的第十八小节，本节继续介绍Python高级程序设计技巧（函数型程序设计、偏函数应用、协程、流水线）。

函数型程序设计（Functional-Style Programming）

函数型程序设计是一种方法。该方法中，计算过程是通过将函数结合在一起实现的，这些函数或者不修改自己的参数，或者不引用或改变程序的状态，或者将结果作为返回值。这种程序设计方法的强烈吸引力在于(理论上)，孤立地开发函数与调试函数型程序都更简单，这是因为，函数型程序设计不涉及状态改变，因此从数学角度推理并实现相关函数是可能的。 与函数型程序设计有很强关联的有3个概念：映射、过滤与降低。映射处理中， 以一个函数与 iterable为参数，产生一个新的iterable (或一个列表)，其中每一项都是对原始iterable中相应项调用该函数所产生的结果。这可以使用内置的map()函数实现，比如：

list(map(lambda x: x ** 2, [1,2, 3, 4])) # returns: [1,4, 9, 16]

map()函数以一个函数与一个iterable作为参数，出于效率的考虑，返回一个迭代子而非列表。这里，我们是强制创建一个列表，以便使结果更清晰：

[x ** 2 for x in [1,2, 3, 4]] # returns: [1,4, 9, 16]

生成器表达式通常可用于替代map(),这里，我们使用了列表内涵，从而不再需要使用list(),为使其成为一个生成器，我们只需要将外部的方括号改为圆括号。 过滤处理以一个函数与一个iterable为参数，并生成一个新的iterable.其中每一 项都来自原始的iterable—假定在相应项上调用函数时返回True,内置的filter()函数支持这种做法：

list(filter(lambda x: x > 0, [1, -2, 3, -4])) # returns: [1,3]

[x for x in [1, -2, 3, -4] if x > 0] # returns: [1,3]

filter()函数以一个函数与一个iterable为参数，并返回一个迭代子。 filter()函数总是可以使用生成器表达式或列表内涵替代。 降低处理以一个函数与一个iterable为参数，并产生单一的结果值，其具体工作过程是：对iterable的头两个值调用函数，之后对计算所得结果与第三个值调用函数，之后对计算所得结果与第四个值调用函数，依此类推，直至所有的值都进行了处理。

Functools模块的functools.reduce()函数支持这种做法，下面给出的是完成相同计算的两行代码：

functools.reduce(lambda x, y: x * y, [1,2, 3, 4]) # returns: 24

functools.reduce(operator.mul, [1,2, 3, 4]) # returns: 24

operator模块实现了所有专用于使得函数型程序设计更容易的Python操作符的对应函数，这里，在第2行，我们使用的是operator.mul()函数，而不再像第1行中使用 lambda来创建一个乘法函数。 Python还提供了一些内置的降低处理函数，包括：all()函数，该函数接收一个 iterable,如果该iterable中每个数据项在使用bool()进行处理时结果都为True,则该函数将返回True； any()函数,如果该iterable中任意数据项为True,则该函数将返回True； max()函数，该函数返回iterable中最大的数据项；min()函数，该函数返回iterable中最小的数据项；sum()函数，该函数返回iterable中所有数据项的和。 前面讲解了一些关键概念，下面看几个实例。我们首先看一些釆用两种方法来获取files列表中所有文件的大小：

functools.reduce(operator.add, (os.path.getsize(x) for x in files))

functools.reduce(operator.add, map(os.path.getsize, files))

使用map()所需要的代码通常会比等价的列表内涵或生成器表达式所需要的代码更短(除了存在条件的情况)。我们使用operator.add()作为加函数，而没有使用lambda x, y: x + y。 如果只需要统计.py文件大小，那么可以过滤非Python文件，下面给出其3种实现方式：

functools.reduce(operator.add, map(os.path.getsize,

filter(lambda x: x.endswith(".py"), files)))

functools.reduce(operator.add, map(os.path.getsize,

(x for x in files if x.endswith(".py"))))

functools.reduce(operator.add, (os.path.getsize(x) for x in files if x.endswith(".py")))

可以论证的是，第二种方法与第三种方法更好一些，因为这两种方法都不需要我们创建lambda函数，但在使用生成器表达式(或列表内涵)以及map()与filter()之间进行选择时，则纯粹取决于个人的程序设计风格偏好。

使用map()、filter()与functools.reduce()通常可以消除循环，如前面的实例所展示的。在对使用函数型语言编写的代码进行转换时，这些函数是有用的，但在Python中， 我们通常可以使用列表内涵替代map(),使用带条件的列表内涵替代filter()，很多对 functools.reduce()的使用也可以使用某种Python内置的功能函数来替代，如all()、any()、 max(), min()以及 sum()。 例如：

sum(os.path.getsize(x) for x in files if x.endswith(".py"))

上面的语句与前面3个实例完成同样的功能，但是显然更加紧凑。 除了为Python的操作符提供函数外，operator模块还提供了 operator.attrgetter()函数与operator.itemgetter()函数，前一个函数在本章前面部分已有涉及，这两个函数都返回函数，之后调用返回的函数来提取指定的属性或项。 尽管分片可用于提取列表中的某部分序列，带跳步的分片可以提取多部分序列(比如，使用L[::3]每隔一项取一项)，而operator.itemgetter()可用于提取任意部分组成的序列，比如 operator.itemgetter(4, 5, 6, 11, 18)(L)。 operator.itemgetter()返回的函数不一定马上调用之后就丢弃(就像这里所做的)，而是可以保存并作为参数传递给map()、filter() 或functools.reduce(),也可以用于字典、列表或集合内涵中。 在需要排序时，可以指定一个key函数，该函数可以是任意函数，比如lambda函数、内置的函数或方法(比如str.lower()),或由operator.attrgetter()返回的函数，比如， 假定列表L中存放带priority属性的对象，则可以以优先级顺序对列表进行排序： L.sort(key=operator.attrgetter(“priority”))。 除了已经提及的functools模块与operator模块之外，itertools模块对函数型程序设计也是有用的，比如，尽管通过连接列表在两个或多个列表上进行迭代是可能的， 但以如下方式使用itertools.chain()也是一种替代的方案：

for value in itertools.chain(data_list1, data_list2, data_list3):

total+= value

itertools.chain()函数返回一个迭代子，其中包含给定的第一个序列中的相继值，之后是第二个序列中的相继值，直至用完所有序列中的值。itertools模块还包含很多其他函数，其文档中给出了很多虽小但很有用的实例，值得阅读。

偏函数应用（Partial Function Application）

偏函数是指使用现存函数以及某些参数来创建函数，新建函数与原函数执行的功能相同，但是某些参数是固定的，因此调用者不需要传递这些参数。下面给出一个非常简单的实例：

enumerate1 = functools.partial(enumerate, start=1)

for lino, line in enumerate1(lines):

process_line(i, line)

第一行代码创建了新函数enumerate1(),该函数包装了给定的函数(enumerate()) 以及一个关键字参数(start=1),以便在调用enumerate1()时，实际上是使用固定的参数以及调用时给定的任何其他参数(这里是lines)来调用原始函数。这enumerate1() 函数用于提供常规的行计数，从1开始。 使用偏函数可以简化代码，尤其是在需要多次使用同样的参数调用同样函数的时候，比如，在每次调用open()来处理UTF-8编码的文本文件时，不需要指定模式参数与编码参数，而是可以创建两个函数带有的参数，如下所示：

reader = functools.partial(open, mode="rt", encoding=“utf8")

writer = functools.partial(open,mode="wt", encoding="utf8")

现在，我们可以调用reader(filename)来打开文本文件进行读操作，调用writer (filename)进行写操作。 偏函数的一个非常常见的应用是在GUI (图形用户界面)程序设计中，在GUI程序中，在某组按钮中的任何一个被按下时，调用某个特定函数对其进行处理是非常方便的，比如：

loadButton = tkinter.Button(frame, text="Load",command=functools.partial(doAction, "load"))

saveButton = tkinter.Button(frame,text="Save",command=functools.partial(doAction, "save"))

这一实例使用了作为Python标准附带的GUI库tkinter, tkinter.Button类用于按钮 ——这里创建了两个，都包含在相同的框架中，每个都带有文本指明其用途。每个按钮的命令参数被设置为一个函数，在该按钮被按下时，tkinter必须调用该函数，这里是doAction()函数，我们使用了偏函数来确保赋予doAction()函数的第一个参数是一个字符串，该字符串用于指明是哪个按钮调用了本函数，以便确定应该执行的操作。

协程（Coroutines）

协程也是一种函数，特点是其处理过程可以在特定点挂起与恢复。因此，典型情况下，协程将执行到某个特定语句，之后执行过程被挂起等待某些数据。在这个挂起点上，程序的其他部分可以继续执行(通常是没有被挂起的其他协程)。一旦数据到来， 协程就从挂起点恢复执行，执行一些处理(很可能是以到来的数据为基础)，并可能将其处理结果发送给另一个协程。协程可以有多个入口点与退出点，因为挂起与恢复执行的位置都不止一处。 在需要将多个函数应用于同样的数据时，或需要创建数据处理流水线时，或需要某个主数带有几个从函数时，协程是有用的。协程也可以作为线程的替代，并且更简单，负载也更低。Python Package Index (pypi.python.org/pypi)提供了一些基于协程的包，这些包提供了轻量级的线程功能。 在Python中，协程是一个从yield表达式中提取输入的函数，也可以将处理结果发送给接收者函数(该函数本身必须是一个协程)。协程在执行到yield表达式的任何时候都必然挂起等待数据，而一旦数据到达，协程就从该挂起点恢复执行。协程可以有不止一个yield表达式——尽管我们这里给出的协程实例都只有一个。

对数据执行独立操作

如果我们需要对某些数据执行一组独立的操作,常规的方法是依次执行每个操作。 这种方法的不足之处是，如果某个操作很慢，则整个程序在继续执行下一个操作之前 必须等待这一操作结束。一个更有效的解决方案是使用协程：将所有操作都实现为协 程，之后同时执行所有这些协程。这样如果某个协程慢，也不会影响到其他协程—— 至少也不会耗尽所有待处理数据——因为所有操作都是独立执行的。 如图所示为在并发处理时使用协程的情况。从该图中可以看出，三个协程(假定每个完成不同工作)处理两个相同的数据项——并用不同的总时间来完成自己的工作。其中，coroutine1()工作相当快，coroutine2()工作较慢，coroutine3()则根据情况呈多样化。在三个协程都接受了初始数据之后，如果某个在等待(因为该协程首先执行完)，其他协程继续工作，这可以让处理器空闲时间最小化。使用协程完成工作后，可以对每个协程调用close()函数，这将阻止协程继续等待数据，这也意味着协程不会再消耗处理器时间。

如图发送两个数据项到三个协程中：

步骤 行为 coroutine1() coroutine2() coroutine3()

1 创建协程 等待 等待 等待

2 coroutine1.send("a") 进程"a" 等待 等待

3 coroutine2.send("a") 进程"a" 进程"a" 等待

4 coroutine3.send("a") 等待 进程"a" 进程"a"

5 coroutine1.send("b") 进程"b" 进程"a" 进程"a"

6 coroutine2.send("b") 进程"b" 进程"a"("b"挂起) 进程"a"

7 coroutine3.send("b") 等待 进程"a" ("b"挂起) 进程"b"

8 等待 进程"b" 进程"b"

9 等待 进程"b" 等待

10 等待 进程"b" 等待

11 等待 等待 等待

12 coroutineN.close() 结束 结束 结束

在Python中，要想创建协程，只需要创建包含至少一个yield表达式(通常在一 个无限循环中)的函数即可。到达该yield表达式后，协程的执行被挂起并等待数据； 收到数据后，协程又从该yield表达式处恢复处理，处理完毕后又循环回到该yield表达式等待更多数据。当某个或多个线程挂起等待数据时，其他协程可以继续执行。与顺序逐个执行函数相比，这种方法可以提供更高的效率。 下面通过将几个正则表达式应用于一组HTML文件，来展示在实践中怎样执行独立的操作。操作的目标是输出每个文件的URL与一级、二级标题。我们从查看每个正则表达式开始，之后是创建正则表达式的协程“匹配器"，再之后将查看每个线程及其如何使用。

URL_RE = re.compile(r“”“href=(?P['"])(?P[^\l]+?)"""

r"""(?P=quote)""",re.IGNORECASE)

flags = re.MULTILINE|re.IGNORECASE|re.DOTALL

H1_RE = re.compile(r"(?P.+?)",flags)

H2_RE = re.compile(r"(?P.+?)",flags)

这些正则表达式(后面将直接称之为regexes)会匹配HTML href的URL,以及包含在标题标签与之间的文本(正则表达式将在后面进行讲解，但这对理解这个实例并没有实际影响)。

receiver = reporter()

matchers = (regex_matcher(receiver, URL_RE),regex_matcher(receiver, H1_RE), regex_matcher(receiver, H2_RE))

由于协程总是包含一个yield表达式，因此协程也是生成器。所以，这里我们创建匹配器协程，实际上也就创建了生成器。每个regex_matcher()都是一个协程，以接受者函数(本身也是一个协程)与regex为参数并进行匹配。匹配器匹配到适当的对象后，会将匹配发送给接收者。

@coroutine

def regex_matcher(receiver, regex):

while True:

text = (yield)

for match in regex.finditer(text):

receiver.send(match)

匹配器首先进入一个无限循环，之后立即挂起，等待yield表达式返回一个文本并与正则表达式进行匹配。收到文本后，匹配器对每个匹配进行迭代，并将其发送到接收者。匹配完毕后，协程返回到yield表达式，再一次挂起等待更多的文本。 对(未修饰)的匹配器而言，存在一个小问题——初次创建时，应该着手执行， 以便推进到yield表达式语句等待第一个文本。为此，我们可以调用每个协程内置的next()函数。但为方便起见，这里已经创建了@coroutine修饰器来完成这一任务。

def coroutine(function):

@functools.wraps(function)

def wrapper(*args, **kwargs):

generator = function(*args, **kwargs)

next(generator)

return generator

return wrapper

修饰器@coroutine为协程函数调用内置的next()函数——这会导致协程函数执行到第一个yield表达式，并做好接收数据的准备。 在了解了匹配器协程之后，我们将讲解匹配器如何使用，之后再讲解用于接收匹配器输出的reporter()协程。

try:

for file in sys.argv[1:]:

print(file)

html = open(file, encoding="utf8").read()

for matcher in matchers:

matcher.send(html)

finally:

for matcher in matchers:

matcher.close()

receiver.close()

程序读入命令行中列出的文件名，打印出每个文件名，之后将该文件的整个文本读入到html变量中(UTF-8编码)。之后，程序对所有匹配器(本例中是3个)进行迭代，并将相应的文本发送给每个匹配器。每个匹配器独立运行，并将每一次匹配的结果发送给报告器协程。最后，对每个匹配器协程和报告器协程都调用close()终止这些协程，否则这些协程会继续(挂起)等待文本(报告器协程则是等待匹配)，因为其中包含无限循环。

@coroutine

def reporter()：

ignore = frozenset({"style.css", "favicon.png", "index.htm"})

while True:

match = (yield)

if match is not None:

groups = match.groupdict()

if "url" in groups and groups["url"] not in ignore:

print("URL:", groups["url"])

elif "h1" in groups:

print("H1: ”,groups["h1"])

elif "h2" in groups:

print("H2: “,groups["h2"])

reporter()协程用于输出结果。这是通过前面我们看到的语句receiver = reporter()实现的，并以receiver参数的形式传递给每个匹配器。reporter()协程等待(并挂起)有匹配发送过来，之后打印出每个匹配的详细信息，之后再进入无限循环中的等待—— 只有在调用了close()之后才会终止。 像前面这种方式使用协程会有性能上的优势，但我们也需要考虑一些不同的方法来完成处理任务。

构成流水线

有时候，创建数据处理流水线是有用的。简单地说，流水线就是一个或多个函数的组合，数据项被发送给第一个函数，该函数或者丢弃该数据项(过滤掉)，或者将其传送给下一个函数（以某种方式）。第二个函数从第一个函数接收数据项，重复处理过程，丢弃或将数据项（可能转换为不同的形式）传递给下一个函数，以此类推。最后，数据项在经过多次处理之后以某种形式输出。 典型情况下，流水线包含几个组件，一个获取数据，一个或多个过滤或转换数据， 一个用于输出结果。确切地说，这实际上是前面某节中我们讲过的函数型程序设计方法，那里我们还讲述了Python的一些内置函数，比如filter()与map()。

使用流水线的一个好处是，可以递增式地读入数据项，通常是每次一个，这也需要给流水线提供正好足够的数据项来使其充满（通常是每个组件一个或几个数据项）。 与一次性地将整个数据集读入内存相比，这种方法可以极大地节省内存开销。 如图所示为一个简单的三步骤流水线。流水线的第一个组件（get_data()）依次 获取每个待处理的数据项；第二个组件（process()）对数据进行处理---可以丢弃不需要的数据项---当然，可以存在任意数量的其他处理/过滤组件；最后一个组件 （reporter()）用于输出结果。在该图中，可以看到，数据项"a"、"b"、"c”、"e"与"f"进行了处理并产生输出信息，而数据项"d”则被丢弃。

歩骤 行为 get_data() process () reporter()

1 pipeline = get_data(process(reporter())) 等待 等待 等待

2 pipeline.send("a") 读"a" 等待 等待

3 pipeline.send("b") 读"b" 进程"a" 等待

4 pipeline.send("c") 读"c" 进程"b" 输岀"a"

5 pipeline.send("d") 读"d" 进程"c" 输出"b"

6 pipeline.send("e") 读"e" 丢弃"d" 输岀"c"

7 pipeline, send ("f") 读"f" 进程"e" 等待

8 等待 进程”f” 输岀"e"

9 等待 等待 输岀"f"

10 等待 等待 等待

11 关闭协程 结束 结束 结束

图中展示的流水线是一个过滤器，因为每个数据项在传递之后并没有改变， 而只是或者丢弃，或者以原始形式输出。流水线端点也扮演同样的角色：获取数据， 之后输出结果。但在二者之间，可以根据需要有任意多的组件，每个组件执行过滤或 （与）转换功能。有些情况下，以不同顺序对组件进行组合可以产生完成不同任务的流水线。 我们首先査看一个理论上的实例，以便更好地理解基于协程的流水线的工作方式， 之后讲解一个实际的实例。 假定有一个浮点数序列，我们需要在一个多组件流水线中对其进行处理，以便将每个浮点数转换成为一个整数(四舍五入)，但丢弃那些超出范围的数(大于10,或 者小于0)。如果有4个协程组件，acquire()(用于获取一个数)、to_int()(通过四舍五入将一个浮点数转换为整数)、check()(对在范围内的数向下传递，超出范围的数则直接丢弃)、output()(输出一个数)，我们可以按如下的方式创建流水线。 pipe = acquire(to_int(check(output()))) 之后，通过调用pipe.send()将数据发送到流水线，这里我们看一下流水线对浮点数4.3和9.6的处理过程，使用了与前面步骤图中不同的表达形式： pipe.send(4.3)->acquire(4.3) ->to_int(4.3)->check(4) ->output(4) pipe.send(9.6)->acquire(9.6) ->to_int(9.6)->checks(10) 注意的是，浮点数9.6处理后没有输出信息，这是因为check()协程接收到的是10, 这超出了允许的数据范围，因此被过滤掉了。 接下来我们创建一个不同的流水线，但使用同样的组件，再看会有哪些处理上的变化。 pipe = acquire(check(to_int(output()))) 该流水线在转换(to_int())之前进行过滤(check())，下面展示的是该流水线对4.3与 9.6的处理过程。 pipe.send(4.3) ->acquire(4.3) ->check(4.3) ->to_int(4.3) ->output(4) pipe.send(9.6)->acquire(9.6) ->check(9.6) ->to_int(9.6) ->output(10) 这里，虽然已经超出了范围，但错误地输出了10。这是因为，check()组件先运行， 而此时接收到的是范围内的值9.6,因此简单地向前传递，但是to_int()组件对其接收到的数据进行四舍五入。 下面，我们将查看一个具体的实例----一个文件匹配器，该匹配器读入命令行中给定的所有文件名(也包括命令行中给定目录中递归的文件名)，并输出符合特定标准的文件名的绝对路径。 我们首先看一下流水线的构成方式，之后再看构成流水线组件的协程，下面给出的是最简单的流水线。 pipeline = get_files(receiver) 该流水线打印每个给定的文件名(或递归打印给定目录中的每个文件名)。 get_files()函数是一个协程，提供文件名；接受者是一"reporter()协程----使用语句

receiver =reporter()创建功能是简单地打印出收到的每个文件名。该流水线机会没有比os.walk()函数多做什么(实际上使用了这个函数)，但通过对该流水线的一些组件进行重新组合，可以产生更高级的流水线。 pipeline = get_files(suffix_matcher(receiver, (".htm", "html"))) 该流水线是通过组合get_files()与suffix_matcher()这两个协程实现的，其功能只是打印HTML文件。 类似于这种形式的协程组合可能很快就会变得难于读懂，但这并不能阻止我们釆用这种方式构造多阶段的流水线——尽管这种方法中，我们必须使用从最后到最先的方式创建组件。 pipeline = size_matcher(receiver, minimum=1024 ** 2) pipeline = suffix_matcher(pipeline, (".png", ".jpg", ".Jpeg")) pipeline = get_files(pipeline) 该流水线匹配那些至少1MB大小并且后缀名表明是图像的文件。 这些流水线怎样使用? 我们只需要简单地向其提供文件名或路径，其他工作流水线会自己完成。 for arg in sys.argv[1:]: pipeline.send(arg) 值得注意的是，使用的是哪个流水线并不重要——可以是打印所有文件的那个， 也可以是打印HTML文件的那个，也可以是只关注图像文件的那个——这几个流水线的工作方式都是一致的。在本例中，三个流水线都是过滤器——收到文件之后，或者向前传递给下一个组件(本例中是reporter(),处理方法是打印出来)，或者因为不符合特定标准而被丢弃。 在查看get_files()与匹配器协程之前，我们先看一下简单的reporter()协程(作为接受者传递)，该协程用于输出结果。

@coroutine

def reporter()：

while True:

filename = (yield)

print(filename)

这里，我们使用前面创建的同样的@coroutine修饰器。 get_files()协程实质上是os.walk()函数的一个包裹器,其预期输入是路径或文件名。

@coroutine

def get__files(receiver):

while True:

path = (yield)

if os.path.isfile(path):

receiver.send(os.path.abspath(path))

else:

for root, dirs, files in os.walk(path):

for filename in files:

receiver.send(os.path.abspath(os.path.join(root, filename)))

这一协程的结构现在我们已经很熟悉了： 一个无限循环，在其中等待yield表达式 返回一个可以处理的值，之后将结果发送给接收者。

@coroutine

def suffix_matcher(receiver, suffixes):

while True:

filename = (yield)

if filename.endswith(suffixes):

receiver.send(filename)

这一协程看起来很简单——实际上也简单——要注意的是，该协程只发送那些与后缀匹配的文件名，而对那些不匹配的则过滤出流水线。

@coroutine

def size_matcher(receiver, minimum=None, maximum=None):

while True:

filename = (yield)

size = os.path.getsize(filename)

if ((minimum is None or size >= minimum) and (maximum is None or size <= maximum)):

receiver.send(filename)

这一协程几乎与suffix_matcher()—样，区别在于本协程过滤的是那些大小不在范围之内的文件，而不是后缀名不匹配的文件。 我们创建的流水线存在两个问题。一个问题是没有对协程进行关闭，在本例中这没有影响，因为一旦处理结束程序就会终止，但协程执行完毕后就将其关闭是一个好习惯。另一个问题是，在流水线的不同阶段，可能会向操作系统(底层的)査询同一 个文件的不同信息段——这可能会比较慢。一个解决方案是修改get_files()协程，使其对每个文件返回一个二元组(filename, os.stat())，而不仅仅是文件名，之后将该二元组在流水线中传递。这意味着，我们将得到每个文件的所有相关信息。在本章后面的 练习中，可以尝试解决这两个问题，并添加一些额外的功能。 创建在流水线中使用的协程需要一些新的思维方式。然而，对灵活性而言，使用协程是非常有利的，并且，对大数据集而言，可以降低内存中存储的数据量，并有更快的吞吐率。

以上内容部分摘自视频课程05后端编程Python18高级程序设计（协程和流水线），更多实操示例请参照视频讲解。跟着张员外讲编程，学习更轻松，不花钱还能学习真本领。