Python multiprocess模块(下)

#创建管道的类:
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异常.

from multiprocessing import Process, Pipe # 引入Pipe模块def func(conn):
conn.send("HelloWorld!") # 子进程发送了消息
conn.close() # 子进程关闭通道的一端if __name__ == '__main__':
parent_conn, child_conn = Pipe() # 建立管道,拿到管道的两端,双工通信方式,两端都可以收发消息
p = Process(target=func, args=(child_conn,)) # 将管道的一端给子进程
p.start() # 开启子进程
print("主进程接收>>>", parent_conn.recv()) # 主进程接收了消息
p.join()
print("主进程执行结束!")

from multiprocessing import Process, Pipe # 引入Pipe模块def func(conn):
msg = conn.recv() # (5)子进程通过管道的另一端接收信息
print("The massage from parent_process is>>>", msg)if __name__ == '__main__':
parent_conn, child_conn = Pipe() # (1)创建管道,拿到管道的两端
p = Process(target=func, args=(child_conn,)) # (2)创建子进程func, 把child_conn给func
p.start() # (3)启动子进程
parent_conn.send("Hello,child_process!") # (4)主进程通过parent_conn给子进程发送信息# 执行结果:
# The massage from parent_process is>>> Hello,child_process!

from multiprocessing import Process, Pipedef func(parent_conn, child_conn):
msg = parent_conn.recv() # (5)子进程使用parent_conn接收主进程的消息
print("子进程使用parent_conn接收>>>", msg) # (6)打印接收到的消息
child_conn.send("子进程使用child_conn给主进程发送了一条消息") # (7)子进程发送消息
print("子进程执行完毕")if __name__ == '__main__':
parent_conn, child_conn = Pipe() # (1)创建管道,拿到管道两端
child_conn.send("主进程使用child_conn给子进程发送了一条消息") # (2)主进程发消息
p = Process(target=func, args=(parent_conn, child_conn)) # (3)创建子进程,把管道两端都给子进程
p.start() # (4)开启子进程
p.join() # (8)等待子进程执行完毕
msg = parent_conn.recv() # (9)主进程使用parent_conn接收子进程的消息
print("主进程使用parent_conn接收>>>", msg) # (10)打印接收到的消息
print("主进程执行完毕!")# 执行结果:
# 子进程使用parent_conn接收>>> 主进程使用child_conn给子进程发送了一条消息
# 子进程执行完毕
# 主进程使用parent_conn接收>>> 子进程使用child_conn给主进程发送了一条消息
# 主进程执行完毕!

from multiprocessing import Process, Pipedef f(parent_conn,child_conn):
#parent_conn.close() #不写close将不会引发EOFError
while True:
try:
print(child_conn.recv())
except EOFError:
child_conn.close()
breakif __name__ == '__main__':
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(parent_conn,child_conn,))
p.start()
child_conn.close()
parent_conn.send('hello')
parent_conn.close()
p.join()

from multiprocessing import Process, Manager # 引入Manager模块def func(m_list):
m_list.pop() # 对manager列表进行改变if __name__ == '__main__':
m = Manager() # 创建Manager对象
m_list = m.list(["王力宏", "王乃卉", "王少轩"]) # 创建manager列表
print("主进程>>>", m_list)
p = Process(target=func, args=(m_list,)) # 创建子进程
p.start()
p.join()
print("主进程>>>", m_list)# 执行结果:
# 主进程>>> ['王力宏', '王乃卉', '王少轩']
# 主进程>>> ['王力宏', '王乃卉']

from multiprocessing import Process, Manager # 引入Manager模块def func(m_dic):
m_dic["name"] = "王力宏" # 修改manager字典if __name__ == '__main__':
m = Manager() # 创建Manager对象
m_dic = m.dict({"name": "王乃卉"}) # 创建manager字典
print("主进程>>>", m_dic)
p = Process(target=func, args=(m_dic,)) # 创建子进程
p.start()
p.join()
print("主进程>>>", m_dic)# 执行结果:
# 主进程>>> {'name': '王乃卉'}
# 主进程>>> {'name': '王力宏'}

