Python并发编程之多进程(二)

十、进程同步

进程之间数据不共享,但是共享同一套文件系统,所以访问同一个文件,或同一个打印终端,是没有问题的,

而共享带来的是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理

 --------------------------------------------------------------------注：如果你对python感兴趣，我这有个学习Python基地，里面有很多学习资料，感兴趣的+Q群：895817687--------------------------------------------------------------------# 模拟抢票，购票行为由并行变成了串行，牺牲了效率，提高了数据安全性
from multiprocessing import Process, Lock
import json
import time
import osdef search_ticket():with open("file/ticket", mode="r", encoding="utf-8") as f:ticket_num = int(f.read())print("剩余票数：{0}".format(ticket_num))def get_ticket():with open("file/ticket", mode="r", encoding="utf-8") as f:ticket = int(f.read())time.sleep(0.1)  # 模拟抢票延时if ticket:ticket -= 1print("{0}抢到了一张票，还剩{1}张票".format(os.getpid(), ticket))else:print("{0}没有抢到票".format(os.getpid()))f = open("file/ticket", mode="w", encoding="utf-8")f.write(str(ticket))def task(lock):search_ticket()lock.acquire()get_ticket()lock.release()if __name__ == '__main__':lock = Lock()for i in range(100):p = Process(target=task, args=(lock,))p.start()

总结：加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，没错，速度是慢了，但牺牲了速度却保证了数据安全。

1：虽然可以用文件共享数据实现进程间通信，但问题是：

2：效率低（共享数据基于文件，而文件是硬盘上的数据）
需要自己加锁处理

十一、队列（推荐使用）

进程彼此之间互相隔离，要实现进程间通信（IPC），multiprocessing模块支持两种形式：队列和管道，这两种方式都是使用消息传递的

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方法可以禁止这种行为

from multiprocessing import Queueq = Queue(maxsize=3)q.put(1)
q.put({"name":"dogfa"})
q.put([1,2,3])print(q.full())      # Trueprint(q.get())       # 1
print(q.get())       # {'name': 'dogfa'}
print(q.get())       # [1, 2, 3]print(q.empty()) # True

2：生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

3：为什么使用生产者消费者模式

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

4：什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

5：JoinableQueue()

 #JoinableQueue([maxsize])：这就像是一个Queue对象，但队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。#参数介绍：maxsize是队列中允许最大项数，省略则无大小限制。    #方法介绍：JoinableQueue的实例p除了与Queue对象相同的方法之外还具有：q.task_done()：使用者使用此方法发出信号，表示q.get()的返回项目已经被处理。如果调用此方法的次数大于从队列中删除项目的数量，将引发ValueError异常q.join():生产者调用此方法进行阻塞，直到队列中所有的项目均被处理。阻塞将持续到队列中的每个项目均调用q.task_done（）方法为止

6：生产者消费者模型的实现

from multiprocessing import JoinableQueue, Process
import os
import random
import timedef customer(q):while 1:time.sleep(0.5)print("{0}号顾客吃了{1}".format(os.getpid(), q.get()))q.task_done()def producter(food, q):for i in range(10):time.sleep(random.randint(1, 2))q.put(food)print("{0}号厨师完成了{1}的制作".format(os.getpid(), food))q.join()if __name__ == '__main__':q = JoinableQueue()pro1 = Process(target=producter, args=("包子", q))pro2 = Process(target=producter, args=("油条", q))pro3 = Process(target=producter, args=("花卷", q))cus1 = Process(target=customer, args=(q,))cus2 = Process(target=customer, args=(q,))cus1.daemon = Truecus2.daemon = Truelst = [pro1, pro2, pro3, cus1, cus2][i.start() for i in lst]pro1.join()pro2.join()pro3.join()print("ending...")# 主进程等待pro1，Pro2，pro3执行完成，当pro执行完成意味着cus必定执行完成，所以可以将cus设置成守护进程

7:生产者消费者模式总结

 #程序中有两类角色一类负责生产数据（生产者）一类负责处理数据（消费者）#引入生产者消费者模型为了解决的问题是：平衡生产者与消费者之间的工作能力，从而提高程序整体处理数据的速度#如何实现：生产者<-->队列<——>消费者#生产者消费者模型实现类程序的解耦和

十二、管道（不推荐使用）

