一、进程间通信
进程与进程之间是数据隔离的
管道/队列(管道+锁)
队列:先进先出
堆栈:先进后出
q.put() 放入值
q.get() 获取队列里面的值(同一时刻只能有一个任务来队列中获取数据)
两者在存放值和取值的时候都会出现阻塞的情况(队列满了,队列空了)
q = Queue(5) 括号内可以传参数表示的是这个队列的最大存储数
q.full() 判断队列是否满了
q.empty() 判断队列中的数据是否取完
q.get_nowait() 取值,没有值不等待直接报错
ps:full、get_nowait、empty都不适用于多进程的情况
""" 队列:先进先出 堆栈:先进后出 """ from multiprocessing import Queue q = Queue(5) # 括号内可以传参数 表示的是这个队列的最大存储数 # 往队列中添加数据 q.put(1) q.put(2) # print(q.full()) # 判断队列是否满了 q.put(3) q.put(4) q.put(5) # print(q.full()) # q.put(6) # 当队列满了之后 再放入数据 不会报错 会原地等待 直到队列中有数据被取走(阻塞态) print(q.get()) print(q.get()) print(q.get()) print(q.empty()) # 判断队列中的数据是否取完 print(q.get()) print(q.get()) print(q.empty()) # print(q.get_nowait()) # 取值 没有值不等待直接报错 # print(q.get()) # 当队列中的数据被取完之后 再次获取 程序会阻塞 直到有人往队列中放入值
二、进程间通信IPC机制
子进程释放数据,主进程获取数据
两个子进程相互放、取数据
from multiprocessing import Process,Queue def producer(q): q.put('hello GF~') def consumer(q): print(q.get()) if __name__ == '__main__': q = Queue() p = Process(target=producer,args=(q,)) c = Process(target=consumer, args=(q,)) p.start() c.start()
三、生产者消费者模型
生产者:生产/制造数据(做包子)
消费者:消费/处理数据(吃包子)
两者之间的通信介质:队列/管道
供需不平衡:
1.做包子的远比买包子的多
2.做包子的远比买包子的少
q.task_done() 告诉队列你已经从队列中取出了一个数据,并且处理完毕了
from multiprocessing import Process,Queue,JoinableQueue import random import time def producer(name,food,q): for i in range(10): data = '%s生产了%s%s'%(name,food,i) time.sleep(random.random()) q.put(data) print(data) def consumer(name,q): while True: data = q.get() if data == None:break print('%s吃了%s'%(name,data)) time.sleep(random.random()) q.task_done() # 告诉队列你已经从队列中取出了一个数据 并且处理完毕了 if __name__ == '__main__': q = JoinableQueue() p = Process(target=producer,args=('大厨egon','馒头',q)) p1 = Process(target=producer,args=('跟班tank','生蚝',q)) c = Process(target=consumer,args=('许兆龙',q)) c1 = Process(target=consumer,args=('吃货jerry',q)) p.start() p1.start() c.daemon = True c1.daemon = True c.start() c1.start() p.join() p1.join() q.join() # 等到队列中数据全部取出 # q.put(None) # q.put(None)
四、线程理论
1.什么是线程
进程线程其实都是虚拟单位,都是用来帮助我们形象的描述某种事物
进程:资源单位
线程:执行单位
将内存比如成工厂
那么进程就相当于是工厂里面的车间
而你的线程就相当于是车间里面的流水线
ps:每个进程都自带一个线程:线程才是真正的执行单位,进程只是在线程运行过程中
提供代码运行所需要的资源
2.为什么要有线程
开进程
1.申请内存空间 耗资源
2.拷贝代码 耗资源
开线程
一个进程内可以起多个线程,并且线程与线程之间数据是共享的
ps:开启线程的开销要远远小于开启进程的开销
五、创建线程的两种方式
方式一:
from threading import Thread import time def task(name): print('%s is running'%name) time.sleep(3) print('%s is over'%name) # 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内 t = Thread(target=task,args=('egon',)) t.start() # 告诉操作系统开辟一个线程 线程的开销远远小于进程 # 小的代码执行完 线程就已经开启了 print('主')
方式二
from threading import Thread import time class MyThread(Thread): def __init__(self,name): super().__init__() self.name = name def run(self): print('%s is running'%self.name) time.sleep(3) print('%s is over'%self.name) t = MyThread('egon') t.start() print('主')
六、线程对象及其他方法
开线程不需要在__main__代码块内,但是习惯性的还是写在__main__代码块内
active_count()表示当前正在运行的线程数量
from threading import Thread,current_thread,active_count import time import os def task(name,i): print('%s is running'%name) # print('子current_thread:',current_thread().name) # print('子',os.getpid()) time.sleep(i) print('%s is over'%name) # 开线程不需要在__main__代码块内 但是习惯性的还是写在__main__代码块内 t = Thread(target=task,args=('egon',1)) t1 = Thread(target=task,args=('jason',2)) t.start() # 告诉操作系统开辟一个线程 线程的开销远远小于进程 t1.start() # 告诉操作系统开辟一个线程 线程的开销远远小于进程 t1.join() # 主线程等待子线程运行完毕 print('当前正在活跃的线程数',active_count()) # 小的代码执行完 线程就已经开启了 print('主') # print('主current_thread:',current_thread().name) # print('主',os.getpid())
七、守护线程
主线程的结束也就意味着进程的结束
主线程必须等待其他非守护线程的结束才能结束
意味子线程在运行的时候需要使用进程中的资源,而主线程一旦结束了资源也就销毁了
from threading import Thread,current_thread import time def task(i): print(current_thread().name) time.sleep(i) print('GG') # for i in range(3): # t = Thread(target=task,args=(i,)) # t.start() t = Thread(target=task,args=(1,)) t.daemon = True t.start() print('主')
八、线程间通信
from threading import Thread money = 666 def task(): global money money = 999 t = Thread(target=task) t.start() t.join() print(money)
九、互斥锁
当多个线程操作同一份数据,会出现数据不安全的情况下
设计到多个线程或进程操作同一份数据的时候,通常都需要并行并发变成串行
虽然牺牲了效率但是提高了数据的安全性
针对不同的数据,需要加不同的锁
锁(独立卫生间)
from threading import Thread,Lock import time n = 100 def task(mutex): global n mutex.acquire() tmp = n time.sleep(0.1) n = tmp - 1 mutex.release() t_list = [] mutex = Lock() for i in range(100): t = Thread(target=task,args=(mutex,)) t.start() t_list.append(t) for t in t_list: t.join() print(n) # 结果为0