内容:
进程部分:- 进程间通信=>IPC
- 生产者消费者模型()
线程部分: 1. 线程理论()
2.开启线程的两种方式()
3. 线程对象其他相关的属性或方法
4. 守护线程 5. 线程互斥锁进程部分:
1. 进程间通信=>IPC 基于之前互斥锁的学习，可以实现进程间通过共享硬盘的数据并使用互斥锁的方法处理数据的基础上实现，但是此方法存在问题。 1.需要建立很多的锁 2.在使用完毕后，如果锁没有及时的release，就会造成锁等待，形成阻塞 为了能实现进程间通信，所以寻找一个共享的内存空间，两种方法： 1.管道 一种专门的数据结构，处理数据，但是没有解决进行间锁的问题 2.队列=管道+锁 （推荐优先使用） 队列集成了管道和互斥锁的方法，内部实现。 使用 multiprocessing模块的Queue方法，创建一个进程间的共享内存，其中的Queue()，括号中可以设置队列的数量，如果不写的话，默认是没限制，但是基于 操作系统内存的限制，不能超过内存的容量；如果写的话，不能放置超过队列的数量，超过了就形成锁等待，形成阻塞。 使用 put放置数据 get取出数据 阻塞实例： from multiprocessing import Queue q=Queue(3)#创建一个共享内存 q.put(['first',]) q.put({'x':2}) q.put(3) q.put([1,2])#第四个数据无法放置到队列里，因为已经满了，这里就形成了阻塞 正常实例： from multiprocessing import Queue q=Queue(3)#创建一个共享内存 q.put(['first',]) q.put({'x':2}) q.put(3) q.put([1,2]) print(q.get()) print(q.get()) print(q.get()) # 输出结果： # ['first'] # {'x': 2} # 3 q.put队列中除了有存放的数据的属性，还有block=True,阻塞timeout=？超时等待属性，两个属性必须是联合使用 q.put（存放的数据，block=True，timeout=？） 当队列阻塞了，之后，设置超时的等待时间后，如果等待的时间内没有恢复，就结束进程，不会无限制的等待下去 q.put_nowait(放置的数据)，超过了队列的数量后直接终止，集合了block和timeout。 put端放置block和timeout q.put案例 from multiprocessing import Queue q=Queue(3)#创建一个共享内存 q.put(['first',],block=True,timeout=3) q.put({'x':2},block=True,timeout=3) q.put(3,block=True,timeout=3) q.put([1,2],block=True,timeout=3) print(q.get()) print(q.get()) print(q.get()) 输出结果： 报错之后程序终止 raise Full queue.Full q.put_nowait案例 from multiprocessing import Queue q=Queue(3)#创建一个共享内存 q.put_nowait(1) q.put_nowait(2) q.put_nowait(3) q.put_nowait(4) 输出结果： raise Full queue.Full get端放置block和timeout q.get(block=True,timeout=3)案例 from multiprocessing import Queue q = Queue(3) q.put(['first', ]) q.put({'x': 2}) q.put(3) q.get(block=True,timeout=3) q.get(block=True,timeout=3) q.get(block=True,timeout=3) q.get(block=True,timeout=3) 输出结果： raise Empty queue.Empty q.get_nowait()案例 from multiprocessing import Queue q = Queue(3) q.put(['first', ]) q.put({'x': 2}) q.put(3) q.get_nowait() q.get_nowait() q.get_nowait() q.get_nowait() 输出结果： raise Empty queue.Empty2. 生产者消费者模型()
- 什么是生产者消费者模型
生产者:代指生产数据的任务
消费者:代指处理数据的任务 该模型的工作方式: 生产生产数据传递消费者处理 实现方式: 生产者---->队列<------消费者，其中就使用到了进程间相互通信的队列来实现。为何要用
当程序中出现明细的两类任务,一类负责生产数据,一类负责处理数据
就可以引入生产者消费者模型来实现生产者与消费者的“解耦合”,平衡生产能力与消费能力,从提升效率如何用
案例1：单个生产者和消费者，数据处理，
import time, random from multiprocessing import Process, Queue def producer(name, food, q): for i in range(10): res = '%s%s' % (food, i) time.sleep(random.randint(1, 8)) # 模拟生产数据的时间 q.put(res) print('厨师[%s]生产了----------<%s>' % (name, res)) def consumer(name, q): while True: res = q.get() time.sleep(random.randint(1, 2)) # 模拟处理数据的时间 print('吃货[%s]吃了%s' % (name, res)) if name == 'main': # 创建队列 q = Queue() # 生产者们 p1 = Process(target=producer, args=('Egon', '泔水', q)) # 消费者们 c1 = Process(target=consumer, args=('刘清正', q)) p1.start() c1.start() print('主') 输出结果： 主 厨师[Egon]生产了----------<泔水0> 吃货[刘清正]吃了泔水0 厨师[Egon]生产了----------<泔水1> 吃货[刘清正]吃了泔水1 厨师[Egon]生产了----------<泔水2> 吃货[刘清正]吃了泔水2 厨师[Egon]生产了----------<泔水3> 吃货[刘清正]吃了泔水3 厨师[Egon]生产了----------<泔水4> 吃货[刘清正]吃了泔水4 厨师[Egon]生产了----------<泔水5> 厨师[Egon]生产了----------<泔水6> 吃货[刘清正]吃了泔水5 吃货[刘清正]吃了泔水6 厨师[Egon]生产了----------<泔水7> 吃货[刘清正]吃了泔水7 厨师[Egon]生产了----------<泔水8> 吃货[刘清正]吃了泔水8 厨师[Egon]生产了----------<泔水9> 吃货[刘清正]吃了泔水9 案例2：多个生产者和多个消费者，数据处理分析，生产者给消费者发送生产完成的信号，当数据为None时，数据生产完成，告诉消费者 import time,random from multiprocessing import Process,Queue 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('厨师[%s]生产了----------<%s>' %(name,res)) def consumer(name,q): while True: res=q.get() if res is None:break time.sleep(random.randint(1,3)) #模拟处理数据的时间 print('吃货[%s]吃了<%s>' %(name,res)) if name == 'main': q=Queue() # 生产者们 p1=Process(target=producer,args=('小Egon','泔水',q)) p2=Process(target=producer,args=('中Egon','屎包子',q)) p3=Process(target=producer,args=('大Egon','腰子汤',q)) # 消费者们 c1=Process(target=consumer,args=('刘清正',q)) c2=Process(target=consumer,args=('吴三江',q)) #启动子进程 p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() #放置两个None，结束两个消费者 q.put(None) q.put(None) print('主') 输出结果： 厨师[中Egon]生产了----------<屎包子0> 厨师[小Egon]生产了----------<泔水0> 厨师[中Egon]生产了----------<屎包子1> 吃货[刘清正]吃了<屎包子0> 厨师[中Egon]生产了----------<屎包子2> 厨师[大Egon]生产了----------<腰子汤0> 吃货[吴三江]吃了<泔水0> 厨师[小Egon]生产了----------<泔水1> 厨师[大Egon]生产了----------<腰子汤1> 厨师[小Egon]生产了----------<泔水2> 吃货[刘清正]吃了<屎包子1> 厨师[大Egon]生产了----------<腰子汤2> 主 吃货[吴三江]吃了<屎包子2> 吃货[刘清正]吃了<腰子汤0> 吃货[吴三江]吃了<泔水1> 吃货[吴三江]吃了<泔水2> 吃货[刘清正]吃了<腰子汤1> 吃货[吴三江]吃了<腰子汤2> 案例3，方案2升级版本，使用守护进程，消费者给生产者发送消费完成的信号。 import time,random from multiprocessing import Process,JoinableQueue 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('厨师[%s]生产了----------<%s>' %(name,res)) def consumer(name,q): while True: res=q.get() time.sleep(random.randint(1,3)) #模拟处理数据的时间 print('吃货[%s]吃了<%s>' %(name,res)) q.task_done() if name == 'main': q=JoinableQueue() # 生产者们 p1=Process(target=producer,args=('小Egon','泔水',q)) p2=Process(target=producer,args=('中Egon','屎包子',q)) p3=Process(target=producer,args=('大Egon','腰子汤',q)) # 消费者们 c1=Process(target=consumer,args=('刘清正',q)) c2=Process(target=consumer,args=('吴三江',q)) # 守护进程，主进程结束后，消费者也结束 c1.daemon=True c2.daemon=True c2.daemon=True p1.start() p2.start() p3.start() c1.start() c2.start() p1.join() p2.join() p3.join() q.join() # 主进程等q结束,即q内数据被取干净了 print('主')#当生产者生产完成后，主进程结束后消费者没有正常死亡，消费者没有存在的必要，要一起死掉， 使用守护进程， c1.daemon=True，c2.daemon=True 输出结果： 厨师[小Egon]生产了----------<泔水0> 吃货[刘清正]吃了<泔水0> 厨师[中Egon]生产了----------<屎包子0> 厨师[小Egon]生产了----------<泔水1> 吃货[吴三江]吃了<屎包子0> 厨师[大Egon]生产了----------<腰子汤0> 吃货[刘清正]吃了<泔水1> 厨师[小Egon]生产了----------<泔水2> 厨师[中Egon]生产了----------<屎包子1> 厨师[中Egon]生产了----------<屎包子2> 厨师[大Egon]生产了----------<腰子汤1> 吃货[吴三江]吃了<腰子汤0> 吃货[刘清正]吃了<泔水2> 吃货[刘清正]吃了<屎包子2> 吃货[吴三江]吃了<屎包子1> 厨师[大Egon]生产了----------<腰子汤2> 吃货[刘清正]吃了<腰子汤1> 吃货[吴三江]吃了<腰子汤2> 主线程部分:
1. 线程理论(*****) 1 什么是线程 进程其实一个资源单位,而进程内的线程才是cpu上的执行单位 线程其实指的就是代码的执行过程 2 为何要用线程 线程vs进程 1. 同一进程下的多个线程共享该进程内的资源 2. 创建线程的开销要远远小于进程 3 如何用线程 开启线程的两种方式(*****) 方式1 from threading import Thread import time def task(name): print('%s is running' %name) time.sleep(2) print('%s is done' %name) if __name__ == '__main__': t=Thread(target=task,args=('线程1',)) t.start() print('主') 方式2: from threading import Thread import time class Mythread(Thread): def run(self): print('%s is running' %self.name) time.sleep(2) print('%s is done' %self.name) if __name__ == '__main__': t=Mythread() t.start() print('主') 2. 线程对象其他相关的属性或方法 统一进程下的线程共享同一资源 from threading import Thread import time n=100 def task(): global n n=0 if __name__ == '__main__': t=Thread(target=task) t.start() t.join() print('主',n) 输出结果 主 0 线程的pid from threading import Thread import time,os def task(): print('%s is running' %os.getpid()) if __name__ == '__main__': t=Thread(target=task) t.start() print('主',os.getpid()) 输出结果 16448 is running 主 16448 线程个数，没加join，此例是两个线程 from threading import Thread,active_count,current_thread import time,os def task(): print('%s is running' %current_thread().name) time.sleep(2) if __name__ == '__main__': t=Thread(target=task,) t.start() print('主',active_count()) 输出结果 Thread - 1 is running 主 2 加了join后，线程join后执行完成后就关闭了，只存活主线程所以此例就有1个线程还在。 from threading import Thread,active_count,current_thread import time,os def task(): print('%s is running' %current_thread().name) time.sleep(2) if __name__ == '__main__': t=Thread(target=task,) t.start() t.join() print('主',active_count()) 输出结果 Thread - 1 is running 主 1 当前线程名 from threading import Thread,active_count,current_thread import time,os def task(): print('%s is running' %current_thread().name) time.sleep(2) if __name__ == '__main__': t=Thread(target=task,) t.start() 输出结果 Thread - 1 is running 主 MainThread print('主',current_thread().name) 3. 守护线程 所有的线程都执行完成后,主线程才结束.跟进程的守护进程不一样的是,只要主进程结束, 守护进程无论执行到哪一步,也会立即终止,不再执行. from threading import Thread 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__': t1=Thread(target=foo) t2=Thread(target=bar) 守护线程 t1.daemon=True t1.start() t2.start() print("main-------") 输出结果 123 456 main------- end123 end456 4. 线程互斥锁 线程的进程数据共享,加互斥锁的目的就是让数据更加安全,让大家对数据的处理有个先后,避免对数据处理的混乱 from threading import Thread,Lock import time mutex=Lock()#线程锁 n=100 def task(): global n mutex.acquire() temp=n time.sleep(0.1) n=temp-1 mutex.release() if __name__ == '__main__': t_l=[] for i in range(100): t=Thread(target=task) t_l.append(t) t.start() for t in t_l: t.join() print(n) 输出结果 0 posted on 2018-12-27 09:11 阅读( ...) 评论( ...)