1:管道

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

2:利用管道实现进程间的通信

from multiprocessing import Process,Pipeimport time,os
def consumer(p,name):left,right=pleft.close()while True:try:baozi=right.recv()print('%s 收到包子:%s' %(name,baozi))except EOFError:right.close()break
def producer(seq,p):left,right=pright.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('主进程')

十三、数据共享

进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的，虽然进程间数据独立，但可以通过Manager实现数据共享

from multiprocessing import Process, Manager, Lock
import os
import random
import timedef func(dic, lock):lock.acquire()  # 不加锁肯定会造成数据混乱time.sleep(random.randrange(2))dic["count"] -= 1lock.release()if __name__ == '__main__':lock = Lock()with Manager() as m:dic = m.dict({"count": 100})lst = []for i in range(100):p = Process(target=func, args=(dic, lock))lst.append(p)p.start()[i.join() for i in lst]print(dic["count"])

进程间通信应该尽量避免使用上述共享数据的方式

十三、进程池

在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。多进程是实现并发的手段之一，需要注意的问题是：

很明显需要并发执行的任务通常要远大于核数
一个操作系统不可能无限开启进程，通常有几个核就开几个进程
进程开启过多，效率反而会下降（开启进程是需要占用系统资源的，而且开启多余核数目的进程也无法做到并行）

如果当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个。。。手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。我们就可以通过维护一个进程池来控制进程数目。

ps：对于远程过程调用的高级应用程序而言，应该使用进程池，Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，就重用进程池中的进程。

创建进程池的类：如果指定numprocess为3，则进程池会从无到有创建三个进程，然后自始至终使用这三个进程去执行所有任务，不会开启其他进程

1：创建进程池

ool([numprocess [,initializer [, initargs]]]):创建进程池

2:参数介绍

numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
initializer：是每个工作进程启动时要执行的可调用对象，默认为None
initargs：是要传给initializer的参数组

3:方法介绍

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()之后调用方法apply_async()和map_async（）的返回值是AsyncResul的实例obj。实例具有以下方法:obj.get():返回结果，如果有必要则等待结果到达。
obj.ready():如果调用完成，返回True
obj.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
obj.wait([timeout]):等待结果变为可用。
obj.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数

4:进程池的使用
4.1:使用进程池（异步调用，apply_async）

from multiprocessing import Process, Pool
import os
import time
import socketdef func(i):print(i)time.sleep(1)return i ** 2if __name__ == '__main__':pool = Pool(os.cpu_count() + 1)ret_lst = []for i in range(100):# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去ret = pool.apply_async(func, args=(i,))ret_lst.append(ret)# 没有后面的join，或get，则程序整体结束，进程池中的任务还没来得及全部执行完也都跟着主进程一起结束了print("=======================")# 关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成pool.close()# 调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool,join函数等待所有子进程结束pool.join()# 看到的是<multiprocessing.pool.ApplyResult object at 0x10357c4e0>对象组成的列表,而非最终的结果,但这一步是在join后执行的,证明结果已经计算完毕,剩下的事情就是调用每个对象下的get方法去获取结果print(ret_lst)# 使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get[print(i.get()) for i in ret_lst]

4.2:使用进程池（同步调用，apply）

from multiprocessing import Process, Pool
import os
import time
import socketdef func(i):print(i)time.sleep(1)return i ** 2if __name__ == '__main__':# 维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去pool = Pool(os.cpu_count() + 1)# 同步执行，即执行完一个拿到结果，再去执行另外一个ret_lst = []for i in range(100):ret = pool.apply(func, args=(i,))ret_lst.append(ret)# 看到的就是最终的结果组成的列表，apply是同步的，所以直接得到结果，没有get()方法print(ret_lst)

4.3:进程池实现基于TCP协议的socket并发效果

# 服务端
from multiprocessing import Process, Pool
import os
import time
import socketdef func(conn, client_addr):print("进程：{0}".format(os.getpid()))while 1:try:c_msg = conn.recv(1024).decode("utf-8")if not c_msg: breakprint(c_msg)conn.send(c_msg.upper().encode("utf-8"))except Exception:breakif __name__ == '__main__':sk = socket.socket()sk.bind(("127.0.0.1", 8080))sk.listen(5)pool = Pool(os.cpu_count() + 1)while 1:conn, addr = sk.accept()pool.apply_async(func, args=(conn, addr))# 服务端
import socketsk = socket.socket()sk.connect(("127.0.0.1", 8080))while 1:c_msg = input(">>")if not c_msg: continuesk.send(c_msg.encode("utf-8"))s_msg = sk.recv(1024).decode("utf-8")print(s_msg)

