一、进程:
进程:程序运行在操作系统上的一个实例,就称之为进程。进程需要相应的系统资源:内存、时间片、pid。 创建进程: 首先要导入multiprocessing中的Process: 创建一个Process对象; 创建Process对象时,可以传递参数;
例1:不使用p.join()
import os
from multiprocessing import Process
import time
def pro_func(name):
for i in range(5):
print(f'子进程{name+str(i)}开始执行')
time.sleep(2)
print(f'子进程{name+str(i)}结束执行')
if __name__ == '__main__':
p=Process(target=pro_func,args=('kobe',))
p.start()
print('主进程执行结束')
执行结果:
主进程执行结束
子进程kobe0开始执行
子进程kobe0结束执行
子进程kobe1开始执行
子进程kobe1结束执行
子进程kobe2开始执行
子进程kobe2结束执行
子进程kobe3开始执行
子进程kobe3结束执行
子进程kobe4开始执行
子进程kobe4结束执行
例2:使用p.join():阻塞函数,等待所有的子进程执行完毕后,主进程结束执行
import os
from multiprocessing import Process
import time
def pro_func(name):
for i in range(5):
print(f'子进程{name+str(i)}开始执行')
time.sleep(2)
print(f'子进程{name+str(i)}结束执行')
if __name__ == '__main__':
p=Process(target=pro_func,args=('kobe',))
p.start()
p.join()
print('主进程执行结束')
执行结果:
子进程kobe0开始执行
子进程kobe0结束执行
子进程kobe1开始执行
子进程kobe1结束执行
子进程kobe2开始执行
子进程kobe2结束执行
子进程kobe3开始执行
子进程kobe3结束执行
子进程kobe4开始执行
子进程kobe4结束执行
主进程执行结束
例3:p.terminate():杀死子进程
import os
from multiprocessing import Process
import time
def pro_func(name):
for i in range(5):
print(f'子进程{name+str(i)}开始执行')
time.sleep(2)
print(f'子进程{name+str(i)}结束执行')
if __name__ == '__main__':
p=Process(target=pro_func,args=('kobe',))
p.start()
p.terminate()
p.join()
print('主进程执行结束')
执行结果:
主进程执行结束
二、进程之间的通信-Queue
在初始化Queue()对象时(例如q=Queue(),若在括号中没有指定最大可接受的消息数量,获数量为负值时,那么就代表可接受的消息数量没有上限一直到内存尽头)
Queue.qsize():返回当前队列包含的消息数量
Queue.empty():如果队列为空,返回True,反之False
Queue.full():如果队列满了,返回True,反之False
Queue.get([block[,timeout]]):获取队列中的一条消息,然后将其从队列中移除,block默认
值为True。
如果block使用默认值,且没有设置timeout(单位秒),消息队列如果为空,此时程序将被阻塞(停在读中状态),直到消息队列读到消息为止,如果设置了timeout,则会等待timeout秒,若还没读取到任何消息,则抛出“Queue.Empty"异常:
Queue.put(item,[block[,timeout]]):将item消息写入队列,block默认值为True; 如果block使用默认值,且没有设置timeout(单位秒),消息队列如果已经没有空间可写入,此时程序将被阻塞(停在写入状态),直到从消息队列腾出空间为止,如果设置了timeout,则会等待timeout秒,若还没空间,则抛出”Queue.Full"异常 如果block值为False,消息队列如果没有空间可写入,则会立刻抛出"Queue.Full"异常; Queue.put_nowait(item):相当Queue.put(item,False)
例4:
from multiprocessing import Process,Queue
import os,time,random
def write(q):
for i in range(6):
print(f'生产产品{i}')
q.put(f'生产产品{i},添加到队列中')
time.sleep(2)
def read(q):
while True:
if not q.empty():
value=q.get(True)
print(f'获取到了产品{value}')
time.sleep(2)
else:
break
if __name__ == '__main__':
q=Queue()
pw=Process(target=write,args=(q,))
pr=Process(target=read,args=(q,))
pw.start()
pr.start()
pw.join()
pr.join()
print('所有数据都写入并且读完')
执行结果:
生产产品0
获取到了产品生产产品0,添加到队列中
生产产品1
获取到了产品生产产品1,添加到队列中
生产产品2
获取到了产品生产产品2,添加到队列中
生产产品3
获取到了产品生产产品3,添加到队列中
生产产品4
获取到了产品生产产品4,添加到队列中
生产产品5
获取到了产品生产产品5,添加到队列中
所有数据都写入并且读完
三、进程池中使用Queue
如果要使用Pool创建进程,就需要使用multiprocessing.Manager()中的Queue(),而不是multiprocessing.Queue(),否则会得到如下的错误信息:
RuntimeError: Queue objects should only be shared between processs through inheritance
例5
from multiprocessing import Manager,Pool
import os,time,random
def write(q):
for i in range(6):
q.put(f'产品{i}')
print(f'产品{i}生产完成')
time.sleep(1)
def read(q):
while True:
if not q.empty():
value=q.get(True)
print(f'获取到了{value}')
time.sleep(2)
if __name__ == '__main__':
print(f'主进程开始执行,进程号为{os.getpid()}')
q=Manager().Queue()
po=Pool(3)
po.apply_async(func=write,args=(q,))
po.apply_async(func=read,args=(q,))
po.close()
po.join()
print(f'主进程结束执行,进程号为{os.getpid()}')
执行结果