from multiprocessing import Process, Managerdef func(m_dic):
m_dic["count"] -= 1if __name__ == '__main__':
m = Manager()
m_dic = m.dict({"count": 100})
p_list = []
# 开启20个进程来对共享数据进行修改
for i in range(20):
p = Process(target=func, args=(m_dic, ))
p.start()
p_list.append(p)
[p.join() for p in p_list]
print("主进程>>>", m_dic)# 执行结果:
# 主进程>>> {'count': 80}
# 但是偶尔会出现 主进程>>> {'count': 81} 的情况, 这是因为共享数据不变, 但是当多个子进程同时访问共享数据并对其进行修改时, 由于修改的过程是要重写对共享数据进行赋值的, 在这个赋值的过程中, 可能一个子进程还没来得及赋值成功, 就有另外的一个子进程拿到原先的值, 这样一来, 就会出现多个子进程修改同一个共享数据, 于是就出现了上面代码结果偶尔会少减了一次的现象. 综上所述,共享数据是不够安全的, 而"加锁"是一个很好的解决办法.

from multiprocessing import Process, Manager, Lockdef func(m_dic, m_lock):
with m_lock:
m_dic["count"] -= 1
# 等同于:
# m_lock.acquire()
# m_dic["count"] -= 1
# m_lock.release()if __name__ == '__main__':
m = Manager()
m_lock = Lock()
m_dic = m.dict({"count": 100})
p_list = []
# 开启20个进程来对共享数据进行修改
for i in range(20):
p = Process(target=func, args=(m_dic, m_lock))
p.start()
p_list.append(p)
[p.join() for p in p_list]
print("主进程", m_dic)# 执行结果:
# 主进程 {'count': 80}# 加锁后, 多次尝试运行程序, 执行结果也没有发生改变. 不难看出, 加锁后 共享数据是安全的.

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.join(): 等待所有工作进程退出. 此方法只能在close()或teminate()之后调用.

方法apply_async()和map_async()的返回值是AsyncResul的实例obj. 实例具有以下方法:obj.get(): 返回结果, 如果有必要则等待结果到达. timeout是可选的. 如果在指定时间内还没有到达, 将引发异常. 如果远程操作中引发了异常, 它将在调用此方法时再次被引发.
obj.ready(): 如果调用完成, 返回True
obj.successful(): 如果调用完成且没有引发异常, 返回True, 如果在结果就绪之前调用此方法, 引发异常
obj.wait([timeout]): 等待结果变为可用.
obj.terminate(): 立即终止所有工作进程, 同时不执行任何清理或结束任何挂起工作. 如果p被垃圾回收, 将自动调用此函数.

import time
from multiprocessing import Pooldef func(n):
time.sleep(0.5)
print(n)if __name__ == '__main__':
pool = Pool(4) # 创建进程池对象,进程池中放置了4个进程,一般来说,这个数量是电脑的CPU数量
pool.map(func, range(100)) #参数必须是可迭代的

import time
from multiprocessing import Process, Pooldef func(n):
for i in range(5):
n = n + iif __name__ == '__main__':
pool_start_time = time.time() # 进程池开始执行时间
pool = Pool(4) #创建进程池对象,进程池中设置了4个进程
pool.map(func, range(100)) # map是异步执行的,
pool_end_time = time.time() # 进程池执行完毕时间
pool_different_time = pool_end_time - pool_start_time # 进程池执行时间差 p_start_time = time.time() # 多进程开始执行时间
p_list = []
for i in range(100):
p1 = Process(target=func, args=(i,))
p1.start()
p_list.append(p1)
[p.join() for p in p_list]
p_end_time = time.time() # 多进程执行完毕时间
p_different_time = p_end_time - p_start_time # 多进程执行时间差 print("进程池的执行时间>>>", pool_different_time)
print("多进程的执行时间>>>", p_different_time)# 执行结果:
# 进程池的执行时间>>> 0.16112160682678223
# 多进程的执行时间>>> 3.6605968475341797# 可以明显地看出,进程池的执行效率远远高于多进程.

import time
from multiprocessing import Pooldef func(i):
time.sleep(0.5)
return i**2if __name__ == '__main__':
p = Pool(4)
for i in range(10):
res = p.apply(func, args=(i,))
"""p.apply() --> 同步执行的方法,它会等待子进程的返回结果,所以最后的执行结果是匀速打印出来的"""
print(res)

