管道
进程间通信(IPC)方式二:管道(不推荐使用,了解即可),端口易导致数据不安全的情况出现。
from multiprocessing import Pipe,Process
def func(conn1,conn2):
msg = conn1.recv() # 接收了conn2传递的
# msg1 = conn2.recv() # 接收了conn1传递的
print('>>>',msg)
# print('>>>',msg1)
if __name__ == '__main__':
# 拿到管道的两端,双工通信方式,两端都可以收发消息
conn1,conn2 = Pipe() # 必须在Process之前产生管道
p = Process(target=func,args=(conn1,conn2,)) # 管道给子进程
p.start()
conn1.send('hello')
conn1.close()
conn2.send('小子')
conn2.close()
print('进程结束')
# 注意管道不用了就关闭,防止异常
数据共享
进程之间数据共享模块–Manager,使用较少,进程间的数据是独立的,可以借助队列或者管道实现通信,二者都是基于消息传递的,我们可以通过Manager模块实现数据的共享。
from multiprocessing import Manager,Process,Lock
def func1(dic,loc):
# loc.acquire() # 不加锁易出错
dic['num'] -= 1
# loc.release()
if __name__ == '__main__':
m = Manager()
loc = Lock()
dic = m.dict({'num':100})
p_list = []
for i in range(100):
p = Process(target=func1, args=(dic,loc))
p_list.append(p)
p.start()
[pp.join() for pp in p_list]
print('>>>>>',dic['num'])
# 共享时不加锁,很可能导致同一个数据被多个子进程取用,数据是不安全的,且超多进程消耗大量资源易导致卡死.
多进程共同去处理共享数据的时候,就和我们多进程同时去操作一个文件中的数据是一样的,不加锁就会出现错误的结果,进程不安全的,所以也需要加锁。
总结:进程间应该尽量避免通信,即便需要通信,也应该选择进程安全的工具来避免加锁带来的问题。
进程池 Pool
节省内存空间,以及节省创建进程的时间。
创建进程需要消耗时间,销毁进程(空间,变量,文件信息等等的内容)也需要消耗时间。开启成千上万的进程,操作系统无法让他们同时执行,维护一个很大的进程列表的同时,调度的时候,还需要进行切换并且记录每个进程的执行节点,也就是记录上下文(各种变量等等乱七八糟的东西,虽然你看不到,但是操作系统都要做),这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程,这就需要用到进程池。
Pool([numprocess [,initializer [, initargs]]]):创建进程池
参数介绍:
- numprocess:要创建的进程数,如果省略,将默认使用cpu_count()的值
- initializer:是每个工作进程启动时要执行的可调用对象,默认为None
- 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()之后调用
特点:
- 一般创建进程池的个数为cpu个数加一。
- map(func,iterable),异步的,自带close和join,返回值是一个list
- apply 同步的,apply(func,args=()) 只有当func只想完之后,才回乡下只想其他代码,返回值是func的return
from multiprocessing import Process,Pool
import time
def func1(i):
num = 0
for j in range(3):
num += i
time.sleep(1)
print(num)
return num
if __name__ == '__main__':
pool = Pool(6)
for i in range(10):
res = pool.apply(func1,args=(i,)) # apply 进程同步/串行方法 效率低,不常用
# print(res)
- apply_async(func,args=()) 异步的,只管提交任务,当func被注册如一个进程之后,程序就继续向下执行。返回值是一个apply_async返回的对象。结束的时候需要先close再join来保持多进程和主进程代码的同步性。
from multiprocessing import Process,Pool
import time
def func1(i):
num = 0
for j in range(5):
num += i
time.sleep(1)
# print('>>>>>',num)
return num
if __name__ == '__main__':
pool = Pool(6)
red_list = []
for i in range(10):
res = pool.apply_async(func1,args=(i,))
red_list.append(res)
pool.close() # 不是关闭,只是锁定进程池,告诉主进程不会再添加数据进去
pool.join() # 等待子程序执行完
for ress in red_list:
print(ress.get()) # get方法取出返回值num 按添加顺序取出已保存在缓存区的结果 所以是顺序打印出的
回调函数
回调函数是在主进程中执行的,一般会用在爬虫中。
from multiprocessing import Pool
def func1(n):
return n+1
def func2(m):
print(m)
if __name__ == '__main__':
p = Pool(5)
for i in range(10,20):
p.apply_async(func1,args=(i,),callback=func2)
p.close()
p.join()
爬虫:
import requests
from multiprocessing import Pool
def get(url):
response = requests.get(url)
if response.status_code == 200:
return url,response.content.decode('utf-8')
def call_back(args):
url,content = args
print(url,len(content))
if __name__ == '__main__':
url_list = [
'http://www.cnblogs.com/',
'http://www.baidu.com',
'http://www.sogou.com',
'http://www.sohu.com',
]
p = Pool(5)
for url in url_list:
p.apply_async(get,args=(url,),callback=call_back) #运用时注意一点,回调函数的形参执行有一个,如果你的执行函数有多个返回值,那么也可以被回调函数的这一个形参接收,接收的是一个元祖,包含着你执行函数的所有返回值。
p.close()
p.join()