【第二十一天】filter(),reduce(),并行处理

2.filter函数

和map()函数一样，内置函数filter()的第一个参数也是一个函数对象
它也将这个函数对象作用于可循环对象的多个元素
如果函数对象返回的是True,则该次的元素被放到返回迭代器中
也就是说，filter()通过调用参数来筛选数据

下面是使用filter()函数的一个例子
作为参数的larger100()函数用于判断元素是否比100大：

def larger100(a):
    if a > 100:
        return True
    else:
        return False

for item in filter(larger100,[10,56,101,500]):
    print(item)

101
500

类似的，filter()用于多参数的函数时
也可以在参数中增加更多的可循环对象
总的来说，map()函数和filter()函数的功能有相似的地方
都是把同一个函数应用于多个函数

3.reduce()函数

函数reduce()也是一个常见的高阶函数，函数reduce()在标准库的functools包中
使用之前需要引入，和map(),filter()一样，reduce()函数的第一个参数是函数
但reduce()对作为参数的函数对象有一个特殊要求
就是这个作为参数的函数必须能接收两个参数
reduce()可以把函数对象累进的作用于各个参数
这个功能可以用一个简单的例子来说明：

from functools import reduce

data_list = [1,2,5,7,9]
result = reduce(lambda x,y:x+y,data_list)
print(result)

24

函数reduce()的第一个参数是求和的sum()函数
它接收两个参数x和y，在功能上，reduce()累进的运算传给它的二参函数
上一次运算的结果将作为下一次调用的第一个参数
首先，reduce()将用表中的前两个元素1和2做sum()函数的参数，得到3
该返回值3将作为sum()函数的第一个参数，而将表中的下一个元素5作为sum()函数的第二个参数
进行下一次求和得到8，以此类推，直到表中元素耗尽，返回累进的运算结果
相当于(((1+2)+5)+7)+9,结果为24

拓展：
函数reduce()通过某种形式的二元运算，把多个元素收集起来
形成一个单一的结果，上面的map(),reduce()函数都是单线程的
所以运行效果和循环差不多，但map(),reduce()可以方便地移植到并行化的运算环境中
在并行运算中reduce运算紧接着map运算，map运算的结果分布在多个主机上
reduce运算把结果收集起来，因此，谷歌用于并行运算的软件架构，就称为MapReduce

4.并行处理

下面的程序就是在多进程条件下使用了多线程的map()方法
这段程序多线程地下载同一个URL下的资源
程序使用第三方包requests来进行HTTP下载：

import time
from multiprocessing import Pool
import requests

def decorator_timer(old_function):
    def new_function(*arg,**dict_arg):
        t1 = time.time()
        result = old_function(*arg,**dict_arg)
        t2 = time.time()
        print('time:',t2 - t1)
        return result
    return new_function    ＃这是前面提到的计算程序运行时间的函数
    
def visit_once(i,address='http://www.cnblogs.com'):
    r = requests.get(address)
    return r.status_code   ＃获取网页状态码
    
@decorator_timer
def single_thread(f,counts):
    result = map(f,range(counts))
    return list(result)    ＃以列表形式输出结果

@decorator_timer
def multiple_thread(f,counts,process_number=3)
    p = Pool(process_number)
    return p.map(f,range(counts))

if __name__ == '__main__':
    TOTAL = 5
    print(single_thread(visit_once,TOTAL))
    print(multiple_thread(visit_once,TOTAL))

（相关进程与线程可以参考：https://thief.one/2016/11/23/Python-multiprocessing/）

在上面的程序中，我们启动了3个进程，并行处理5个下载需求
这里把单个下载过程描述为一个函数，即visit_once()
然后用多线程的map()方法，把任务分配给3个进程
从结果可以看到运行时间可以大为缩短