import os
import time
import random
from multiprocessing import Pooldef work(n):
print('%s run' % os.getpid()) # 进程ID号
time.sleep(random.random())
return n**2if __name__ == '__main__':
p = Pool(4) # 进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l = []
for i in range(10):
res = p.apply_async(work, args=(i,))
"""异步运行,根据进程池中的进程数,每次最多4个子进程在异步执行,并且可以执行不同的任务,传送任意的参数了.
返回结果之后,将结果放入列表,归还进程,之后再执行新的任务.需要注意的是,进程池中的三个进程不会同时开启或
者同时结束而是执行完一个就释放一个进程,这个进程就去接收新的任务."""
res_l.append(res) """异步apply_async用法:如果使用异步提交的任务,主进程需要使用join,等待进程池内任务都处理完,然后可以用get收集结果.
否则,主进程结束,进程池可能还没来得及执行,也就跟着一起结束了."""
p.close() # 不是关闭进程池,而是结束进程池接收任务,确保没有新任务再提交过来.
p.join() # 感知进程池中的任务已经执行结束,只有当没有新的任务添加进来的时候,才能感知到任务结束了,所以在join之前必须加上close方法.
for res in res_l:
print(res.get()) # 使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get.

#一：使用进程池（异步调用,apply_async）
#coding: utf-8
from multiprocessing import Process,Pool
import timedef func(msg):
print( "msg:", msg)
time.sleep(1)
return msgif __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
msg = "hello %d" %(i)
res=pool.apply_async(func, (msg, )) #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
res_l.append(res)
# s = res.get() #如果直接用res这个结果对象调用get方法获取结果的话，这个程序就变成了同步，因为get方法直接就在这里等着你创建的进程的结果，第一个进程创建了，并且去执行了，那么get就会等着第一个进程的结果，没有结果就一直等着，那么主进程的for循环是无法继续的，所以你会发现变成了同步的效果
print("==============================>") #没有后面的join，或get，则程序整体结束，进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了 pool.close() #关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
pool.join() #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束 print(res_l) #看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果
for i in res_l:
print(i.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get#二：使用进程池（同步调用,apply）
#coding: utf-8
from multiprocessing import Process,Pool
import timedef func(msg):
print( "msg:", msg)
time.sleep(0.1)
return msgif __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
msg = "hello %d" %(i)
res=pool.apply(func, (msg, )) #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
res_l.append(res) #同步执行，即执行完一个拿到结果，再去执行另外一个
print("==============================>")
pool.close()
pool.join() #调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束 print(res_l) #看到的就是最终的结果组成的列表
for i in res_l: #apply是同步的，所以直接得到结果，没有get()方法
print(i)

需要回调函数的场景: 进程池中任何一个任务一旦处理完了, 就立即告知主进程自己已处理完毕了. 主进程则调用一个函数去处理该任务的执行结果, 该函数即回调函数, 这是进程池特有的, 普通进程没有这个机制, 但是我们也可以通过进程通信来拿到返回值, 进程池的这个回调也是进程通信的机制完成的.我们可以把比较消耗时间(阻塞)的任务放到进程池中, 然后指定回调函数(主进程负责执行), 这样主进程在执行回调函数时就省去了I/O的过程, 直接拿到的是任务的结果.

import os
from multiprocessing import Pooldef func1(n):
print('func1>>',os.getpid())
print('func1')
return n*ndef func2(nn):
print('func2>>',os.getpid())
print('func2')
print(nn)
# import time
# time.sleep(0.5)
if __name__ == '__main__':
print('主进程：',os.getpid())
p = Pool(5)
#args里面的10给了func1，func1的返回值作为回调函数的参数给了callback对应的函数，不能直接给回调函数直接传参数，他只能是你任务函数func1的函数的返回值
# for i in range(10,20): #如果是多个进程来执行任务，那么当所有子进程将结果给了回调函数之后，回调函数又是在主进程上执行的，那么就会出现打印结果是同步的效果。我们上面func2里面注销的时间模块打开看看
# p.apply_async(func1,args=(i,),callback=func2)
p.apply_async(func1,args=(10,),callback=func2) p.close()
p.join()#结果
# 主进程： 11852 #发现回调函数是在主进程中完成的，其实如果是在子进程中完成的，那我们直接将代码写在子进程的任务函数func1里面就行了，对不对，这也是为什么称为回调函数的原因。
# func1>> 17332
# func1
# func2>> 11852
# func2
#