守护进程
主进程创建守护进程
其一:守护进程会在主进程代码执行结束后就终止
其二:守护进程内无法再开启子进程,否则抛出异常:AssertionError: daemonic processes are not allowed to have children
注意:进程之间是互相独立的,主进程代码运行结束,守护进程随即终止
import multiprocessing
import time
def taks_one():
print("zi run")
time.sleep(3)
print("zi over")
if __name__ == '__main__':
print("父 start")
p = multiprocessing.Process(target=taks_one)
p.daemon = True # 在进程开启前,设置守护进程
p.start()
time.sleep(5)
print("父 over")
以上代码中,主进程相当于被守护进程,子进程相当于守护进程
进程同步的安全问题
问题:
import multiprocessing
import time
def task1():
for i in range(1000):
print("------------name is nick")
print("------------gender is girl")
print("------------age is 18")
def task2():
for i in range(1000):
print("++++++++++++++name is bgon")
print("++++++++++++++gender is oldman")
print("++++++++++++++age is 48")
if __name__ == '__main__':
pass
p1 = multiprocessing.Process(target=task1)
p2 = multiprocessing.Process(target=task2)
p1.start()
p2.start()
执行以上代码时,可能会存在task1函数中的打印和task2函数中的打印交叉的情况。如果将task1中的打印看做是一次完整的打印,这样就会造成task1中只打印了一半,然后就会打印task2中的信息。对于我们而言,这并不是我们想要看到的结果。
解决办法一:加join,使其变为串行
import multiprocessing
import time
def task1():
for i in range(1000):
print("------------name is nick")
print("------------gender is girl")
print("------------age is 18")
def task2():
for i in range(1000):
print("++++++++++++++name is bgon")
print("++++++++++++++gender is oldman")
print("++++++++++++++age is 48")
if __name__ == '__main__':
pass
p1 = multiprocessing.Process(target=task1)
p2 = multiprocessing.Process(target=task2)
p1.start()
p1.join()
p2.start()
优点:
- 解决了进程安全问题
缺点:
- 效率变低
解决办法二:增加互斥锁
import multiprocessing
import time,random
# 创建锁
mutext = multiprocessing.Lock()
def task1(mutext):
# 并发
for i in range(1000):
print(1)
# 串行
mutext.acquire() # 加锁
print("------------name is nick")
time.sleep(random.random())
print("------------gender is girl")
print("------------age is 18")
mutext.release() # 解锁
def task2(mutext):
# 并发
for i in range(1000):
print(2)
# 串行
mutext.acquire() # 加锁
print("++++++++++++++name is bgon")
time.sleep(random.random())
print("++++++++++++++gender is oldman")
print("+++++++++++++age is 48")
mutext.release() # 解锁
if __name__ == '__main__':
p1 = multiprocessing.Process(target=task1,args=(mutext,)) # 传参(重点)
p2 = multiprocessing.Process(target=task2,args=(mutext,)) # 传参(重点)
p1.start()
p2.start()
锁其实只是给执行代码加了限制,本质是一个标志为True或False
lock = False
def task1():
global lock
if lock = False:
lock = True
# 代码块1
lock = False
def task2():
global lock
if lock = False:
lock = True
# 代码块2
lock = False
优点:
- 可以让子进程中的部分代码变成串行,这样解决了进程安全问题
- 没有加锁的部分,依然可以并发。用户选择权更多
PS:需要注意,如果你想要多个子进程中的部分代码串行,必须是同一把锁才能实现。
进程通信
a进程如果想要和b继承通信,应该怎么做?
-
方案一:创建一个共享文件
缺点:硬盘上数据,读写I/O不高,效率低
优点:理论上交换的数据量可以非常大
适用场景:交互不频繁,且数据量较大的情况。
-
方案二:共享内存(重点)
缺点:数据量不能太大
优点:效率高
适用场景:交互频繁,但是数据量小
-
方案三:管道
缺点:基于文件,它是单向的,比较复杂
-
方案四:socket
缺点:编程复杂,用在这个地方有些大材小用
适用场景:基于网络交换数据
今天主要说一下,第二种方案,
第一种方法:Manager
应用:
import multiprocessing
def task1(data):
data["age"] = 19
if __name__ == '__main__':
data = {"age":20}
m = multiprocessing.Manager() # 创建Manager对象
m_data = m.dict(data) # 创建公共字典,除了字典外,还可以共享很多其他的数据类型
p = multiprocessing.Process(target=task1,args=(m_data,))
p.start()
p.join() # 必须要加上join()
print(m_data)
print(data)
{'age': 19}
{'age': 20}
内存的第二种方法:队列(推荐使用)
进程彼此之间互相隔离,要实现进程间通信(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()一样
其他方法(了解):
q.cancel_join_thread():不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
q.close():关闭队列,防止队列中加入更多数据。调用此方法,后台线程将继续写入那些已经入队列但尚未写入的数据,但将在此方法完成时马上关闭。如果q被垃圾收集,将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如,如果某个使用者正在被阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。
q.join_thread():连接队列的后台线程。此方法用于在调用q.close()方法之后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为
应用:
import multiprocessing
def task1(data):
d = data.get()
print(d)
if __name__ == '__main__':
q = multiprocessing.Queue() # 创建队列
for i in range(1,11):
q.put(i) # 向队列中存放数据
print(q.empty()) # 打印队列是否是空,False
a_list = []
# 消费队列中的数据
for i in range(10):
p = multiprocessing.Process(target=task1,args=(q,)) # 将队列传参到子进程中
a_list.append(p)
p.start()
for i in a_list:
i.join()
print(q.empty()) # 打印队列是否是空,True
print(q)
内存的第3种方法:管道(不推荐使用,了解即可)
方法:
#创建管道的类:
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异常。
应用:
import multiprocessing
def consumer(p,name):
left,right=p
left.close() # 关闭左边的管道,右边只负责接受
while True:
try:
baozi=right.recv() # 右边接受数据
print('%s 收到包子:%s' %(name,baozi))
except EOFError:
right.close() # 异常则关闭连接
break
def producer(seq,p):
left,right=p
right.close() # 关闭右边管道,左边只负责发送数据
for i in seq:
left.send(i)
# time.sleep(1)
else:
left.close()
if __name__ == '__main__':
pass
left,right = multiprocessing.Pipe()
c1 = multiprocessing.Process(target=consumer,args=((left,right),'c1'))
c1.start()
seq = (i for i in range(10))
producer(seq,(left,right))
right.close() # 关闭管道
left.close() # 关闭管道
c1.join()
print("主进程 over")
c1 收到包子:0
c1 收到包子:1
c1 收到包子:2
c1 收到包子:3
c1 收到包子:4
c1 收到包子:5
c1 收到包子:6
c1 收到包子:7
c1 收到包子:8
c1 收到包子:9
主进程 over
**注意:生产者和消费者都没有使用管道的某个端点,就应该将其关闭,如在生产者中关闭管道的右端,在消费者中关闭管道的左端。如果忘记执行这些步骤,程序可能再消费者中的recv()操作上挂起。管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生产EOFError异常。因此在生产者中关闭管道不会有任何效果,付费消费者中也关闭了相同的管道端点。
管道可以用于双向通信,利用通常在客户端/服务器中使用的请求/响应模型或远程过程调用,就可以使用管道编写与进程交互的程序,如下:
from multiprocessing import Process,Pipe
import time,os
def adder(p,name):
server,client=p
client.close()
while True:
try:
x,y=server.recv()
except EOFError:
server.close()
break
res=x+y
server.send(res)
print('server done')
if __name__ == '__main__':
server,client=Pipe()
c1=Process(target=adder,args=((server,client),'c1'))
c1.start()
server.close()
client.send((10,20))
print(client.recv())
client.close()
c1.join()
print('主进程')
#注意:send()和recv()方法使用pickle模块对对象进行序列化