首页 > 代码库 > Python 多进程实战 & 回调函数理解与实战

Python 多进程实战 & 回调函数理解与实战

这篇博文主要讲下笔者在工作中Python多进程的实战运用和回调函数的理解和运用。

 

多进程实战

 

实战一、批量文件下载

从一个文件中按行读取 url ,根据 url 下载文件到指定位置,用多进程实现。

 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
#!/usr/local/python27/bin/python2.7
from multiprocessing import Process,Pool
import os,time,random,sys
import urllib
 
# 文件下载函数
def filedown(url,file):
    urllib.urlretrieve(url,file)
 
 
 
if __name__ == ‘__main__‘:
    = Pool(100)
    count = 0
    # 打开存有url的文件
    = open(‘11.csv‘,‘r‘)
    while True:
        count += 1
        # 按行读取
        url1 = f.readline()
        # 当文件读取完毕时,跳出循环
        if url1 == ‘‘:
            break;
        url = url1.strip()
        file = (‘/root/tuchao/d2/work/strfile/‘+url.split(‘/‘)[4])
        print(count)
        # 使用异步多进程的方式,启动子进程,并将功能函数和参数传入.
        # 注意: 这里的 args 必须传参数列表,就算是一个参数,也得写逗号结尾。
        p.apply_async(filedown, args=(url,file,))
    p.close()
    p.join()

 

 

实战二、批量文本处理。  

读取一个目录下的每个文件,过滤掉文件中的数字和中文,把每个英语单词提取出来写入 Mongodb。

使用多进程处理

 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
#!/usr/local/python27/bin/python2.7
import re
import sys
import os
import pymongo
from multiprocessing import Process,Pool
import time
 
# Mongodb 连接,验证身份
conn = pymongo.MongoClient(‘localhost‘,27017)
conn.words.authenticate(‘words_user‘,‘woiu32k32x01‘)
db = conn.words
 
# 单词处理函数
def wordsevent(filename,mongo_insert):
    with open(filename) as f:
        wordsall=[]
        for line in f:
        # 把当前行转为小写后,判断里面是否包含小写字母。 有,表示这行是英文行,则做单词提取。  没有,表示当前行是数字或者是中文,不做处理,continue 进入下一次循环。
            if line.lower().islower():
             
         # 单词提取 re.findall 多重匹配。(r‘(\w|\‘)+)‘ 表示匹配字母或者单引号出现一次或多次。这样会出现一个问题,提取出来的单词都会拆分成一个一个字母,因为正则会安装括号里面的规则去提取,\w 按字母匹配的,所以会提取字母。  
         # 所以才要这样写 (r‘((?:\w|\‘)+)‘ 这里 ?: 写在括号的里面,表示此括号的规则只做匹配,而不提取内容。 外面还有一层括号,所以正则将会提取外面这层括号匹配的内容。 那就是一个个的单词了。       
         # 在正则中一对括号表示一组。       
                wordslist = re.findall(r‘((?:\w|\‘)+)‘,line)
                # 列表合并,把多个list合并到一个。
                wordsall.extend(wordslist)
            else:
                continue
        # 把list转成集合去重,因为集合中的元素是 确定性、无序性、互异性    
        s1=set(wordsall)
        if len(s1) == 0:
            pass
        else:
            mongo_insert(s1)
 
             
def mongo_insert(x):
    db.test2.insert_many([{"word":i} for in x])
 
 
if __name__ == ‘__main__‘:
    fileall=os.listdir(‘strfile‘)
    = Pool(10)
    count = 0
    for in fileall:
        count += 1
        filename = (‘/root/tuchao/d2/work/strfile/%s‘ % i)
        print(count,filename)
        # 启动异步多进程
        p.apply_async(wordsevent,args=(filename,mongo_insert,))
    p.close()
    p.join()

 

回调函数

 

什么是回调函数?  (第一次听说回调函数的同学,请认真看下补课)

 

编程分为两类:系统编程(system programming)和应用编程(application programming)。所谓系统编程,简单来说,就是编写库;而应用编程就是利用写好的各种库来编写具某种功用的程序,也就是应用。系统程序员会给自己写的库留下一些接口,即API(application programming interface,应用编程接口),以供应用程序员使用。所以在抽象层的图示里,库位于应用的底下。

当程序跑起来时,一般情况下,应用程序(application program)会时常通过API调用库里所预先备好的函数。但是有些库函数(library function)却要求应用先传给它一个函数,好在合适的时候调用,以完成目标任务。这个被传入的、后又被调用的函数就称为回调函数(callback function)。

打个比方,有一家旅馆提供叫醒服务,但是要求旅客自己决定叫醒的方法。可以是打客房电话,也可以是派服务员去敲门,睡得死怕耽误事的,还可以要求往自己头上浇盆水。这里,“叫醒”这个行为是旅馆提供的,相当于库函数,但是叫醒的方式是由旅客决定并告诉旅馆的,也就是回调函数。而旅客告诉旅馆怎么叫醒自己的动作,也就是把回调函数传入库函数的动作,称为登记回调函数(to register a callback function)


技术分享

 

可以看到,回调函数通常和应用处于同一抽象层(因为传入什么样的回调函数是在应用级别决定的)。而回调就成了一个高层调用底层,底层再过头来用高层的过程。

回调机制的优势

从上面的例子可以看出,回调机制提供了非常大的灵活性。请注意,从现在开始,我们把图中的库函数改称为中间函数了,这是因为回调并不仅仅用在应用和库之间。任何时候,只要想获得类似于上面情况的灵活性,都可以利用回调。

这种灵活性是怎么实现的呢?乍看起来,回调似乎只是函数间的调用,但仔细一琢磨,可以发现两者之间的一个关键的不同:在回调中,我们利用某种方式,把回调函数像参数一样传入中间函数。可以这么理解,在传入一个回调函数之前,中间函数是不完整的。换句话说,程序可以在运行时,通过登记不同的回调函数,来决定、改变中间函数的行为。这就比简单的函数调用要灵活太多了。


作者:桥头堡
链接:https://www.zhihu.com/question/19801131/answer/27459821
来源:知乎

 

是不是还没太明白,只是大概有点了解咋回事了。  别急看下面代码。

 

一个简单的回调函数的程序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#!/usr/local/python27/bin/python2.7
 
def a(i):
    print("this is a start")
    print(i)
    print("this is a stop")
 
 
def b(func):
    print("this is b start")
    for in range(10):
        func(i)
    print("this is b stop")
 
if __name__ == ‘__main__‘:
    b(a)

输出如下:

技术分享

 

一个使用多进程结合回调函数的示例程序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#!/usr/local/python27/bin/python2.7
from multiprocessing import Process,Pool
 
 
def a(x):
    print("this is a start")
    print(x)
    print("this is a stop")
 
 
def b(num):
    return(num)
 
 
if __name__ == ‘__main__‘:
    = Pool(5)
    for in range(10):
    # 这里表示,当b函数执行完成之后就会调用a函数,并且把b函数的返回值传给a函数。
        p.apply_async(b, args=(i,), callback=a)
    p.close()
    p.join()

输出如下:

技术分享

 

多进程结合回调函数写文件的示例程序

http://blog.csdn.net/Q_AN1314/article/details/51923022

 

相信现在差不多明白了吧,还不明白的再返回上面看看理论。 理解也不是难事了。

转自别处。

 

Python 多进程实战 & 回调函数理解与实战