守护进程
基于进程启动的子进程,会和主进程一起结束。主进程结束的依据是程序的代码执行完毕。
#创建守护进程
p=Process(task)
p.daemon = True
p.start()
子进程需要和主进程同时结束时,把子进程设置为守护进程。主进程挂了以后子进程就没有存在意义时一定要用守护进程。(如果该任务的执行周期与主进程的执行周期是一致的,那必须把子进程设置为守护进程)
开子进程的目的,并发执行任务
主进程执行完毕,不会立马死掉,要等所有的非守护的子进程都死掉,主才会死掉。主运行完最后一行代码就会执行完毕。
守护进程会在主进程执行完毕后才会死。
守护进程不允许开子进程。
进程同步
进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的, 竞争带来的结果就是错乱,如何控制,就是加锁处理。多个进程同时操作一个文件,容易出现资源抢占和数据混乱。
互斥锁
mutilprocess 的 Lock方法。
lock.acquire()
lock.release()
#上下文管理协议,with。
#with lock:
只能acquire一次,
互斥锁与join的区别,join把整体变成串行了,互斥锁知识把局部变成串行的。
#抢占一个输出终端的问题
#由并发变成了串行,牺牲了运行效率,但避免了竞争
from multiprocessing import Process,Lock
import os,time
def 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()
#模拟同时往一个数据源写入数据
#文件db的内容为:{"count":1}
#注意一定要用双引号,不然json无法识别
from multiprocessing import Process,Lock
import time,json,random
def search():
dic=json.load(open('db.txt'))
print(' 33[43m剩余票数%s 33[0m' %dic['count'])
def get():
dic=json.load(open('db.txt'))
time.sleep(0.1) #模拟读数据的网络延迟
if dic['count'] >0:
dic['count']-=1
time.sleep(0.2) #模拟写数据的网络延迟
json.dump(dic,open('db.txt','w'))
print(' 33[43m购票成功 33[0m')
def task(lock):
search()
lock.acquire()
get()
lock.release()
if __name__ == '__main__':
lock=Lock()
for i in range(100): #模拟并发100个客户端抢票
p=Process(target=task,args=(lock,))
p.start()
加锁可以保证多个进程修改同一块数据时,同一时间只能有一个任务可以进行修改,即串行的修改,没错,速度是慢了,但牺牲了速度却保证了数据安全。
虽然可以用文件共享数据实现进程间通信,但问题是:
- 效率低(共享数据基于文件,而文件是硬盘上的数据)
- 需要自己加锁处理
能够兼顾:1、效率高(多个进程共享一块内存的数据)
2、帮我们处理好锁问题。这就是mutiprocessing模块为我们提供的基于消息的IPC通信机制:队列和管道。
- 队列和管道都是将数据存放于内存中
- 队列又是基于(管道+锁)实现的,可以让我们从复杂的锁问题中解脱出来, 我们应该尽量避免使用共享数据,尽可能使用消息传递和队列,避免处理复杂的同步和锁问题,而且在进程数目增多时,往往可以获得更好的可获展性。
IPC机制
数据存放于内存、高效率、处理好了锁)
mutilprocess模块,两种实现手段,
- 管道,
- 队列(基于管道+锁实现的,先进先出)
队列
用于生成数据的一端把生成的数据或指令放到队列中,使用该数据来操作的另一端从队列中拿走数据,并执行相应操作。
队列是:先进先出,即先放进来的数据会被先拿出去。
Queue([maxsize]):创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递。
maxsize: 队列中允许的最大项数。默认无限制
put方法用以插入数据到队列中,put方法还有两个可选参数:blocked和timeout。
如果blocked为True(默认值),并且timeout为正值,该方法会阻塞timeout指定的时间,直到该队列有剩余的空间。如果超时,会抛出Queue.Full异常。如果blocked为False,但该Queue已满,会立即抛出Queue.Full异常。
get方法可以从队列读取并且删除一个元素。同样,get方法有两个可选参数:blocked和timeout。
如果blocked为True(默认值),并且timeout为正值,那么在等待时间内没有取到任何元素,会抛出Queue.Empty异常。如果blocked为False,有两种情况存在,如果Queue有一个值可用,则立即返回该值,否则,如果队列为空,则立即抛出Queue.Empty异常.
get_nowait():同q.get(False)
put_nowait():同q.put(False)
#下面几个没啥用:
empty():调用此方法时q为空则返回True,该结果不可靠,比如在返回True的过程中,如果队列中又加入了项目。
full():调用此方法时q已满则返回True,该结果不可靠,比如在返回True的过程中,如果队列中的项目被取走。
qsize():返回队列中目前项目的正确数量,结果也不可靠,理由同q.empty()和q.full()一样
cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
close():关闭队列,防止队列中加入更多数据。调用此方法,后台线程将继续写入那些已经入队列但尚未写入的数据,但将在此方法完成时马上关闭。如果q被垃圾收集,将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如,如果某个使用者正在被阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。
join_thread():连接队列的后台线程。此方法用于在调用q.close()方法之后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为
管道
#创建管道的类:
Pipe([duplex]):在进程之间创建一条管道,并返回元组(conn1,conn2),其中conn1,conn2表示管道两端的连接对象,强调一点:必须在产生Process对象之前产生管道
#参数介绍:
dumplex:默认管道是全双工的,如果将duplex射成False,conn1只能用于接收,conn2只能用于发送。
#主要方法:
conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收,recv方法会一直阻塞。如果连接的另外一端已经关闭,那么recv方法会抛出EOFError。
conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
#其他方法:
conn1.close():关闭连接。如果conn1被垃圾回收,将自动调用此方法 conn1.fileno():返回连接使用的整数文件描述符
conn1.poll([timeout]):如果连接上的数据可用,返回True。timeout指定等待的最长时限。如果省略此参数,方法将立即返回结果。如果将timeout射成None,操作将无限期地等待数据到达。
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息,超过了这个最大值,将引发IOError异常,并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭,再也不存在任何数据,将引发EOFError异常。
conn.send_bytes(buffer [, offset [, size]]):通过连接发送字节数据缓冲区,buffer是支持缓冲区接口的任意对象,offset是缓冲区中的字节偏移量,而size是要发送字节数。结果数据以单条消息的形式发出,然后调用c.recv_bytes()函数进行接收
conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息,并把它保存在buffer对象中,该对象支持可写入的缓冲区接口(即bytearray对象或类似的对象)。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间,将引发BufferTooShort异常。
实现进程间通信,与队列类似,但是需要加锁
注意:生产者和消费者都没有使用管道的某个端点,就应该将其关闭,如在生产者中关闭管道的右端,则在消费者中关闭管道的左端。
共享数据
multiprocess模块的manager方法
提供一个共享内存,不提供锁机制,也不提供封装的操作方法。只能通过加锁的方式解决。
回调机制(★)
进程池中任何一个任务一旦处理完了,就立即告知主进程:我完成了,你可以处理我的结果了。主进程则调用一个函数去处理该结果,该函数即回调函数。
我们可以把耗时间(阻塞)的任务放到进程池中,然后指定回调函数(主进程负责执行),这样主进程在执行回调函数时就省去了I/O的过程,直接拿到的是任务的结果。
用于处理回调后数据的只能单进程进行。
例如爬虫,在爬取完毕后就使用回调,回调后使用一个方法去进行解析,解析这个进程是单进程,但是最耗时的点在爬去这里,所以回调后,使用单进程处理足够。