multiprocessing模块
Process（进程）：
- 创建进程的类
-
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）强调：1. 需要使用关键字的方式来指定参数2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
- 参数
-
group参数未使用，值始终为Nonetarget表示调用对象，即子进程要执行的任务args表示调用对象的位置参数元组，args=(1,2,'egon',)kwargs表示调用对象的字典,kwargs={ 'name':'egon','age':18}name为子进程的名称 -
方法：
-
p.start()：启动进程，并调用该子进程中的p.run() p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁p.is_alive():如果p仍然运行，返回Truep.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
- 属性：
-
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置p.name:进程的名称p.pid：进程的pidp.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
- 使用：
- 注意：在windows中Process()必须放到# if __name__ == '__main__': 下
-
why?Since Windows has no fork, the multiprocessing module starts a new Python process and imports the calling module. If Process() gets called upon import, then this sets off an infinite succession of new processes (or until your machine runs out of resources). This is the reason for hiding calls to Process() insideif __name__ == "__main__"since statements inside this if-statement will not get called upon import.由于Windows没有fork，多处理模块启动一个新的Python进程并导入调用模块。 如果在导入时调用Process（），那么这将启动无限继承的新进程（或直到机器耗尽资源）。 这是隐藏对Process（）内部调用的原，使用if __name__ == “__main __”，这个if语句中的语句将不会在导入时被调用。
- 创建并开启子进程的两种方式
- 方法一
#开进程的方法一:import timeimport randomfrom multiprocessing import Processdef piao(name): print('%s piaoing' %name) time.sleep(random.randrange(1,5)) print('%s piao end' %name)p1=Process(target=piao,args=('egon',)) #必须加,号p2=Process(target=piao,args=('alex',))p3=Process(target=piao,args=('wupeqi',))p4=Process(target=piao,args=('yuanhao',))p1.start()p2.start()p3.start()p4.start()print('主线程') - 方法二
#开进程的方法二:import timeimport randomfrom multiprocessing import Processclass Piao(Process): def __init__(self,name): super().__init__() self.name=name def run(self): print('%s piaoing' %self.name) time.sleep(random.randrange(1,5)) print('%s piao end' %self.name)p1=Piao('egon')p2=Piao('alex')p3=Piao('wupeiqi')p4=Piao('yuanhao')p1.start() #start会自动调用runp2.start()p3.start()p4.start()print('主线程')方法二 - 进程直接的内存空间是隔离的
- View Code
from multiprocessing import Processn=100 #在windows系统中应该把全局变量定义在if __name__ == '__main__'之上就可以了def work(): global n n=0 print('子进程内: ',n)if __name__ == '__main__': p=Process(target=work) p.start() print('主进程内: ',n) - Process对象的join方法
- View Code
from multiprocessing import Processimport timeimport randomclass Piao(Process): def __init__(self,name): self.name=name super().__init__() def run(self): print('%s is piaoing' %self.name) time.sleep(random.randrange(1,3)) print('%s is piao end' %self.name)p=Piao('egon')p.start()p.join(0.0001) #等待p停止,等0.0001秒就不再等了print('开始')join：主进程等，等待子进程结束 - Process对象的其他方法或属性
- terminate与is_alive
#进程对象的其他方法一:terminate,is_alivefrom multiprocessing import Processimport timeimport randomclass Piao(Process): def __init__(self,name): self.name=name super().__init__() def run(self): print('%s is piaoing' %self.name) time.sleep(random.randrange(1,5)) print('%s is piao end' %self.name)p1=Piao('egon1')p1.start()p1.terminate()#关闭进程,不会立即关闭,所以is_alive立刻查看的结果可能还是存活print(p1.is_alive()) #结果为Trueprint('开始')print(p1.is_alive()) #结果为Falseterminate与is_alive - name与pid
from multiprocessing import Processimport timeimport randomclass Piao(Process): def __init__(self,name): # self.name=name # super().__init__() #Process的__init__方法会执行self.name=Piao-1, # #所以加到这里,会覆盖我们的self.name=name #为我们开启的进程设置名字的做法 super().__init__() self.name=name def run(self): print('%s is piaoing' %self.name) time.sleep(random.randrange(1,3)) print('%s is piao end' %self.name)p=Piao('egon')p.start()print('开始')print(p.pid) #查看pid
- 守护进程：
-
主进程创建守护进程
其一：守护进程会在主进程代码执行结束后就终止
其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children
注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止
- 例子
from multiprocessing import Processimport timeimport randomclass Piao(Process): def __init__(self,name): self.name=name super().__init__() def run(self): print('%s is piaoing' %self.name) time.sleep(random.randrange(1,3)) print('%s is piao end' %self.name)p=Piao('egon')p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行p.start()print('主')
-
Lock（锁）：
- 进程同步(互斥锁):
-
进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,
竞争带来的结果就是错乱，如何控制，就是加锁处理
- 并发运行,效率高,但竞争同一打印终端,带来了打印错乱
#并发运行,效率高,但竞争同一打印终端,带来了打印错乱from multiprocessing import Processimport os,timedef work(): print('%s is running' %os.getpid()) time.sleep(2) print('%s is done' %os.getpid())if __name__ == '__main__': for i in range(3): p=Process(target=work) p.start()并发运行,效率高,但竞争同一打印终端,带来了打印错乱 加锁：由并发变成了串行,牺牲了运行效率,但避免了竞争#由并发变成了串行,牺牲了运行效率,但避免了竞争from multiprocessing import Process,Lockimport os,timedef work(lock): lock.acquire() print('%s is running' %os.getpid()) time.sleep(2) print('%s is done' %os.getpid()) lock.release()if __name__ == '__main__': lock=Lock() for i in range(3): p=Process(target=work,args=(lock,)) p.start()加锁：由并发变成了串行,牺牲了运行效率,但避免了竞争 - 总结：
-
#加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。虽然可以用文件共享数据实现进程间通信，但问题是：1.效率低（共享数据基于文件，而文件是硬盘上的数据）2.需要自己加锁处理#因此我们最好找寻一种解决方案能够兼顾：1、效率高（多个进程共享一块内存的数据）2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制：队列和管道。队列和管道都是将数据存放于内存中队列又是基于（管道+锁）实现的，可以让我们从复杂的锁问题中解脱出来，我们应该尽量避免使用共享数据，尽可能使用消息传递和队列，避免处理复杂的同步和锁问题，而且在进程数目增多时，往往可以获得更好的可获展性。
-
Queue（队列）：
- 队列：
- 进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的
- 创建队列的类（底层就是以管道和锁定的方式实现）：
-
Queue([maxsize]):创建共享的进程队列，Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。
参数介绍：
-
maxsize是队列中允许最大项数，省略则无大小限制。
- 方法介绍：
-
# 主要方法q.put方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。 如果超时，会抛出Queue.Full异常。如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常。 q.get方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。如果blocked为True（默认值），并且timeout为正值，那么在等待时间内没有取到任何元素， 会抛出Queue.Empty异常。如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常. q.get_nowait():同q.get(False)q.put_nowait():同q.put(False)q.empty():调用此方法时q为空则返回True，该结果不可靠，比如在返回True的过程中，如果队列中又加入了项目。q.full()：调用此方法时q已满则返回True，该结果不可靠，比如在返回True的过程中，如果队列中的项目被取走。q.qsize():返回队列中目前项目的正确数量，结果也不可靠，理由同q.empty()和q.full()一样
-
# 其他方法1 q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞2 q.close():关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。 关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。 3 q.join_thread()：连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。 调用q.cancel_join_thread方法可以禁止这种行为
- 用法：
- View Code
'''multiprocessing模块支持进程间通信的两种主要形式:管道和队列都是基于消息传递实现的,但是队列接口'''from multiprocessing import Process,Queueimport timeq=Queue(3)#put ,get ,put_nowait,get_nowait,full,emptyq.put(3)q.put(3)q.put(3)print(q.full()) #满了print(q.get())print(q.get())print(q.get())print(q.empty()) #空了
- 生产者消费者模型：
-
生产者消费者模型
在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
-
为什么要使用生产者和消费者模式
在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
-
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。
基于队列实现生产者消费者模型
- View Code
from multiprocessing import Process,Queueimport time,random,osdef consumer(q): while True: res=q.get() if res is None:break #收到结束信号则结束 time.sleep(random.randint(1,3)) print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))def producer(q): for i in range(2): time.sleep(random.randint(1,3)) res='包子%s' %i q.put(res) print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res))if __name__ == '__main__': q=Queue() #生产者们:即厨师们 p1=Process(target=producer,args=(q,)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) #开始 p1.start() c1.start() p1.join() q.put(None) #发送结束信号 print('主')主进程在生产者生产完毕后发送结束信号None
-
-
-
#生产者消费者模型总结 #程序中有两类角色 一类负责生产数据（生产者） 一类负责处理数据（消费者） #引入生产者消费者模型为了解决的问题是： 平衡生产者与消费者之间的速度差 #如何实现： 生产者-》队列——》消费者 #生产者消费者模型实现类程序的解耦和
-
#JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。 #参数介绍： maxsize是队列中允许最大项数，省略则无大小限制。 #方法介绍： JoinableQueue的实例p除了与Queue对象相同的方法之外还具有： q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常 q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止
- 用法
from multiprocessing import Process,JoinableQueueimport time,random,osdef consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res)) q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了def producer(name,q): for i in range(10): time.sleep(random.randint(1,3)) res='%s%s' %(name,i) q.put(res) print('\033[44m%s 生产了 %s\033[0m' %(os.getpid(),res)) q.join()if __name__ == '__main__': q=JoinableQueue() #生产者们:即厨师们 p1=Process(target=producer,args=('包子',q)) p2=Process(target=producer,args=('骨头',q)) p3=Process(target=producer,args=('泔水',q)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) c2=Process(target=consumer,args=(q,)) c1.daemon=True c2.daemon=True #开始 p_l=[p1,p2,p3,c1,c2] for p in p_l: p.start() p1.join() p2.join() p3.join() print('主') #主进程等--->p1,p2,p3等---->c1,c2 #p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据 #因而c1,c2也没有存在的价值了,应该随着主进程的结束而结束,所以设置成守护进程
-
Semaphore(信号量)：
- View Code
# 互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数# 量的线程更改数据 # 信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念from multiprocessing import Process,Semaphoreimport time,randomdef go_wc(sem,user): sem.acquire() print('%s 占到一个茅坑' %user) time.sleep(random.randint(0,3)) sem.release()if __name__ == '__main__': sem=Semaphore(5) p_l=[] for i in range(13): p=Process(target=go_wc,args=(sem,'user%s' %i,)) p.start() p_l.append(p) for i in p_l: i.join() print('============》')
Event（事件）:
- View Code
# python线程的事件用于主线程控制其他线程的执行，事件主要提供了三 # 个方法 set、wait、clear。# 事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么# 当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么 # event.wait 方法时便不再阻塞。# clear：将“Flag”设置为False# set：将“Flag”设置为True #_*_coding:utf-8_*_#!/usr/bin/env pythonfrom multiprocessing import Process,Eventimport time,randomdef car(e,n): while True: if not e.is_set(): #Flase print('\033[31m红灯亮\033[0m，car%s等着' %n) e.wait() print('\033[32m车%s 看见绿灯亮了\033[0m' %n) time.sleep(random.randint(3,6)) if not e.is_set(): continue print('走你,car', n) breakdef police_car(e,n): while True: if not e.is_set(): print('\033[31m红灯亮\033[0m，car%s等着' % n) e.wait(1) print('灯的是%s，警车走了,car %s' %(e.is_set(),n)) breakdef traffic_lights(e,inverval): while True: time.sleep(inverval) if e.is_set(): e.clear() #e.is_set() ---->False else: e.set()if __name__ == '__main__': e=Event() # for i in range(10): # p=Process(target=car,args=(e,i,)) # p.start() for i in range(5): p = Process(target=police_car, args=(e, i,)) p.start() t=Process(target=traffic_lights,args=(e,10)) t.start() print('============》')Event（同线程一样）
Pool （进程池）:
- 在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：
-
- 很明显需要并发执行的任务通常要远大于核数
- 一个操作系统不可能无限开启进程，通常有几个核就开几个进程
- 进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）
例如当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。
我们就可以通过维护一个进程池来控制进程数目，比如httpd的进程模式，规定最小进程数和最大进程数...
ps：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程
-
Pool([numprocess [,initializer [, initargs]]]):创建进程池
参数介绍：
-
1 numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值2 initializer：是每个工作进程启动时要执行的可调用对象，默认为None3 initargs：是要传给initializer的参数组
方法介绍：
-
# 主要方法p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是：此操作并不会在所有池工作进程中并执行func函数。 如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async() p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。 当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。 p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成P.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用
# 其他方法方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法obj.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。obj.ready():如果调用完成，返回Trueobj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常obj.wait([timeout]):等待结果变为可用。obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数
- 同步调用- apply
from multiprocessing import Poolimport os,timedef work(n): print('%s run' %os.getpid()) time.sleep(3) return n**2if __name__ == '__main__': p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 res_l=[] for i in range(10): res=p.apply(work,args=(i,)) #同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞，但不管该任务是否存在阻塞，同步调用都会在原地等着，只是等的过程中若是任务发生了阻塞就会被夺走cpu的执行权限 res_l.append(res) print(res_l)同步调用apply 异步调用- apply_asyncfrom multiprocessing import Poolimport os,timedef work(n): print('%s run' %os.getpid()) time.sleep(3) return n**2if __name__ == '__main__': p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务 res_l=[] for i in range(10): res=p.apply_async(work,args=(i,)) #同步运行,阻塞、直到本次任务执行完毕拿到res res_l.append(res) #异步apply_async用法：如果使用异步提交的任务，主进程需要使用jion，等待进程池内任务都处理完，然后可以用get收集结果，否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了 p.close() p.join() for res in res_l: print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get异步调用apply_async - 回调函数：
-
需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数
我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。
- View Code
from multiprocessing import Poolimport requestsimport jsonimport osdef get_page(url): print(' <进程%s> get %s' %(os.getpid(),url)) respone=requests.get(url) if respone.status_code == 200: return { 'url':url,'text':respone.text}def pasrse_page(res): print(' <进程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text'])) with open('db.txt','a') as f: f.write(parse_res)if __name__ == '__main__': urls=[ 'https://www.baidu.com', 'https://www.python.org', 'https://www.openstack.org', 'https://help.github.com/', 'http://www.sina.com.cn/' ] p=Pool(3) res_l=[] for url in urls: res=p.apply_async(get_page,args=(url,),callback=pasrse_page) res_l.append(res) p.close() p.join() print([res.get() for res in res_l]) #拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了'''打印结果: <进程3388> get https://www.baidu.com <进程3389> get https://www.python.org <进程3390> get https://www.openstack.org <进程3388> get https://help.github.com/ <进程3387> parse https://www.baidu.com <进程3389> get http://www.sina.com.cn/ <进程3387> parse https://www.python.org <进程3387> parse https://help.github.com/ <进程3387> parse http://www.sina.com.cn/ <进程3387> parse https://www.openstack.org[{'url': 'https://www.baidu.com', 'text': '\r\n...',...}]''' - 爬虫案例
from multiprocessing import Poolimport time,randomimport requestsimport redef get_page(url,pattern): response=requests.get(url) if response.status_code == 200: return (response.text,pattern)def parse_page(info): page_content,pattern=info res=re.findall(pattern,page_content) for item in res: dic={ 'index':item[0], 'title':item[1], 'actor':item[2].strip()[3:], 'time':item[3][5:], 'score':item[4]+item[5] } print(dic)if __name__ == '__main__': pattern1=re.compile(r'- .*?board-index.*?>(\d+)<.*?title="(.*?)".*?star.*?>(.*?)<.*?releasetime.*?>(.*?)<.*?integer.*?>(.*?)<.*?fraction.*?>(.*?)<',re.S) url_dic={ 'http://maoyan.com/board/7':pattern1, } p=Pool() res_l=[] for url,pattern in url_dic.items(): res=p.apply_async(get_page,args=(url,pattern),callback=parse_page) res_l.append(res) for i in res_l: i.get() # res=requests.get('http://maoyan.com/board/7') # print(re.findall(pattern,res.text))爬虫案例
-
threading
-
multiprocess模块的完全模仿了threading模块的接口，二者在使用层面，有很大的相似性，因而不再详细介绍
- 用法:
- 一
#方式一from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.start() print('主线程') - 二
#方式二from threading import Threadimport timeclass Sayhi(Thread): def __init__(self,name): super().__init__() self.name=name def run(self): time.sleep(2) print('%s say hello' % self.name)if __name__ == '__main__': t = Sayhi('egon') t.start() print('主线程') -
线程与进程的区别 ：
- 开启速度
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello')if __name__ == '__main__': #在主进程下开启线程 t=Thread(target=work) t.start() print('主线程/主进程') ''' 打印结果: hello 主线程/主进程 ''' #在主进程下开启子进程 t=Process(target=work) t.start() print('主线程/主进程') ''' 打印结果: 主线程/主进程 hello ''' - pid
from threading import Threadfrom multiprocessing import Processimport osdef work(): print('hello',os.getpid())if __name__ == '__main__': #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样 t1=Thread(target=work) t2=Thread(target=work) t1.start() t2.start() print('主线程/主进程pid',os.getpid()) #part2:开多个进程,每个进程都有不同的pid p1=Process(target=work) p2=Process(target=work) p1.start() p2.start() print('主线程/主进程pid',os.getpid())瞅一瞅pid 同一进程内的线程共享该进程的数据？from threading import Threadfrom multiprocessing import Processimport osdef work(): global n n=0if __name__ == '__main__': # n=100 # p=Process(target=work) # p.start() # p.join() # print('主',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100 n=1 t=Thread(target=work) t.start() t.join() print('主',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据 - 其他方法：
-
Thread实例对象的方法 # isAlive(): 返回线程是否活动的。 # getName(): 返回线程名。 # setName(): 设置线程名。threading模块提供的一些方法： # threading.currentThread(): 返回当前的线程变量。 # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。 # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
- View Code
from threading import Threadimport threadingfrom multiprocessing import Processimport osdef work(): import time time.sleep(3) print(threading.current_thread().getName())if __name__ == '__main__': #在主进程下开启线程 t=Thread(target=work) t.start() print(threading.current_thread().getName()) print(threading.current_thread()) #主线程 print(threading.enumerate()) #连同主线程在内有两个运行的线程 print(threading.active_count()) print('主线程/主进程') ''' 打印结果: MainThread <_MainThread(MainThread, started 140735268892672)> [<_MainThread(MainThread, started 140735268892672)>,from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.start() t.join() print('主线程') print(t.is_alive()) ''' egon say hello 主线程 False ''' - 守护线程：
-
无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁
需要强调的是：运行完毕并非终止运行
-
#1.对主进程来说，运行完毕指的是主进程代码运行完毕#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕
#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。
- View Code
from threading import Threadimport timedef sayhi(name): time.sleep(2) print('%s say hello' %name)if __name__ == '__main__': t=Thread(target=sayhi,args=('egon',)) t.setDaemon(True) #必须在t.start()之前设置 t.start() print('主线程') print(t.is_alive()) ''' 主线程 True '''
-
-
Python GIL(Global Interpreter Lock)
- 略
- 递归锁 与 死锁
- 所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁
- View Code
from threading import Thread,Lockimport timemutexA=Lock()mutexB=Lock()class MyThread(Thread): def run(self): self.func1() self.func2() def func1(self): mutexA.acquire() print('\033[41m%s 拿到A锁\033[0m' %self.name) mutexB.acquire() print('\033[42m%s 拿到B锁\033[0m' %self.name) mutexB.release() mutexA.release() def func2(self): mutexB.acquire() print('\033[43m%s 拿到B锁\033[0m' %self.name) time.sleep(2) mutexA.acquire() print('\033[44m%s 拿到A锁\033[0m' %self.name) mutexA.release() mutexB.release()if __name__ == '__main__': for i in range(10): t=MyThread() t.start()'''Thread-1 拿到A锁Thread-1 拿到B锁Thread-1 拿到B锁Thread-2 拿到A锁然后就卡住，死锁了''' -
解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：
-
mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止
-
条件Condition：
- 使得线程等待，只有满足某条件时，才释放n个线程
- View Code
import threading def run(n): con.acquire() con.wait() print("run the thread: %s" %n) con.release() if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start() while True: inp = input('>>>') if inp == 'q': break con.acquire() con.notify(int(inp)) con.release() View Codedef condition_func(): ret = False inp = input('>>>') if inp == '1': ret = True return retdef run(n): con.acquire() con.wait_for(condition_func) print("run the thread: %s" %n) con.release()if __name__ == '__main__': con = threading.Condition() for i in range(10): t = threading.Thread(target=run, args=(i,)) t.start()
- 定时器Timer
- View Code
# 定时器，指定n秒后执行某操作from threading import Timer def hello(): print("hello, world") t = Timer(1, hello)t.start() # after 1 seconds, "hello, world" will be printed
-
- queue队列：
- queue队列 ：使用import queue，用法与进程Queue一样
- class
queue.Queue(maxsize=0) #先进先出 - View Code
import queueq=queue.Queue()q.put('first')q.put('second')q.put('third')print(q.get())print(q.get())print(q.get())'''结果(先进先出):firstsecondthird''' - class
queue.LifoQueue(maxsize=0) #last in fisrt out 后进先出 - View Code
import queueq=queue.LifoQueue()q.put('first')q.put('second')q.put('third')print(q.get())print(q.get())print(q.get())'''结果(后进先出):thirdsecondfirst''' - class
queue.PriorityQueue(maxsize=0) #存储数据时可设置优先级的队列 - View Code
import queueq=queue.PriorityQueue()#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高q.put((20,'a'))q.put((10,'b'))q.put((30,'c'))print(q.get())print(q.get())print(q.get())'''结果(数字越小优先级越高,优先级高的优先出队):(10, 'b')(20, 'a')(30, 'c')'''
- class
- queue队列 ：使用import queue，用法与进程Queue一样
concurrent.futures
官方文档:
#1 介绍concurrent.futures模块提供了高度封装的异步调用接口ThreadPoolExecutor：线程池，提供异步调用ProcessPoolExecutor: 进程池，提供异步调用Both implement the same interface, which is defined by the abstract Executor class.#2 基本方法#submit(fn, *args, **kwargs)异步提交任务#map(func, *iterables, timeout=None, chunksize=1) 取代for循环submit的操作#shutdown(wait=True) 相当于进程池的pool.close()+pool.join()操作wait=True，等待池内所有任务执行完毕回收完资源后才继续wait=False，立即返回，并不会等待池内的任务执行完毕但不管wait参数为何值，整个程序都会等到所有任务执行完毕submit和map必须在shutdown之前#result(timeout=None)取得结果#add_done_callback(fn)回调函数
#介绍The ProcessPoolExecutor class is an Executor subclass that uses a pool of processes to execute calls asynchronously. ProcessPoolExecutor uses the multiprocessing module, which allows it to side-step the Global Interpreter Lock but also means that only picklable objects can be executed and returned.class concurrent.futures.ProcessPoolExecutor(max_workers=None, mp_context=None)An Executor subclass that executes calls asynchronously using a pool of at most max_workers processes. If max_workers is None or not given, it will default to the number of processors on the machine. If max_workers is lower or equal to 0, then a ValueError will be raised.#用法from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,randomdef task(n): print('%s is runing' %os.getpid()) time.sleep(random.randint(1,3)) return n**2if __name__ == '__main__': executor=ProcessPoolExecutor(max_workers=3) futures=[] for i in range(11): future=executor.submit(task,i) futures.append(future) executor.shutdown(True) print('+++>') for future in futures: print(future.result())ProcessPoolExecutor #介绍ThreadPoolExecutor is an Executor subclass that uses a pool of threads to execute calls asynchronously.class concurrent.futures.ThreadPoolExecutor(max_workers=None, thread_name_prefix='')An Executor subclass that uses a pool of at most max_workers threads to execute calls asynchronously.Changed in version 3.5: If max_workers is None or not given, it will default to the number of processors on the machine, multiplied by 5, assuming that ThreadPoolExecutor is often used to overlap I/O instead of CPU work and the number of workers should be higher than the number of workers for ProcessPoolExecutor.New in version 3.6: The thread_name_prefix argument was added to allow users to control the threading.Thread names for worker threads created by the pool for easier debugging.#用法与ProcessPoolExecutor相同ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorimport os,time,randomdef task(n): print('%s is runing' %os.getpid()) time.sleep(random.randint(1,3)) return n**2if __name__ == '__main__': executor=ThreadPoolExecutor(max_workers=3) # for i in range(11): # future=executor.submit(task,i) executor.map(task,range(1,12)) #map取代了for+submitmap的用法 from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutorfrom multiprocessing import Poolimport requestsimport jsonimport osdef get_page(url): print(' <进程%s> get %s' %(os.getpid(),url)) respone=requests.get(url) if respone.status_code == 200: return { 'url':url,'text':respone.text}def parse_page(res): res=res.result() print(' <进程%s> parse %s' %(os.getpid(),res['url'])) parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text'])) with open('db.txt','a') as f: f.write(parse_res)if __name__ == '__main__': urls=[ 'https://www.baidu.com', 'https://www.python.org', 'https://www.openstack.org', 'https://help.github.com/', 'http://www.sina.com.cn/' ] # p=Pool(3) # for url in urls: # p.apply_async(get_page,args=(url,),callback=pasrse_page) # p.close() # p.join() p=ProcessPoolExecutor(3) for url in urls: p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj，需要用obj.result()拿到结果回调函数