当连接数达到开启的进程池中的最大进程数量时，再有其它客户端进行连接，将会阻塞等待，当另外的客户端结束连接时才会建立起会话连接。

4.5:回调函数（callback()）

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

我们可以把耗时间（阻塞）的任务放到进程池中，然后指定回调函数（主进程负责执行），这样主进程在执行回调函数时就省去了I/O的过程，直接拿到的是任务的结果。

from multiprocessing import Pool
import requests
import json
import osdef get_page(url):print('<进程%s> get %s' %(os.getpid(),url))respone=requests.get(url)if respone.status_code == 200:return {'url':url,'text':respone.text}def pasrse_page(res):print('<进程%s> parse %s' %(os.getpid(),res['url']))parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))with open('db.txt','a') as f:f.write(parse_res)if __name__ == '__main__':urls=['https://www.baidu.com','https://www.python.org','https://www.openstack.org','https://help.github.com/','http://www.sina.com.cn/']p=Pool(3)res_l=[]for url in urls:res=p.apply_async(get_page,args=(url,),callback=pasrse_page)res_l.append(res)p.close()p.join()print([res.get() for res in res_l]) # 拿到的是get_page的结果,其实完全没必要拿该结果,该结果已经传给回调函数处理了'''
打印结果:
<进程3388> get https://www.baidu.com
<进程3389> get https://www.python.org
<进程3390> get https://www.openstack.org
<进程3388> get https://help.github.com/
<进程3387> parse https://www.baidu.com
<进程3389> get http://www.sina.com.cn/
<进程3387> parse https://www.python.org
<进程3387> parse https://help.github.com/
<进程3387> parse http://www.sina.com.cn/
<进程3387> parse https://www.openstack.org
[{'url': 'https://www.baidu.com', 'text': '<!DOCTYPE html>\r\n...',...}]
'''

如果在主进程中等待进程池中所有任务都执行完毕后，再统一处理结果，则无需回调函数

十四、信号量

互斥锁同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去，如果指定信号量为3，那么来一个人获得一把锁，计数加1，当计数等于3时，后面的人均需要等待。一旦释放，就有人可以获得一把锁

信号量与进程池的概念很像，但是要区分开，信号量涉及到加锁的概念

from multiprocessing import Process,Semaphore
import time,randomdef go_wc(sem,user):sem.acquire()print('%s 占到一个茅坑' %user)time.sleep(random.randint(0,3)) #模拟每个人拉屎速度不一样，0代表有的人蹲下就起来了sem.release()if __name__ == '__main__':sem=Semaphore(5)p_l=[]for i in range(13):p=Process(target=go_wc,args=(sem,'user%s' %i,))p.start()p_l.append(p)for i in p_l:i.join()print('============》')

十五、事件

Python线程的事件用于主线程控制其他线程的执行，事件主要提供了三个方法 set、wait、clear。

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

clear：将“Flag”设置为False
set：将“Flag”设置为True

from multiprocessing import Process,Event
import time,randomdef car(e,n):while True:if not e.is_set(): #Flaseprint('\033[31m红灯亮\033[0m，car%s等着' %n)e.wait()print('\033[32m车%s 看见绿灯亮了\033[0m' %n)time.sleep(random.randint(3,6))if not e.is_set():continueprint('走你,car', n)breakdef police_car(e,n):while True:if not e.is_set():print('\033[31m红灯亮\033[0m，car%s等着' % n)e.wait(1)print('灯的是%s，警车走了,car %s' %(e.is_set(),n))breakdef traffic_lights(e,inverval):while True:time.sleep(inverval)if e.is_set():e.clear() #e.is_set() ---->Falseelse:e.set()if __name__ == '__main__':e=Event()# for i in range(10):#     p=Process(target=car,args=(e,i,))#     p.start()for i in range(5):p = Process(target=police_car, args=(e, i,))p.start()t=Process(target=traffic_lights,args=(e,10))t.start()print('============》')