一，前言
-
进程：是程序，资源集合，进程控制块组成，是最小的资源单位
- 特点：就对Python而言，可以实现真正的并行效果
- 缺点：进程切换很容易消耗cpu资源，进程之间的通信相对线程来说比较麻烦
-
线程：是进程中最小的执行单位。
- 特点无法利用多核，无法实现真正意义上是并行效果。
- 优点：对于IO密集型的操作可以很好利用IO阻塞的时间
二，多进程
2.1 multiprocessing模块介绍
在上一节多线程中讲到，由于GIL的原因，多线程无法利用多核优势，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。Python提供了非常好用的多进程包multiprocessing。multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务（比如函数），该模块与多线程模块threading的编程接口类似。multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。
2.2 process类介绍
类的实例化（创建进程）
Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）
强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号
参数介绍
group参数未使用，值始终为None
target表示调用对象，即子进程要执行的任务
args表示调用对象的位置参数元组，args=(1,2,‘hexin‘,)
kwargs表示调用对象的字典,kwargs={‘name‘:‘hexin‘,‘age‘:18}
name为子进程的名称
# process方法
p.start()：启动进程，并调用该子进程中的p.run()
p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法
p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
p.is_alive():如果p仍然运行，返回True
p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程
# process 属性
p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置 p.name:进程的名称 p.pid：进程的pid p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可) p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）
2.3 多进程创建
创建形式一：普通形式
import time
import random
from multiprocessing import Process
def fun(name):
print(‘%s begin‘ %name)
time.sleep(random.randrange(1, 3))
print(‘%s end‘ % name)
if __name__ == ‘__main__‘:
p1 = Process(target=fun, args=(‘w‘,))
p2 = Process(target=fun,args=(‘a‘,))
p3 = Process(target=fun,args=(‘l‘,))
p4 = Process(target=fun,args=(‘l‘,))
p1.start()
p2.start()
p3.start()
p4.start()
print(‘主线程‘)
输出结果：
主线程 w begin a begin l begin l begin a end l end w end l end
创建方式二：继承方式
import time
import random
from multiprocessing import Process
class Sleep(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
print(‘%s sleep begin‘ % self.name)
time.sleep(random.randrange(1,5))
print(‘%s end‘ % self.name)
if __name__ == ‘__main__‘:
for i in [‘a‘, ‘b‘, ‘c‘]:
Sleep(i).start()
print("main")
输出：
main b sleep begin a sleep begin c sleep begin c end b end a end
2.4 进程同步
进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的。
共享同一打印终端，发现会有多行内容打印到一行的现象（多个进程共享并抢占同一个打印终端，乱了）
既然可以用文件共享数据，那么进程间通信用文件作为数据传输介质就可以了啊，可以，但是有问题：1.效率 2.需要自己加锁处理
加锁的目的是为了保证多个进程修改同一块数据时，同一时间只能有一个修改，即串行的修改，没错，速度是慢了，牺牲了速度而保证了数据安全。
文件当做数据库，模拟抢票（Lock互斥锁）
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#文件db的内容为：{"count":2}
#注意一定要用双引号，不然json无法识别
from multiprocessing import Process,Lock
import json
import time
import random
import os
def work(filename,lock): #买票
# lock.acquire()
with lock:
with open(filename,encoding=‘utf-8‘) as f:
dic=json.loads(f.read())
# print(‘剩余票数: %s‘ % dic[‘count‘])
if dic[‘count‘] > 0:
dic[‘count‘]-=1
time.sleep(random.randint(1,3)) #模拟网络延迟
with open(filename,‘w‘,encoding=‘utf-8‘) as f:
f.write(json.dumps(dic))
print(‘%s 购票成功‘ %os.getpid())
else:
print(‘%s 购票失败‘ %os.getpid())
# lock.release()
if __name__ == ‘__main__‘:
lock=Lock()
p_l=[]
for i in range(10):
p=Process(target=work,args=(‘db‘,lock))
p_l.append(p)
p.start()
for p in p_l:
p.join()
print(‘主线程‘)
输出：
购票成功 购票成功 购票失败 购票失败 购票失败 购票失败 购票失败 购票失败 购票失败 购票失败 主线程
三，进程间通信
进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的。
3.1 进程间通信（IPC）方式一：队列（推荐使用）
队列方式一：Queue()
队列先进先出，栈后进先出，创建队列的类（底层就是以管道和锁定的方式实现）：
# 实例创建 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()一样
实例：基于队列实现生产者和消费者模型
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
time.sleep(random.randint(1,3))
res=q.get()
if res is None:break
print(‘\033[45m消费者拿到了：%s\033[0m‘ %res)
def producer(seq,q):
for item in seq:
time.sleep(random.randint(1,3))
print(‘\033[46m生产者生产了：%s\033[0m‘ %item)
q.put(item)
if __name__ == ‘__main__‘:
q=Queue()
c=Process(target=consumer,args=(q,))
c.start()
producer((‘包子%s‘ %i for i in range(5)),q)
q.put(None)
c.join()
print(‘主线程‘)
输出：
生产者生产了：包子0 消费者拿到了：包子0 生产者生产了：包子1 消费者拿到了：包子1 生产者生产了：包子2 消费者拿到了：包子2 生产者生产了：包子3 消费者拿到了：包子3 生产者生产了：包子4 消费者拿到了：包子4 主线程
队列方式二：JoinableQueue()
JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
maxsize是队列中允许最大项数，省略则无大小限制。
JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：
q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常
q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止
生产者消费者模型：
from multiprocessing import Process,JoinableQueue
import time,random
def consumer(q):
while True:
# time.sleep(random.randint(1,2))
res=q.get()
print(‘消费者拿到了 %s‘ %res)
q.task_done()
def producer(seq,q):
for item in seq:
# time.sleep(random.randrange(1,2))
q.put(item)
print(‘生产者做好了 %s‘ %item)
q.join()
if __name__ == ‘__main__‘:
q=JoinableQueue()
seq=(‘包子%s‘ %i for i in range(5))
p=Process(target=consumer,args=(q,))
p.daemon=True #设置为守护进程，在主线程停止时p也停止，但是不用担心，producer内调用q.join保证了consumer已经处理完队列中的所有元素
p.start()
producer(seq,q)
print(‘主线程‘)
3.2 进程间通信（IPC）方式二：管道
# 创建实例 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
import time,os
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__‘:
left,right=Pipe()
c1=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(‘主进程‘)
注意：生产者和消费者都没有使用管道的某个端点，就应该将其关闭，如在生产者中关闭管道的右端，在消费者中关闭管道的左端。如果忘记执行这些步骤，程序可能再消费者中的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(‘主进程‘)
输出：
30 server done 主进程
四，进程池
4.1 进程池介绍
开多进程的目的是为了并发，如果有多核，通常有几个核就开几个进程，进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行），但很明显需要并发执行的任务要远大于核数，这时我们就可以通过维护一个进程池来控制进程数目，比如httpd的进程模式，规定最小进程数和最大进程数。
当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个目标，手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。而且对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。
4.2 进程池使用
4.2.1进程池方式一：
同步调用：
from multiprocessing import Pool
import os,time
def work(n):
print(‘{} run‘.format(os.getpid()))
time.sleep(1)
return n ** 2 # ret
if __name__ == ‘__main__‘:
p = Pool(3) # 创建3个进程
res_1 = []
for i in range(20):
res = p.apply(work,args=(i,))
‘‘‘同步调用，直到本次任务执行完毕拿到res，等待任务work执行的过程中可能有阻塞也可能没有阻塞，
但不管该任务是否存在阻塞，同步调用都会在原地等着，只是等的过程中若是任务发生了阻塞就会被夺走cpu的执行权限;
个人理解：程序判断两个子程序执行的间隔时间，过长则判断存在阻塞，届时夺走上次进程ID的使用权限，从进程池分配新的进程ID‘‘‘
res_1.append(res)
print(res_1)
异步调用：
from multiprocessing import Pool
import os,time
def work(n):
print(‘%s run‘ %os.getpid())
time.sleep(3)
return n**2
if __name__ == ‘__main__‘:
p=Pool(3) #进程池中从无到有创建三个进程,以后一直是这三个进程在执行任务
res_l=[]
for i in range(10):
print(i) # for循环会提前运行完毕，进程池内的任务还未执行。
res=p.apply_async(work,args=(i,)) #同步运行,阻塞、直到本次任务执行完毕拿到res
res_l.append(res) #将调用apply_async方法，得到返回进程内存地址结果
#异步apply_async用法：如果使用异步提交的任务，主进程需要使用join，等待进程池内任务都处理完，然后可以用get收集结果，
# 否则，主进程结束，进程池可能还没来得及执行，也就跟着一起结束了
p.close()
p.join()
for res in res_l:
print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get
apply_sync和apply方法
# apply_async方法
from multiprocessing import Process,Pool
import time
def func(msg):
print( "msg:", msg)
time.sleep(1)
return msg
if __name__ == "__main__":
pool = Pool(processes = 3)
res_l=[]
for i in range(10):
print(i)
msg = "hello %d" %(i)
res=pool.apply_async(func, (msg, )) #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
res_l.append(res) # 将apply_async方法得到的内存地址结果加入列表
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方法
from multiprocessing import Process,Pool
import time
def func(msg):
print( "msg:", msg)
time.sleep(0.1)
return msg
if __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)
map方法
#map
import os,time
from multiprocessing import Pool
def func(i):
time.sleep(1)
print(‘子进程{}‘.format(os.getpid()))
return i
if __name__ == ‘__main__‘:
p = Pool(5)
ret = p.map(func,range(10)) # func(next(range(10)))
print(ret)
4.2.1 concurrent.futures 模块实现‘池’
同步调用：
# #同步调用：提交/调用一个任务，然后就在原地等着，等到该任务执行完毕拿到结果，再执行下一行代码
from concurrent.futures import ProcessPoolExecutor
import time
def save_test(name,n):
n = n+1
time.sleep(n)
print("{}的名字次数为{}".format(name,n))
return n
if __name__ == ‘__main__‘:
start = time.time()
ex = ProcessPoolExecutor(max_workers=3)
lista = ["Tom","Jerry","XiaoHua","Ming"]
for i,j in enumerate(lista):
#存在两个以上的参数时，直接用逗号隔开，不需要用括号
task = ex.submit(save_test,j,i).result()
print(task)
#ex.shutdown(wait=True)是进程池内部的进程都执行完毕，才会关闭，然后执行后续代码
ex.shutdown(wait=True)
print("主进程直接运行")
stop = time.time()
print(stop-start)
异步调用：
#异步调用: 提交/调用一个任务，不在原地等着，直接执行下一行代码
from concurrent.futures import ProcessPoolExecutor
import time
def save_test(name,n):
n = n+1
time.sleep(n)
print("{}的名字次数为{}".format(name,n))
return n
if __name__ == ‘__main__‘:
obj = list()
start = time.time()
ex = ProcessPoolExecutor(max_workers=3)
lista = ["letme","Mlxg","XiaoHu","Ming"]
for i,j in enumerate(lista):
#存在两个以上的参数时，直接用逗号隔开，不需要用括号
task = ex.submit(save_test,j,i)
obj.append(task)
#ex.shutdown(wait=True)是进程池内部的进程都执行完毕，才会关闭，然后执行后续代码
ex.shutdown(wait=True)
print("主进程直接运行")
for i in obj:
print(i.result())
stop = time.time()
print(stop-start)
map方法使用
from concurrent.futures import ProcessPoolExecutor
import urllib.request
URLS = [‘http://www.163.com‘, ‘https://www.baidu.com/‘, ‘https://github.com/‘]
def load_url(url):
with urllib.request.urlopen(url, timeout=60) as conn:
print(‘%r page is %d bytes‘ % (url, len(conn.read())))
if __name__ == ‘__main__‘:
executor = ProcessPoolExecutor(max_workers=3)
executor.map(load_url,URLS)
print(‘主进程‘)
五，总结
用futures的写法上更简洁一些，concurrent.futures的性能并没有更好，只是让编码变得更简单。考虑并发编程的时候，任何简化都是好事。从长远来看，concurrent.futures编写的代码更容易维护。
使用map时，future是逐个迭代提交，multiprocessing.Pool是批量提交jobs，因此对于大批量jobs的处理，multiprocessing.Pool效率会更高一些。对于需要长时间运行的作业，用future更佳，future提供了更多的功能（callback, check status, cancel）。
concurrent.futures.ProcessPoolExecutor是对multiprocessing的封装，在运行时需导入__main__，不能直接在交互窗口工作。
原文地址：https://www.cnblogs.com/tashanzhishi/p/10775657.html
时间： 2024-10-09 06:08:51