复习
上节课复习： 1、开启进程的两种方式： #方式一： from multiprocessing import Process def task(x): pass def func(): pass if __name__ == "__main__": p=Process(target=task,args=(1,)) p.start() # 仅仅只是向操作系统发送一个开启进程的信号 func() #方式二： from multiprocessing import Process class Process: def start(self): self.run() #p.run() class Myprocess(Process): def run(self): pass if __name__ == "__main__": p=Myprocess() p.start() # 仅仅只是向操作系统发送一个开启进程的信号 func() 2、进程之间内存空间彼此隔离 3、p.join():让p的父进程等待p执行完毕，并且回收子进程（僵尸进程）占用的pid 4、pid：进程id号，即进程在操作系统内的身份证号 ppid:当前进程父进程的id号 今日内容： 进程： 1、守护进程 2、进程的互斥锁 3、进程间通信：IPC PIPE:管道 Queue：队列 4、生产者消费者模型（******） 线程： 1、开启线程的两种方式（*****） 2、线程vs进程（*****）
补充
from multiprocessing import Process import time import os def task(): print('%s is running' %os.getpid()) time.sleep(3) print('%s is done' % os.getpid()) if __name__ == "__main__": p=Process(target=task) p.start() # 仅仅只是向操作系统发送一个开启进程的信号 print(p.pid) p.join() # 等到子进程p执行完毕后，将p占用的操作系统的pid回收 print(p.pid) #这时候可以看到pid但是这时候的pid已经没有意义了，操作系统已经回收了，可以被其他进程利用 print('主')
守护进程
1、守护进程 守护进程其实就是一个“子进程” 守护=》伴随 守护进程会伴随主进程的代码运行完毕后而死掉 2、为何用守护进程 关键字就两个： 进程： 当父进程需要将一个任务并发出去执行，需要将该任务放到一个子进程里 守护： 当该子进程内的代码在父进程代码运行完毕后就没有存在的意义了，就应该 将该子进程设置为守护进程，会在父进程代码结束后死掉 from multiprocessing import Process import time def foo(): print(123) time.sleep(1) print("end123") def bar(): print(456) time.sleep(3) print("end456") if __name__ == '__main__': p1=Process(target=foo) p2=Process(target=bar) p1.daemon=True # 一定要放到p.start()之前 p1.start() p2.start() print("main-------") #这里的输出主要看机器性能 一般来说是第一种输出 ''' main------- 456 enn456 ''' ''' main------- 123 456 enn456 ''' ''' 123 main------- 456 end456 '''
互斥锁
# 进程间需要互相通信 购票 看票并发，购票串行 # 保证共享数据安全的核心是把并发变成串行 #互斥锁：可以将要执行任务的部分代码（只涉及到修改共享数据的代码）变成串行 #join：是要执行任务的所有代码整体串行 from multiprocessing import Process,Lock import json import os import time import random def check(): time.sleep(1) # 模拟网路延迟 with open('db.txt','rt',encoding='utf-8') as f: dic=json.load(f) print('%s 查看到剩余票数 [%s]' %(os.getpid(),dic['count'])) def get(): with open('db.txt','rt',encoding='utf-8') as f: dic=json.load(f) time.sleep(2) if dic['count'] > 0: # 有票 dic['count']-=1 time.sleep(random.randint(1,3)) with open('db.txt','wt',encoding='utf-8') as f: json.dump(dic,f) print('%s 购票成功' %os.getpid()) else: print('%s 没有余票' %os.getpid()) def task(mutex): # 查票 check() #购票 mutex.acquire() # 互斥锁不能连续的acquire，必须是release以后才能重新acquire get() mutex.release() # with mutex: # get() if __name__ == '__main__': mutex=Lock() for i in range(10): p=Process(target=task,args=(mutex,)) p.start() # p.join()
IPC机制
#IPC：进程间通信，有两种实现方式 #1、pipe： #2、queue：pipe+锁 *** #注意： #1、队列占用的是内存空间 #2、不应该往队列中放大数据，应该只存放数据量较小的消息 #一 # from multiprocessing import Queue # q=Queue(3) #先进先出 # 掌握的 block=True,timeout=无时间限制 只在block为true才有意义 # q.put('first') # q.put({'k':'sencond'}) # q.put(['third',]) # q.put(4) #阻塞住了 直到有人取走 才能继续放 # print(q.get()) # print(q.get()) # print(q.get()) # print(q.get()) #阻塞住了 直到有人放才会取到 # #二 #了解的 # q=Queue(3) #先进先出 # q.put('first',block=True,timeout=3) #block timeout 对前三次无效 满了之后才起作用 # q.put({'k':'sencond'},block=True,timeout=3) # q.put(['third',],block=True,timeout=3) # print('===>') # # q.put(4,block=True,timeout=3) #block=True 满了就不让继续放了 阻塞 timeout=3 超过3秒还没有人取走队列里的东西就会报错 # # # print(q.get(block=True,timeout=3)) #取完阻塞，直到有人继续放入，timeout只在block=true才有意义 # print(q.get(block=True,timeout=3)) # print(q.get(block=True,timeout=3)) # print(q.get(block=True,timeout=3)) #三 # q=Queue(3) #先进先出 # q.put('first',block=False,) # q.put({'k':'sencond'},block=False,) # q.put(['third',],block=False,) # print('===>') # # q.put(4,block=False,) # 队列满了直接抛出异常，不会阻塞 # # print(q.get(block=False)) # print(q.get(block=False)) # print(q.get(block=False)) # print('get over') # print(q.get(block=False))#取完队列，直接抛出异常，不会阻塞 # #四 # q=Queue(3) #先进先出 # # q.put_nowait('first') #q.put('first',block=False,) # q.put_nowait(2) # q.put_nowait(3) # q.put_nowait(4) # print(q.get_nowait()) # print(q.get_nowait()) # print(q.get_nowait()) # print(q.get_nowait())
生产者消费者模型 ******
''' 1 什么是生产者消费者模型 生产者:比喻的是程序中负责产生数据的任务 消费者:比喻的是程序中负责处理数据的任务 生产者->共享的介质(队列)<-消费者 2 为何用 实现了生产者与消费者的解耦和,生产者可以不停地生产,消费者也可以不停地消费 从而平衡了生产者的生产能力与消费者消费能力,提升了程序整体运行的效率 什么时候用? 当我们的程序中存在明显的两类任务,一类负责产生数据,另外一类负责处理数据 此时就应该考虑使用生产者消费者模型来提升程序的效率 ''' # from multiprocessing import Queue,Process # import time # import os # import random # # def producer(q): # for i in range(10): # res='包子%s' %i # time.sleep(random.randint(1,3)) # # 往队列里丢 # q.put(res) # print(' 33[45m%s 生产了 %s[0m' %(os.getpid(),res)) # q.put(None) #最后放标志位 # # def consumer(q): # while True: # #从队列里取走 # res=q.get() # if res is None:break #当队列里取值为none表明取值完毕 # time.sleep(random.randint(1,3)) # print(' 33[46m%s 吃了 %s[0m' %(os.getpid(),res)) # # if __name__ == '__main__': # q=Queue() # # 生产者们 # p1=Process(target=producer,args=(q,)) # # 消费者们 # c1=Process(target=consumer,args=(q,)) # # p1.start() # c1.start() # # print('主') # from multiprocessing import Queue,Process # import time # import os # import random # # def producer(name,food,q): # for i in range(3): # res='%s%s' %(food,i) # time.sleep(random.randint(1,3)) # # 往队列里丢 # q.put(res) # print(' 33[45m%s 生产了 %s 33[0m' %(name,res)) # # q.put(None) # # def consumer(name,q): # while True: # #从队列里取走 # res=q.get() # if res is None:break # time.sleep(random.randint(1,3)) # print(' 33[46m%s 吃了 %s 33[0m' %(name,res)) # # if __name__ == '__main__': # q=Queue() # # 生产者们 # p1=Process(target=producer,args=('egon','包子',q,)) # p2=Process(target=producer,args=('杨军','泔水',q,)) # p3=Process(target=producer,args=('猴老师','翔',q,)) # # 消费者们 # c1=Process(target=consumer,args=('Alex',q,)) # c2=Process(target=consumer,args=('wupeiqidsb',q,)) # # p1.start() # p2.start() # p3.start() # c1.start() # c2.start() # # p1.join() # p2.join() # p3.join() # # 在p1p2p3都结束后,才应该往队列里放结束信号,有几个消费者就应该放几个None # q.put(None) # q.put(None) # print('主') #解决生产者和消费者都结束后 程序不结束问题 终极版本 from multiprocessing import JoinableQueue,Process import time import os import random def producer(name,food,q): for i in range(3): res='%s%s' %(food,i) time.sleep(random.randint(1,3)) # 往队列里丢 q.put(res) print(' 33[45m%s 生产了 %s 33[0m' %(name,res)) # q.put(None) def consumer(name,q): while True: #从队列里取走 res=q.get() if res is None:break time.sleep(random.randint(1,3)) print(' 33[46m%s 吃了 %s 33[0m' %(name,res)) q.task_done() #告诉队列有一个值被取走 if __name__ == '__main__': q=JoinableQueue() #等队列结束 队列被取空 就算结束 取一个减一次 # 生产者们 p1=Process(target=producer,args=('egon','包子',q,)) p2=Process(target=producer,args=('杨军','泔水',q,)) p3=Process(target=producer,args=('猴老师','翔',q,)) # 消费者们 c1=Process(target=consumer,args=('Alex',q,)) c2=Process(target=consumer,args=('wupeiqidsb',q,)) c1.daemon=True #主进程的代码运行完毕--->(生产者运行完毕)+队列中的数据也被取干净了->消费者没有存在的意义 c2.daemon=True #所以子进程做成守护进程随主进程一起结束 p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() #等生产者不在生产 才执行下一行代码 q.join() #等待队列被取干净 统计队列的值 一个taskdone 就减一次 知道为0 执行下一行代码 #为保证生产者生产完了 所以在上面p1.join() # q.join() 结束意味着 # 主进程的代码运行完毕--->(生产者运行完毕)+队列中的数据也被取干净了->消费者没有存在的意义 # print('主')