生产者消费者模型
import time import random from multiprocessing import Process,Queue def consumer(q): while True: obj = q.get() print('消费了一个数据%s'%obj) time.sleep(random.randint(1,3)) if __name__ == '__main__': q =Queue() p = Process(target=consumer,args=(q,)) p.daemon = True p.start() for i in range(10): time.sleep(random.randint(1,5)) q.put('food%s'%i) print('生产了一个数据food%s'%i)
import time import random from multiprocessing import Process, Queue def consumer(q): while True: obj = q.get() if obj == None:break print('消费了一个数据%s' % obj) time.sleep(random.randint(1, 3)) def producer(q): for i in range(10): time.sleep(random.randint(1, 5)) q.put('food%s' % i) print('生产了一个数据food%s' % i) q.put(None) if __name__ == '__main__': q = Queue() for i in range(2): p_consumer = Process(target=consumer, args=(q,)) p_consumer.start() for i in range(2): p_producer = Process(target=producer, args=(q,)) p_producer.start()
import time import random from multiprocessing import Process, Queue def consumer(q): while True: obj = q.get() if obj == None:break print('消费了一个数据%s' % obj) time.sleep(random.randint(1, 3)) def producer(q): for i in range(10): time.sleep(random.randint(1, 5)) q.put('food%s' % i) print('生产了一个数据food%s' % i) if __name__ == '__main__': q = Queue() p1_consumer = Process(target=consumer, args=(q,)) p2_consumer = Process(target=consumer, args=(q,)) p1_consumer.start() p2_consumer.start() p1_producer = Process(target=producer, args=(q,)) p2_producer = Process(target=producer, args=(q,)) p1_producer.start() p2_producer.start() p1_producer.join() q.put(None) p2_producer.join() q.put(None)
每个进程的生产者在生产完后向队列中put一个None,消费者get时判断,收到None则退出
multiprocessing.JoinableQueue()
创建可连接的共享进程队列。这就像是一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。
q = JoinableQueue()
q.join()
q.task_done
from multiprocessing import Process,JoinableQueue import time,random,os def consumer(q): while True: res=q.get() time.sleep(random.randint(1,3)) print(' 33[45m%s 吃 %s 33[0m' %(os.getpid(),res)) q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了 def producer(name,q): for i in range(10): time.sleep(random.randint(1,3)) res='%s%s' %(name,i) q.put(res) print(' 33[44m%s 生产了 %s 33[0m' %(os.getpid(),res)) q.join() #生产完毕，使用此方法进行阻塞，直到队列中所有项目均被处理。 if __name__ == '__main__': q=JoinableQueue() #生产者们:即厨师们 p1=Process(target=producer,args=('包子',q)) p2=Process(target=producer,args=('骨头',q)) p3=Process(target=producer,args=('泔水',q)) #消费者们:即吃货们 c1=Process(target=consumer,args=(q,)) c2=Process(target=consumer,args=(q,)) c1.daemon=True c2.daemon=True #开始 p_l=[p1,p2,p3,c1,c2] for p in p_l: p.start() p1.join() p2.join() p3.join() print('主') #主进程等--->p1,p2,p3等---->c1,c2 #p1,p2,p3结束了,证明c1,c2肯定全都收完了p1,p2,p3发到队列的数据 #因而c1,c2也没有存在的价值了,不需要继续阻塞在进程中影响主进程了。应该随着主进程的结束而结束,所以设置成守护进程就可以了。
multiprocessing.Pipe管道
#创建管道的类：
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异常。
介绍
(pickle实现) 可以传递更多类型数据
from multiprocessing import Pipe #(pickle实现) lp,rp = Pipe() lp.send([1,2,3,]) print(rp.recv()) rp.send([1,2,3,]) print(lp.recv())
from multiprocessing import Pipe,Process def consumer(lp,rp): lp.close() while True: try: print(rp.recv()) except EOFError:break if __name__ == '__main__': lp, rp = Pipe() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() rp.close() for i in range(100): lp.send('food%s'%i) lp.close()
应该特别注意管道端点的正确管理问题。如果是生产者或消费者中都没有使用管道的某个端点，就应将它关闭。这也说明了为何在生产者中关闭了管道的输出端，在消费者中关闭管道的输入端。如果忘记执行这些步骤，程序可能在消费者中的recv（）操作上挂起。管道是由操作系统进行引用计数的，必须在所有进程中关闭管道后才能生成EOFError异常。因此，在生产者中关闭管道不会有任何效果，除非消费者也关闭了相同的管道端点。
from multiprocessing import Pipe,Process # lp,rp = Pipe() # lp.send([1,2,3]) # print(rp.recv()) # rp.send('aa') # print(lp.recv()) def consumer(lp,rp): # lp.close() #不写close将不会引发EOFError,消费者一直阻塞 while True: # try: print(rp.recv()) # except EOFError:break if __name__ == '__main__': lp, rp = Pipe() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() Process(target=consumer,args=(lp,rp)).start() rp.close() for i in range(500): lp.send('food%s'%i) lp.close()
管道的EOFError是怎么报出来的
管道空了,且一端关闭了
管道是由操作系统进行引用计数的，必须在所有进程中关闭管道后才能生成EOFError异常。
管道在数据管理上是不安全的,没有锁机制
队列的实现机制 就是 管道+锁
multiprocessing.Manager 进程之间的数据共享
with Manager() as m: == m = Manager()
dic = m.dict({'count':100})
#创建100个进程对dic进行减1操作,得到的结果不准确,默认没有锁,需要加锁
from multiprocessing import Manager,Process,Lock def work(d,lock): # with lock: #不加锁而操作共享的数据,肯定会出现数据错乱 d['count'] -= 1 if __name__ == '__main__': lock=Lock() with Manager() as m: # m = Manager() #Manager(list,dict)数据不安全的需要加锁 dic=m.dict({'count':100}) p_l=[] for i in range(100): p=Process(target=work,args=(dic,lock)) p_l.append(p) p.start() for p in p_l: p.join() print(dic)
展望未来，基于消息传递的并发编程是大势所趋
即便是使用线程，推荐做法也是将程序设计为大量独立的线程集合，通过消息队列交换数据。
这样极大地减少了对使用锁定和其他同步手段的需求，还可以扩展到分布式系统中。
但进程间应该尽量避免通信，即便需要通信，也应该选择进程安全的工具来避免加锁带来的问题。
以后我们会尝试使用数据库来解决现在进程之间的数据共享问题。
multiprocessing.Pool进程池
进程池
cpu数量有限,同一时间启动进程数量有限,多进程适合处理高计算型程序,阻塞性不适合用多进程,会一直占用cpu
进程的个数和cpu数量相关,一般情况设置 进程的个数 = cpu个数或cpu个数加1 ,1切换时间,2进程数量有限,多一个被调用到几率大(不是绝对的)
- 对于纯计算型的代码,使用进程池更好
- 对于高IO的代码,直接多进程更好(不科学,相对的)
- 进程池比起多进程来说,节省了开启进程回收进程资源的时间,给操作系统调度降低了难度
-
- p = Pool(5) 创建一个进程池对象
- p.apply(func=xxx,args=(123,)) 同步提交任务 ,没有多进程的优势,还增加了开启和关闭进程时间
- p.apply_async(func=xxx,args=(123,),callback=call) 异步提交任务,归还进程后立即派发 (常用) 创建时可以拿到返回值,和close join结合使用的
- 回调函数 callback 程序执行结束后return内容,返回给callback函数,使用的是主进程的资源
- 图片
-
- p.close() 关闭进程池(阻止向进程池中添加新的任务)
- p.join() join依赖close,一个进程池必须先close 再join
- p.map(func=wahaha,iterable=(range(101))) 接收一个任务函数,和一个iterable,节省了for循环 close(),join,是一种简便写法. 拿不到返回值
- p.get():返回结果，如果有必要则等待结果到达。timeout是可选的。如果在指定时间内还没有到达，将引发一场。如果远程操作中引发了异常，它将在调用此方法时再次被引发。
- p.ready():如果调用完成，返回True
- p.successful():如果调用完成且没有引发异常，返回True，如果在结果就绪之前调用此方法，引发异常
- p.wait([timeout]):等待结果变为可用
- p.terminate()：立即终止所有工作进程，同时不执行任何清理或结束任何挂起工作。如果p被垃圾回收，将自动调用此函数
在程序实际处理问题过程中，忙时会有成千上万的任务需要被执行，闲时可能只有零星任务。那么在成千上万个任务需要被执行的时候，我们就需要去创建成千上万个进程么？首先，创建进程需要消耗时间，销毁进程也需要消耗时间。第二即便开启了成千上万的进程，操作系统也不能让他们同时执行，这样反而会影响程序的效率。因此我们不能无限制的根据任务开启或者结束进程。那么我们要怎么做呢？ 在这里，要给大家介绍一个进程池的概念，定义一个池子，在里面放上固定数量的进程，有需求来了，就拿一个池中的进程来处理任务，等到处理完毕，进程并不关闭，而是将进程再放回进程池中继续等待任务。如果有很多任务需要执行，池中的进程数量不够，任务就要等待之前的进程执行任务完毕归来，拿到空闲进程才能继续执行。也就是说，池中进程的数量是固定的，那么同一时间最多有固定数量的进程在运行。这样不会增加操作系统的调度难度，还节省了开闭进程的时间，也一定程度上能够实现并发效果。
import time import random from multiprocessing import Pool,Process def wahaha(i): print(i ** i) time.sleep(random.randint(1,5)) return i ** i*'-' if __name__ == '__main__': start = time.time() p = Pool(5) for i in range(101): p.apply_async(func=wahaha,args=(i,)) # 异步提交了一个任务 p.close() p.join() print(time.time() - start) start = time.time() p_lst = [Process(target=wahaha, args=(i,)) for i in range(101)] for p in p_lst:p.start() for p in p_lst:p.join() print( time.time()- start) p = Pool(5) for i in range(101): p.apply_async(func=wahaha,args=(i,)) # 异步提交了一个任务 p.close() p.join() p = Pool(5) for i in range(101): p.apply(func=wahaha,args=(i,)) # 同步提交了一个任务 p.close() p.join() p = Pool() p.map(func=wahaha,iterable=range(101)) p = Pool(5) result_lst = [] for i in range(101): r = p.apply_async(func=wahaha,args=(i,)) # 异步提交了一个任务 result_lst.append(r) for r in result_lst:print(r.get()) p.close() p.join()
进程池和多进程效率对比
import time import random from multiprocessing import Process,Pool def wahaha(i): print(i*i) # time.sleep(random.randint(1,5)) return i * i * '-' def ret(argv): print(argv) if __name__ == '__main__': start = time.time() #apply_async 异步 p = Pool(5) for i in range(101): # p.apply_async(func=wahaha,args=(i,),callback=ret) #向池中添加新任务 p.apply_async(func=wahaha,args=(i,)) #向池中添加新任务 p.close() #停止往池中添加新任务 p.join() # join依赖close,一个进程池必须先close(完成提交任务)再join print(time.time() - start) start = time.time() #多进程执行 p_lst = [Process(target=wahaha, args=(i,)) for i in range(101)] for p in p_lst:p.start() for p in p_lst:p.join() print( time.time()- start) ======================= 0.6436324119567871 # 8.800149202346802 #
import time import random from multiprocessing import Process,Pool def wahaha(i): print(i*i) time.sleep(random.randint(1,5)) return i * i * '-' def ret(argv): print(argv) if __name__ == '__main__': start = time.time() #apply_async 异步 p = Pool(5) for i in range(101): # p.apply_async(func=wahaha,args=(i,),callback=ret) #向池中添加新任务 p.apply_async(func=wahaha,args=(i,)) #向池中添加新任务 p.close() #停止往池中添加新任务 p.join() # join依赖close,一个进程池必须先close(完成提交任务)再join print(time.time() - start) start = time.time() #多进程执行 p_lst = [Process(target=wahaha, args=(i,)) for i in range(101)] for p in p_lst:p.start() for p in p_lst:p.join() print( time.time()- start) ================================= 59.929256200790405 17.863743543624878
进程池比起多进程来说 节省了开启进程回收进程资源的时间,给操作系统调度进程降低了难度